/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar
1553.5.10 by Martin Pool
New DirectoryNotEmpty exception, and raise this from local and memory
1
# Copyright (C) 2004, 2005, 2006 by Canonical Ltd
1887.1.1 by Adeodato Simó
Do not separate paragraphs in the copyright statement with blank lines,
2
#
1530.1.3 by Robert Collins
transport implementations now tested consistently.
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
1887.1.1 by Adeodato Simó
Do not separate paragraphs in the copyright statement with blank lines,
7
#
1530.1.3 by Robert Collins
transport implementations now tested consistently.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
# GNU General Public License for more details.
1887.1.1 by Adeodato Simó
Do not separate paragraphs in the copyright statement with blank lines,
12
#
1530.1.3 by Robert Collins
transport implementations now tested consistently.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
17
"""Tests for Transport implementations.
18
19
Transport implementations tested here are supplied by
20
TransportTestProviderAdapter.
21
"""
22
23
import os
24
from cStringIO import StringIO
1530.1.15 by Robert Collins
Move put mode tests into test_transport_implementation.
25
import stat
26
import sys
1530.1.3 by Robert Collins
transport implementations now tested consistently.
27
1755.3.7 by John Arbash Meinel
Clean up and write tests for permissions. Now we use fstat which should be cheap, and lets us check the permissions and the file size
28
from bzrlib import (
29
    osutils,
30
    urlutils,
31
    )
1553.5.10 by Martin Pool
New DirectoryNotEmpty exception, and raise this from local and memory
32
from bzrlib.errors import (DirectoryNotEmpty, NoSuchFile, FileExists,
1185.85.84 by John Arbash Meinel
[merge] bzr.dev 1573, lots of updates
33
                           LockError, PathError,
1185.85.76 by John Arbash Meinel
Adding an InvalidURL so transports can report they expect utf-8 quoted paths. Updated tests
34
                           TransportNotPossible, ConnectionError,
35
                           InvalidURL)
1685.1.9 by John Arbash Meinel
Updated LocalTransport so that it's base is now a URL rather than a local path. This helps consistency with all other functions. To do so, I added local_abspath() which returns the local path, and local_path_to/from_url
36
from bzrlib.osutils import getcwd
1955.3.29 by John Arbash Meinel
Use applyDeprecated instead of callDeprecated
37
from bzrlib.symbol_versioning import zero_eleven
1530.1.9 by Robert Collins
Test bogus urls with http in the new infrastructure.
38
from bzrlib.tests import TestCaseInTempDir, TestSkipped
1871.1.2 by Robert Collins
Reduce code duplication in transport-parameterised tests.
39
from bzrlib.tests.test_transport import TestTransportImplementation
1685.1.45 by John Arbash Meinel
Moved url functions into bzrlib.urlutils
40
from bzrlib.transport import memory
1530.1.3 by Robert Collins
transport implementations now tested consistently.
41
import bzrlib.transport
42
43
44
def _append(fn, txt):
45
    """Append the given text (file-like object) to the supplied filename."""
46
    f = open(fn, 'ab')
1530.1.21 by Robert Collins
Review feedback fixes.
47
    try:
48
        f.write(txt.read())
49
    finally:
50
        f.close()
1530.1.3 by Robert Collins
transport implementations now tested consistently.
51
52
1871.1.2 by Robert Collins
Reduce code duplication in transport-parameterised tests.
53
class TransportTests(TestTransportImplementation):
54
1530.1.3 by Robert Collins
transport implementations now tested consistently.
55
    def check_transport_contents(self, content, transport, relpath):
56
        """Check that transport.get(relpath).read() == content."""
1530.1.21 by Robert Collins
Review feedback fixes.
57
        self.assertEqualDiff(content, transport.get(relpath).read())
1530.1.15 by Robert Collins
Move put mode tests into test_transport_implementation.
58
1530.1.3 by Robert Collins
transport implementations now tested consistently.
59
    def assertListRaises(self, excClass, func, *args, **kwargs):
1530.1.21 by Robert Collins
Review feedback fixes.
60
        """Fail unless excClass is raised when the iterator from func is used.
61
        
62
        Many transport functions can return generators this makes sure
1530.1.3 by Robert Collins
transport implementations now tested consistently.
63
        to wrap them in a list() call to make sure the whole generator
64
        is run, and that the proper exception is raised.
65
        """
66
        try:
67
            list(func(*args, **kwargs))
68
        except excClass:
69
            return
70
        else:
71
            if hasattr(excClass,'__name__'): excName = excClass.__name__
72
            else: excName = str(excClass)
73
            raise self.failureException, "%s not raised" % excName
74
75
    def test_has(self):
76
        t = self.get_transport()
77
78
        files = ['a', 'b', 'e', 'g', '%']
79
        self.build_tree(files, transport=t)
80
        self.assertEqual(True, t.has('a'))
81
        self.assertEqual(False, t.has('c'))
1685.1.45 by John Arbash Meinel
Moved url functions into bzrlib.urlutils
82
        self.assertEqual(True, t.has(urlutils.escape('%')))
1530.1.3 by Robert Collins
transport implementations now tested consistently.
83
        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
84
                [True, True, False, False, True, False, True, False])
85
        self.assertEqual(True, t.has_any(['a', 'b', 'c']))
1685.1.45 by John Arbash Meinel
Moved url functions into bzrlib.urlutils
86
        self.assertEqual(False, t.has_any(['c', 'd', 'f', urlutils.escape('%%')]))
1530.1.3 by Robert Collins
transport implementations now tested consistently.
87
        self.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']))),
88
                [True, True, False, False, True, False, True, False])
89
        self.assertEqual(False, t.has_any(['c', 'c', 'c']))
90
        self.assertEqual(True, t.has_any(['b', 'b', 'b']))
91
92
    def test_get(self):
93
        t = self.get_transport()
94
95
        files = ['a', 'b', 'e', 'g']
96
        contents = ['contents of a\n',
97
                    'contents of b\n',
98
                    'contents of e\n',
99
                    'contents of g\n',
100
                    ]
1551.2.39 by abentley
Fix line endings in tests
101
        self.build_tree(files, transport=t, line_endings='binary')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
102
        self.check_transport_contents('contents of a\n', t, 'a')
103
        content_f = t.get_multi(files)
104
        for content, f in zip(contents, content_f):
105
            self.assertEqual(content, f.read())
106
107
        content_f = t.get_multi(iter(files))
108
        for content, f in zip(contents, content_f):
109
            self.assertEqual(content, f.read())
110
111
        self.assertRaises(NoSuchFile, t.get, 'c')
112
        self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])
113
        self.assertListRaises(NoSuchFile, t.get_multi, iter(['a', 'b', 'c']))
114
1955.3.3 by John Arbash Meinel
Implement and test 'get_bytes'
115
    def test_get_bytes(self):
116
        t = self.get_transport()
117
118
        files = ['a', 'b', 'e', 'g']
119
        contents = ['contents of a\n',
120
                    'contents of b\n',
121
                    'contents of e\n',
122
                    'contents of g\n',
123
                    ]
124
        self.build_tree(files, transport=t, line_endings='binary')
125
        self.check_transport_contents('contents of a\n', t, 'a')
126
127
        for content, fname in zip(contents, files):
128
            self.assertEqual(content, t.get_bytes(fname))
129
130
        self.assertRaises(NoSuchFile, t.get_bytes, 'c')
131
1530.1.3 by Robert Collins
transport implementations now tested consistently.
132
    def test_put(self):
133
        t = self.get_transport()
134
135
        if t.is_readonly():
1955.3.6 by John Arbash Meinel
Lots of deprecation warnings, but no errors
136
            return
137
1955.3.29 by John Arbash Meinel
Use applyDeprecated instead of callDeprecated
138
        self.applyDeprecated(zero_eleven, t.put, 'a', 'string\ncontents\n')
1955.3.6 by John Arbash Meinel
Lots of deprecation warnings, but no errors
139
        self.check_transport_contents('string\ncontents\n', t, 'a')
140
1955.3.29 by John Arbash Meinel
Use applyDeprecated instead of callDeprecated
141
        self.applyDeprecated(zero_eleven,
142
                             t.put, 'b', StringIO('file-like\ncontents\n'))
1955.3.6 by John Arbash Meinel
Lots of deprecation warnings, but no errors
143
        self.check_transport_contents('file-like\ncontents\n', t, 'b')
144
1955.3.1 by John Arbash Meinel
Add put_bytes() and a base-level implementation for it
145
    def test_put_bytes(self):
146
        t = self.get_transport()
147
148
        if t.is_readonly():
149
            self.assertRaises(TransportNotPossible,
150
                    t.put_bytes, 'a', 'some text for a\n')
151
            return
152
153
        t.put_bytes('a', 'some text for a\n')
154
        self.failUnless(t.has('a'))
155
        self.check_transport_contents('some text for a\n', t, 'a')
156
157
        # The contents should be overwritten
158
        t.put_bytes('a', 'new text for a\n')
159
        self.check_transport_contents('new text for a\n', t, 'a')
160
161
        self.assertRaises(NoSuchFile,
162
                          t.put_bytes, 'path/doesnt/exist/c', 'contents')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
163
1955.3.27 by John Arbash Meinel
rename non_atomic_put_* to put_*non_atomic, and re-order the functions
164
    def test_put_bytes_non_atomic(self):
165
        t = self.get_transport()
166
167
        if t.is_readonly():
168
            self.assertRaises(TransportNotPossible,
169
                    t.put_bytes_non_atomic, 'a', 'some text for a\n')
170
            return
171
172
        self.failIf(t.has('a'))
173
        t.put_bytes_non_atomic('a', 'some text for a\n')
174
        self.failUnless(t.has('a'))
175
        self.check_transport_contents('some text for a\n', t, 'a')
176
        # Put also replaces contents
177
        t.put_bytes_non_atomic('a', 'new\ncontents for\na\n')
178
        self.check_transport_contents('new\ncontents for\na\n', t, 'a')
179
180
        # Make sure we can create another file
181
        t.put_bytes_non_atomic('d', 'contents for\nd\n')
182
        # And overwrite 'a' with empty contents
183
        t.put_bytes_non_atomic('a', '')
184
        self.check_transport_contents('contents for\nd\n', t, 'd')
185
        self.check_transport_contents('', t, 'a')
186
187
        self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'no/such/path',
188
                                       'contents\n')
189
        # Now test the create_parent flag
190
        self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'dir/a',
191
                                       'contents\n')
192
        self.failIf(t.has('dir/a'))
193
        t.put_bytes_non_atomic('dir/a', 'contents for dir/a\n',
194
                               create_parent_dir=True)
195
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
196
        
197
        # But we still get NoSuchFile if we can't make the parent dir
198
        self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'not/there/a',
199
                                       'contents\n',
200
                                       create_parent_dir=True)
201
202
    def test_put_bytes_permissions(self):
203
        t = self.get_transport()
204
205
        if t.is_readonly():
206
            return
207
        if not t._can_roundtrip_unix_modebits():
208
            # Can't roundtrip, so no need to run this test
209
            return
210
        t.put_bytes('mode644', 'test text\n', mode=0644)
211
        self.assertTransportMode(t, 'mode644', 0644)
212
        t.put_bytes('mode666', 'test text\n', mode=0666)
213
        self.assertTransportMode(t, 'mode666', 0666)
214
        t.put_bytes('mode600', 'test text\n', mode=0600)
215
        self.assertTransportMode(t, 'mode600', 0600)
216
        # Yes, you can put_bytes a file such that it becomes readonly
217
        t.put_bytes('mode400', 'test text\n', mode=0400)
218
        self.assertTransportMode(t, 'mode400', 0400)
219
220
        # The default permissions should be based on the current umask
221
        umask = osutils.get_umask()
222
        t.put_bytes('nomode', 'test text\n', mode=None)
223
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
224
        
225
    def test_put_bytes_non_atomic_permissions(self):
226
        t = self.get_transport()
227
228
        if t.is_readonly():
229
            return
230
        if not t._can_roundtrip_unix_modebits():
231
            # Can't roundtrip, so no need to run this test
232
            return
233
        t.put_bytes_non_atomic('mode644', 'test text\n', mode=0644)
234
        self.assertTransportMode(t, 'mode644', 0644)
235
        t.put_bytes_non_atomic('mode666', 'test text\n', mode=0666)
236
        self.assertTransportMode(t, 'mode666', 0666)
237
        t.put_bytes_non_atomic('mode600', 'test text\n', mode=0600)
238
        self.assertTransportMode(t, 'mode600', 0600)
239
        t.put_bytes_non_atomic('mode400', 'test text\n', mode=0400)
240
        self.assertTransportMode(t, 'mode400', 0400)
241
242
        # The default permissions should be based on the current umask
243
        umask = osutils.get_umask()
244
        t.put_bytes_non_atomic('nomode', 'test text\n', mode=None)
245
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
246
        
247
    def test_put_file(self):
248
        t = self.get_transport()
249
250
        if t.is_readonly():
251
            self.assertRaises(TransportNotPossible,
252
                    t.put_file, 'a', StringIO('some text for a\n'))
253
            return
254
255
        t.put_file('a', StringIO('some text for a\n'))
256
        self.failUnless(t.has('a'))
257
        self.check_transport_contents('some text for a\n', t, 'a')
258
        # Put also replaces contents
259
        t.put_file('a', StringIO('new\ncontents for\na\n'))
260
        self.check_transport_contents('new\ncontents for\na\n', t, 'a')
261
        self.assertRaises(NoSuchFile,
262
                          t.put_file, 'path/doesnt/exist/c',
263
                              StringIO('contents'))
264
265
    def test_put_file_non_atomic(self):
266
        t = self.get_transport()
267
268
        if t.is_readonly():
269
            self.assertRaises(TransportNotPossible,
270
                    t.put_file_non_atomic, 'a', StringIO('some text for a\n'))
271
            return
272
273
        self.failIf(t.has('a'))
274
        t.put_file_non_atomic('a', StringIO('some text for a\n'))
275
        self.failUnless(t.has('a'))
276
        self.check_transport_contents('some text for a\n', t, 'a')
277
        # Put also replaces contents
278
        t.put_file_non_atomic('a', StringIO('new\ncontents for\na\n'))
279
        self.check_transport_contents('new\ncontents for\na\n', t, 'a')
280
281
        # Make sure we can create another file
282
        t.put_file_non_atomic('d', StringIO('contents for\nd\n'))
283
        # And overwrite 'a' with empty contents
284
        t.put_file_non_atomic('a', StringIO(''))
285
        self.check_transport_contents('contents for\nd\n', t, 'd')
286
        self.check_transport_contents('', t, 'a')
287
288
        self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'no/such/path',
289
                                       StringIO('contents\n'))
290
        # Now test the create_parent flag
291
        self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'dir/a',
292
                                       StringIO('contents\n'))
293
        self.failIf(t.has('dir/a'))
294
        t.put_file_non_atomic('dir/a', StringIO('contents for dir/a\n'),
1955.3.20 by John Arbash Meinel
Add non_atomic_put_bytes() and tests for it
295
                              create_parent_dir=True)
1946.1.8 by John Arbash Meinel
Update non_atomic_put to have a create_parent_dir flag
296
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
297
        
298
        # But we still get NoSuchFile if we can't make the parent dir
1955.3.27 by John Arbash Meinel
rename non_atomic_put_* to put_*non_atomic, and re-order the functions
299
        self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'not/there/a',
1955.3.20 by John Arbash Meinel
Add non_atomic_put_bytes() and tests for it
300
                                       StringIO('contents\n'),
301
                                       create_parent_dir=True)
302
1955.3.7 by John Arbash Meinel
Fix the deprecation warnings in the transport tests themselves
303
    def test_put_file_permissions(self):
1955.3.18 by John Arbash Meinel
[merge] Transport.non_atomic_put()
304
1530.1.15 by Robert Collins
Move put mode tests into test_transport_implementation.
305
        t = self.get_transport()
306
307
        if t.is_readonly():
308
            return
1711.4.32 by John Arbash Meinel
Skip permission tests on win32 no modebits
309
        if not t._can_roundtrip_unix_modebits():
310
            # Can't roundtrip, so no need to run this test
311
            return
1955.3.7 by John Arbash Meinel
Fix the deprecation warnings in the transport tests themselves
312
        t.put_file('mode644', StringIO('test text\n'), mode=0644)
1530.1.21 by Robert Collins
Review feedback fixes.
313
        self.assertTransportMode(t, 'mode644', 0644)
1955.3.7 by John Arbash Meinel
Fix the deprecation warnings in the transport tests themselves
314
        t.put_file('mode666', StringIO('test text\n'), mode=0666)
1530.1.21 by Robert Collins
Review feedback fixes.
315
        self.assertTransportMode(t, 'mode666', 0666)
1955.3.7 by John Arbash Meinel
Fix the deprecation warnings in the transport tests themselves
316
        t.put_file('mode600', StringIO('test text\n'), mode=0600)
1530.1.21 by Robert Collins
Review feedback fixes.
317
        self.assertTransportMode(t, 'mode600', 0600)
1530.1.15 by Robert Collins
Move put mode tests into test_transport_implementation.
318
        # Yes, you can put a file such that it becomes readonly
1955.3.7 by John Arbash Meinel
Fix the deprecation warnings in the transport tests themselves
319
        t.put_file('mode400', StringIO('test text\n'), mode=0400)
1530.1.21 by Robert Collins
Review feedback fixes.
320
        self.assertTransportMode(t, 'mode400', 0400)
1955.3.7 by John Arbash Meinel
Fix the deprecation warnings in the transport tests themselves
321
322
        # XXX: put_multi is deprecated, so do we really care anymore?
1955.3.29 by John Arbash Meinel
Use applyDeprecated instead of callDeprecated
323
        self.applyDeprecated(zero_eleven, t.put_multi,
324
                             [('mmode644', StringIO('text\n'))], mode=0644)
1530.1.21 by Robert Collins
Review feedback fixes.
325
        self.assertTransportMode(t, 'mmode644', 0644)
1755.3.7 by John Arbash Meinel
Clean up and write tests for permissions. Now we use fstat which should be cheap, and lets us check the permissions and the file size
326
327
        # The default permissions should be based on the current umask
328
        umask = osutils.get_umask()
1955.3.7 by John Arbash Meinel
Fix the deprecation warnings in the transport tests themselves
329
        t.put_file('nomode', StringIO('test text\n'), mode=None)
1755.3.7 by John Arbash Meinel
Clean up and write tests for permissions. Now we use fstat which should be cheap, and lets us check the permissions and the file size
330
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
331
        
1955.3.27 by John Arbash Meinel
rename non_atomic_put_* to put_*non_atomic, and re-order the functions
332
    def test_put_file_non_atomic_permissions(self):
333
        t = self.get_transport()
334
335
        if t.is_readonly():
336
            return
337
        if not t._can_roundtrip_unix_modebits():
338
            # Can't roundtrip, so no need to run this test
339
            return
340
        t.put_file_non_atomic('mode644', StringIO('test text\n'), mode=0644)
341
        self.assertTransportMode(t, 'mode644', 0644)
342
        t.put_file_non_atomic('mode666', StringIO('test text\n'), mode=0666)
343
        self.assertTransportMode(t, 'mode666', 0666)
344
        t.put_file_non_atomic('mode600', StringIO('test text\n'), mode=0600)
345
        self.assertTransportMode(t, 'mode600', 0600)
346
        # Yes, you can put_file_non_atomic a file such that it becomes readonly
347
        t.put_file_non_atomic('mode400', StringIO('test text\n'), mode=0400)
348
        self.assertTransportMode(t, 'mode400', 0400)
349
350
        # The default permissions should be based on the current umask
351
        umask = osutils.get_umask()
352
        t.put_file_non_atomic('nomode', StringIO('test text\n'), mode=None)
353
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
354
        
355
    def test_put_multi(self):
356
        t = self.get_transport()
357
358
        if t.is_readonly():
359
            return
1955.3.29 by John Arbash Meinel
Use applyDeprecated instead of callDeprecated
360
        self.assertEqual(2, self.applyDeprecated(zero_eleven,
1955.3.27 by John Arbash Meinel
rename non_atomic_put_* to put_*non_atomic, and re-order the functions
361
            t.put_multi, [('a', StringIO('new\ncontents for\na\n')),
362
                          ('d', StringIO('contents\nfor d\n'))]
363
            ))
364
        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd'])),
365
                [True, False, False, True])
366
        self.check_transport_contents('new\ncontents for\na\n', t, 'a')
367
        self.check_transport_contents('contents\nfor d\n', t, 'd')
368
1955.3.29 by John Arbash Meinel
Use applyDeprecated instead of callDeprecated
369
        self.assertEqual(2, self.applyDeprecated(zero_eleven,
1955.3.27 by John Arbash Meinel
rename non_atomic_put_* to put_*non_atomic, and re-order the functions
370
            t.put_multi, iter([('a', StringIO('diff\ncontents for\na\n')),
371
                              ('d', StringIO('another contents\nfor d\n'))])
372
            ))
373
        self.check_transport_contents('diff\ncontents for\na\n', t, 'a')
374
        self.check_transport_contents('another contents\nfor d\n', t, 'd')
375
1530.1.3 by Robert Collins
transport implementations now tested consistently.
376
    def test_mkdir(self):
377
        t = self.get_transport()
378
379
        if t.is_readonly():
380
            # cannot mkdir on readonly transports. We're not testing for 
381
            # cache coherency because cache behaviour is not currently
382
            # defined for the transport interface.
383
            self.assertRaises(TransportNotPossible, t.mkdir, '.')
384
            self.assertRaises(TransportNotPossible, t.mkdir, 'new_dir')
385
            self.assertRaises(TransportNotPossible, t.mkdir_multi, ['new_dir'])
386
            self.assertRaises(TransportNotPossible, t.mkdir, 'path/doesnt/exist')
387
            return
388
        # Test mkdir
389
        t.mkdir('dir_a')
390
        self.assertEqual(t.has('dir_a'), True)
391
        self.assertEqual(t.has('dir_b'), False)
392
393
        t.mkdir('dir_b')
394
        self.assertEqual(t.has('dir_b'), True)
395
396
        t.mkdir_multi(['dir_c', 'dir_d'])
397
398
        t.mkdir_multi(iter(['dir_e', 'dir_f']))
399
        self.assertEqual(list(t.has_multi(
400
            ['dir_a', 'dir_b', 'dir_c', 'dir_q',
401
             'dir_d', 'dir_e', 'dir_f', 'dir_b'])),
402
            [True, True, True, False,
403
             True, True, True, True])
404
405
        # we were testing that a local mkdir followed by a transport
406
        # mkdir failed thusly, but given that we * in one process * do not
407
        # concurrently fiddle with disk dirs and then use transport to do 
408
        # things, the win here seems marginal compared to the constraint on
409
        # the interface. RBC 20051227
410
        t.mkdir('dir_g')
411
        self.assertRaises(FileExists, t.mkdir, 'dir_g')
412
413
        # Test get/put in sub-directories
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
414
        t.put_bytes('dir_a/a', 'contents of dir_a/a')
415
        t.put_file('dir_b/b', StringIO('contents of dir_b/b'))
1530.1.3 by Robert Collins
transport implementations now tested consistently.
416
        self.check_transport_contents('contents of dir_a/a', t, 'dir_a/a')
417
        self.check_transport_contents('contents of dir_b/b', t, 'dir_b/b')
418
1530.1.4 by Robert Collins
integrate Memory tests into transport interface tests.
419
        # mkdir of a dir with an absent parent
420
        self.assertRaises(NoSuchFile, t.mkdir, 'missing/dir')
421
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
422
    def test_mkdir_permissions(self):
423
        t = self.get_transport()
424
        if t.is_readonly():
425
            return
1608.2.7 by Martin Pool
Rename supports_unix_modebits to _can_roundtrip_unix_modebits for clarity
426
        if not t._can_roundtrip_unix_modebits():
1608.2.5 by Martin Pool
Add Transport.supports_unix_modebits, so tests can
427
            # no sense testing on this transport
428
            return
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
429
        # Test mkdir with a mode
430
        t.mkdir('dmode755', mode=0755)
1530.1.21 by Robert Collins
Review feedback fixes.
431
        self.assertTransportMode(t, 'dmode755', 0755)
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
432
        t.mkdir('dmode555', mode=0555)
1530.1.21 by Robert Collins
Review feedback fixes.
433
        self.assertTransportMode(t, 'dmode555', 0555)
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
434
        t.mkdir('dmode777', mode=0777)
1530.1.21 by Robert Collins
Review feedback fixes.
435
        self.assertTransportMode(t, 'dmode777', 0777)
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
436
        t.mkdir('dmode700', mode=0700)
1530.1.21 by Robert Collins
Review feedback fixes.
437
        self.assertTransportMode(t, 'dmode700', 0700)
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
438
        t.mkdir_multi(['mdmode755'], mode=0755)
1530.1.21 by Robert Collins
Review feedback fixes.
439
        self.assertTransportMode(t, 'mdmode755', 0755)
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
440
1755.3.7 by John Arbash Meinel
Clean up and write tests for permissions. Now we use fstat which should be cheap, and lets us check the permissions and the file size
441
        # Default mode should be based on umask
442
        umask = osutils.get_umask()
443
        t.mkdir('dnomode', mode=None)
444
        self.assertTransportMode(t, 'dnomode', 0777 & ~umask)
445
1530.1.3 by Robert Collins
transport implementations now tested consistently.
446
    def test_copy_to(self):
1534.4.21 by Robert Collins
Extend the copy_to tests to smoke test server-to-same-server copies to catch optimised code paths, and fix sftps optimised code path by removing dead code.
447
        # FIXME: test:   same server to same server (partly done)
448
        # same protocol two servers
449
        # and    different protocols (done for now except for MemoryTransport.
450
        # - RBC 20060122
1530.1.3 by Robert Collins
transport implementations now tested consistently.
451
        from bzrlib.transport.memory import MemoryTransport
1534.4.21 by Robert Collins
Extend the copy_to tests to smoke test server-to-same-server copies to catch optimised code paths, and fix sftps optimised code path by removing dead code.
452
453
        def simple_copy_files(transport_from, transport_to):
454
            files = ['a', 'b', 'c', 'd']
455
            self.build_tree(files, transport=transport_from)
1563.2.3 by Robert Collins
Change the return signature of transport.append and append_multi to return the length of the pre-append content.
456
            self.assertEqual(4, transport_from.copy_to(files, transport_to))
1534.4.21 by Robert Collins
Extend the copy_to tests to smoke test server-to-same-server copies to catch optimised code paths, and fix sftps optimised code path by removing dead code.
457
            for f in files:
458
                self.check_transport_contents(transport_to.get(f).read(),
459
                                              transport_from, f)
460
1530.1.3 by Robert Collins
transport implementations now tested consistently.
461
        t = self.get_transport()
1685.1.42 by John Arbash Meinel
A couple more fixes to make sure memory:/// works correctly.
462
        temp_transport = MemoryTransport('memory:///')
1534.4.21 by Robert Collins
Extend the copy_to tests to smoke test server-to-same-server copies to catch optimised code paths, and fix sftps optimised code path by removing dead code.
463
        simple_copy_files(t, temp_transport)
464
        if not t.is_readonly():
465
            t.mkdir('copy_to_simple')
466
            t2 = t.clone('copy_to_simple')
467
            simple_copy_files(t, t2)
1530.1.3 by Robert Collins
transport implementations now tested consistently.
468
469
470
        # Test that copying into a missing directory raises
471
        # NoSuchFile
472
        if t.is_readonly():
1530.1.21 by Robert Collins
Review feedback fixes.
473
            self.build_tree(['e/', 'e/f'])
1530.1.3 by Robert Collins
transport implementations now tested consistently.
474
        else:
475
            t.mkdir('e')
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
476
            t.put_bytes('e/f', 'contents of e')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
477
        self.assertRaises(NoSuchFile, t.copy_to, ['e/f'], temp_transport)
478
        temp_transport.mkdir('e')
479
        t.copy_to(['e/f'], temp_transport)
480
481
        del temp_transport
1685.1.42 by John Arbash Meinel
A couple more fixes to make sure memory:/// works correctly.
482
        temp_transport = MemoryTransport('memory:///')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
483
484
        files = ['a', 'b', 'c', 'd']
485
        t.copy_to(iter(files), temp_transport)
486
        for f in files:
487
            self.check_transport_contents(temp_transport.get(f).read(),
488
                                          t, f)
489
        del temp_transport
490
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
491
        for mode in (0666, 0644, 0600, 0400):
1685.1.42 by John Arbash Meinel
A couple more fixes to make sure memory:/// works correctly.
492
            temp_transport = MemoryTransport("memory:///")
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
493
            t.copy_to(files, temp_transport, mode=mode)
494
            for f in files:
1530.1.21 by Robert Collins
Review feedback fixes.
495
                self.assertTransportMode(temp_transport, f, mode)
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
496
1530.1.3 by Robert Collins
transport implementations now tested consistently.
497
    def test_append(self):
498
        t = self.get_transport()
499
500
        if t.is_readonly():
1955.3.15 by John Arbash Meinel
Deprecate 'Transport.append' in favor of Transport.append_file or Transport.append_bytes
501
            return
502
        t.put_bytes('a', 'diff\ncontents for\na\n')
503
        t.put_bytes('b', 'contents\nfor b\n')
504
1955.3.29 by John Arbash Meinel
Use applyDeprecated instead of callDeprecated
505
        self.assertEqual(20, self.applyDeprecated(zero_eleven,
1955.3.15 by John Arbash Meinel
Deprecate 'Transport.append' in favor of Transport.append_file or Transport.append_bytes
506
            t.append, 'a', StringIO('add\nsome\nmore\ncontents\n')))
507
508
        self.check_transport_contents(
509
            'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
510
            t, 'a')
511
512
        # And we can create new files, too
1955.3.29 by John Arbash Meinel
Use applyDeprecated instead of callDeprecated
513
        self.assertEqual(0, self.applyDeprecated(zero_eleven,
1955.3.15 by John Arbash Meinel
Deprecate 'Transport.append' in favor of Transport.append_file or Transport.append_bytes
514
            t.append, 'c', StringIO('some text\nfor a missing file\n')))
515
        self.check_transport_contents('some text\nfor a missing file\n',
516
                                      t, 'c')
517
    def test_append_file(self):
518
        t = self.get_transport()
519
520
        if t.is_readonly():
1530.1.3 by Robert Collins
transport implementations now tested consistently.
521
            self.assertRaises(TransportNotPossible,
1955.3.15 by John Arbash Meinel
Deprecate 'Transport.append' in favor of Transport.append_file or Transport.append_bytes
522
                    t.append_file, 'a', 'add\nsome\nmore\ncontents\n')
1955.3.2 by John Arbash Meinel
Implement and test 'Transport.append_bytes', cleanup the tests of plain append
523
            return
1955.3.10 by John Arbash Meinel
clean up append, append_bytes, and append_multi tests
524
        t.put_bytes('a', 'diff\ncontents for\na\n')
525
        t.put_bytes('b', 'contents\nfor b\n')
1955.3.2 by John Arbash Meinel
Implement and test 'Transport.append_bytes', cleanup the tests of plain append
526
527
        self.assertEqual(20,
1955.3.15 by John Arbash Meinel
Deprecate 'Transport.append' in favor of Transport.append_file or Transport.append_bytes
528
            t.append_file('a', StringIO('add\nsome\nmore\ncontents\n')))
1530.1.3 by Robert Collins
transport implementations now tested consistently.
529
530
        self.check_transport_contents(
531
            'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
532
            t, 'a')
533
1955.3.10 by John Arbash Meinel
clean up append, append_bytes, and append_multi tests
534
        # a file with no parent should fail..
535
        self.assertRaises(NoSuchFile,
1955.3.15 by John Arbash Meinel
Deprecate 'Transport.append' in favor of Transport.append_file or Transport.append_bytes
536
                          t.append_file, 'missing/path', StringIO('content'))
1955.3.10 by John Arbash Meinel
clean up append, append_bytes, and append_multi tests
537
538
        # And we can create new files, too
539
        self.assertEqual(0,
1955.3.15 by John Arbash Meinel
Deprecate 'Transport.append' in favor of Transport.append_file or Transport.append_bytes
540
            t.append_file('c', StringIO('some text\nfor a missing file\n')))
1955.3.10 by John Arbash Meinel
clean up append, append_bytes, and append_multi tests
541
        self.check_transport_contents('some text\nfor a missing file\n',
542
                                      t, 'c')
543
544
    def test_append_bytes(self):
545
        t = self.get_transport()
546
547
        if t.is_readonly():
548
            self.assertRaises(TransportNotPossible,
549
                    t.append_bytes, 'a', 'add\nsome\nmore\ncontents\n')
550
            return
551
552
        self.assertEqual(0, t.append_bytes('a', 'diff\ncontents for\na\n'))
553
        self.assertEqual(0, t.append_bytes('b', 'contents\nfor b\n'))
554
555
        self.assertEqual(20,
556
            t.append_bytes('a', 'add\nsome\nmore\ncontents\n'))
557
558
        self.check_transport_contents(
559
            'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
560
            t, 'a')
561
562
        # a file with no parent should fail..
563
        self.assertRaises(NoSuchFile,
564
                          t.append_bytes, 'missing/path', 'content')
565
566
    def test_append_multi(self):
567
        t = self.get_transport()
568
569
        if t.is_readonly():
570
            return
571
        t.put_bytes('a', 'diff\ncontents for\na\n'
572
                         'add\nsome\nmore\ncontents\n')
573
        t.put_bytes('b', 'contents\nfor b\n')
574
1955.3.2 by John Arbash Meinel
Implement and test 'Transport.append_bytes', cleanup the tests of plain append
575
        self.assertEqual((43, 15),
576
            t.append_multi([('a', StringIO('and\nthen\nsome\nmore\n')),
577
                            ('b', StringIO('some\nmore\nfor\nb\n'))]))
578
1530.1.3 by Robert Collins
transport implementations now tested consistently.
579
        self.check_transport_contents(
580
            'diff\ncontents for\na\n'
581
            'add\nsome\nmore\ncontents\n'
582
            'and\nthen\nsome\nmore\n',
583
            t, 'a')
584
        self.check_transport_contents(
585
                'contents\nfor b\n'
586
                'some\nmore\nfor\nb\n',
587
                t, 'b')
588
1955.3.2 by John Arbash Meinel
Implement and test 'Transport.append_bytes', cleanup the tests of plain append
589
        self.assertEqual((62, 31),
590
            t.append_multi(iter([('a', StringIO('a little bit more\n')),
591
                                 ('b', StringIO('from an iterator\n'))])))
1530.1.3 by Robert Collins
transport implementations now tested consistently.
592
        self.check_transport_contents(
593
            'diff\ncontents for\na\n'
594
            'add\nsome\nmore\ncontents\n'
595
            'and\nthen\nsome\nmore\n'
596
            'a little bit more\n',
597
            t, 'a')
598
        self.check_transport_contents(
599
                'contents\nfor b\n'
600
                'some\nmore\nfor\nb\n'
601
                'from an iterator\n',
602
                t, 'b')
603
1955.3.2 by John Arbash Meinel
Implement and test 'Transport.append_bytes', cleanup the tests of plain append
604
        self.assertEqual((80, 0),
605
            t.append_multi([('a', StringIO('some text in a\n')),
606
                            ('d', StringIO('missing file r\n'))]))
607
1530.1.3 by Robert Collins
transport implementations now tested consistently.
608
        self.check_transport_contents(
609
            'diff\ncontents for\na\n'
610
            'add\nsome\nmore\ncontents\n'
611
            'and\nthen\nsome\nmore\n'
612
            'a little bit more\n'
613
            'some text in a\n',
614
            t, 'a')
615
        self.check_transport_contents('missing file r\n', t, 'd')
616
1955.3.15 by John Arbash Meinel
Deprecate 'Transport.append' in favor of Transport.append_file or Transport.append_bytes
617
    def test_append_file_mode(self):
1955.3.10 by John Arbash Meinel
clean up append, append_bytes, and append_multi tests
618
        """Check that append accepts a mode parameter"""
1666.1.6 by Robert Collins
Make knit the default format.
619
        # check append accepts a mode
620
        t = self.get_transport()
621
        if t.is_readonly():
1955.3.10 by John Arbash Meinel
clean up append, append_bytes, and append_multi tests
622
            self.assertRaises(TransportNotPossible,
1955.3.15 by John Arbash Meinel
Deprecate 'Transport.append' in favor of Transport.append_file or Transport.append_bytes
623
                t.append_file, 'f', StringIO('f'), mode=None)
1666.1.6 by Robert Collins
Make knit the default format.
624
            return
1955.3.15 by John Arbash Meinel
Deprecate 'Transport.append' in favor of Transport.append_file or Transport.append_bytes
625
        t.append_file('f', StringIO('f'), mode=None)
1666.1.6 by Robert Collins
Make knit the default format.
626
        
1955.3.2 by John Arbash Meinel
Implement and test 'Transport.append_bytes', cleanup the tests of plain append
627
    def test_append_bytes_mode(self):
628
        # check append_bytes accepts a mode
629
        t = self.get_transport()
630
        if t.is_readonly():
1955.3.10 by John Arbash Meinel
clean up append, append_bytes, and append_multi tests
631
            self.assertRaises(TransportNotPossible,
632
                t.append_bytes, 'f', 'f', mode=None)
1955.3.2 by John Arbash Meinel
Implement and test 'Transport.append_bytes', cleanup the tests of plain append
633
            return
1955.3.10 by John Arbash Meinel
clean up append, append_bytes, and append_multi tests
634
        t.append_bytes('f', 'f', mode=None)
1955.3.2 by John Arbash Meinel
Implement and test 'Transport.append_bytes', cleanup the tests of plain append
635
        
1530.1.3 by Robert Collins
transport implementations now tested consistently.
636
    def test_delete(self):
637
        # TODO: Test Transport.delete
638
        t = self.get_transport()
639
640
        # Not much to do with a readonly transport
641
        if t.is_readonly():
1534.4.9 by Robert Collins
Add a readonly decorator for transports.
642
            self.assertRaises(TransportNotPossible, t.delete, 'missing')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
643
            return
644
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
645
        t.put_bytes('a', 'a little bit of text\n')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
646
        self.failUnless(t.has('a'))
647
        t.delete('a')
648
        self.failIf(t.has('a'))
649
650
        self.assertRaises(NoSuchFile, t.delete, 'a')
651
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
652
        t.put_bytes('a', 'a text\n')
653
        t.put_bytes('b', 'b text\n')
654
        t.put_bytes('c', 'c text\n')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
655
        self.assertEqual([True, True, True],
656
                list(t.has_multi(['a', 'b', 'c'])))
657
        t.delete_multi(['a', 'c'])
658
        self.assertEqual([False, True, False],
659
                list(t.has_multi(['a', 'b', 'c'])))
660
        self.failIf(t.has('a'))
661
        self.failUnless(t.has('b'))
662
        self.failIf(t.has('c'))
663
664
        self.assertRaises(NoSuchFile,
665
                t.delete_multi, ['a', 'b', 'c'])
666
667
        self.assertRaises(NoSuchFile,
668
                t.delete_multi, iter(['a', 'b', 'c']))
669
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
670
        t.put_bytes('a', 'another a text\n')
671
        t.put_bytes('c', 'another c text\n')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
672
        t.delete_multi(iter(['a', 'b', 'c']))
673
674
        # We should have deleted everything
675
        # SftpServer creates control files in the
676
        # working directory, so we can just do a
677
        # plain "listdir".
678
        # self.assertEqual([], os.listdir('.'))
679
1534.4.15 by Robert Collins
Remove shutil dependency in upgrade - create a delete_tree method for transports.
680
    def test_rmdir(self):
681
        t = self.get_transport()
682
        # Not much to do with a readonly transport
683
        if t.is_readonly():
684
            self.assertRaises(TransportNotPossible, t.rmdir, 'missing')
685
            return
686
        t.mkdir('adir')
687
        t.mkdir('adir/bdir')
688
        t.rmdir('adir/bdir')
1948.3.12 by Vincent LADEUIL
Fix Aaron's third review remarks.
689
        # ftp may not be able to raise NoSuchFile for lack of
690
        # details when failing
691
        self.assertRaises((NoSuchFile, PathError), t.rmdir, 'adir/bdir')
1534.4.15 by Robert Collins
Remove shutil dependency in upgrade - create a delete_tree method for transports.
692
        t.rmdir('adir')
1948.3.12 by Vincent LADEUIL
Fix Aaron's third review remarks.
693
        self.assertRaises((NoSuchFile, PathError), t.rmdir, 'adir')
1534.4.15 by Robert Collins
Remove shutil dependency in upgrade - create a delete_tree method for transports.
694
1553.5.10 by Martin Pool
New DirectoryNotEmpty exception, and raise this from local and memory
695
    def test_rmdir_not_empty(self):
696
        """Deleting a non-empty directory raises an exception
697
        
698
        sftp (and possibly others) don't give us a specific "directory not
699
        empty" exception -- we can just see that the operation failed.
700
        """
701
        t = self.get_transport()
702
        if t.is_readonly():
703
            return
704
        t.mkdir('adir')
705
        t.mkdir('adir/bdir')
706
        self.assertRaises(PathError, t.rmdir, 'adir')
707
1553.5.13 by Martin Pool
New Transport.rename that mustn't overwrite
708
    def test_rename_dir_succeeds(self):
709
        t = self.get_transport()
710
        if t.is_readonly():
711
            raise TestSkipped("transport is readonly")
712
        t.mkdir('adir')
713
        t.mkdir('adir/asubdir')
714
        t.rename('adir', 'bdir')
715
        self.assertTrue(t.has('bdir/asubdir'))
716
        self.assertFalse(t.has('adir'))
717
718
    def test_rename_dir_nonempty(self):
719
        """Attempting to replace a nonemtpy directory should fail"""
720
        t = self.get_transport()
721
        if t.is_readonly():
722
            raise TestSkipped("transport is readonly")
723
        t.mkdir('adir')
724
        t.mkdir('adir/asubdir')
725
        t.mkdir('bdir')
726
        t.mkdir('bdir/bsubdir')
727
        self.assertRaises(PathError, t.rename, 'bdir', 'adir')
728
        # nothing was changed so it should still be as before
729
        self.assertTrue(t.has('bdir/bsubdir'))
730
        self.assertFalse(t.has('adir/bdir'))
731
        self.assertFalse(t.has('adir/bsubdir'))
732
1534.4.15 by Robert Collins
Remove shutil dependency in upgrade - create a delete_tree method for transports.
733
    def test_delete_tree(self):
734
        t = self.get_transport()
735
736
        # Not much to do with a readonly transport
737
        if t.is_readonly():
738
            self.assertRaises(TransportNotPossible, t.delete_tree, 'missing')
739
            return
740
741
        # and does it like listing ?
742
        t.mkdir('adir')
743
        try:
744
            t.delete_tree('adir')
745
        except TransportNotPossible:
746
            # ok, this transport does not support delete_tree
747
            return
748
        
749
        # did it delete that trivial case?
750
        self.assertRaises(NoSuchFile, t.stat, 'adir')
751
752
        self.build_tree(['adir/',
753
                         'adir/file', 
754
                         'adir/subdir/', 
755
                         'adir/subdir/file', 
756
                         'adir/subdir2/',
757
                         'adir/subdir2/file',
758
                         ], transport=t)
759
760
        t.delete_tree('adir')
761
        # adir should be gone now.
762
        self.assertRaises(NoSuchFile, t.stat, 'adir')
763
1530.1.3 by Robert Collins
transport implementations now tested consistently.
764
    def test_move(self):
765
        t = self.get_transport()
766
767
        if t.is_readonly():
768
            return
769
770
        # TODO: I would like to use os.listdir() to
771
        # make sure there are no extra files, but SftpServer
772
        # creates control files in the working directory
773
        # perhaps all of this could be done in a subdirectory
774
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
775
        t.put_bytes('a', 'a first file\n')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
776
        self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
777
778
        t.move('a', 'b')
779
        self.failUnless(t.has('b'))
780
        self.failIf(t.has('a'))
781
782
        self.check_transport_contents('a first file\n', t, 'b')
783
        self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
784
785
        # Overwrite a file
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
786
        t.put_bytes('c', 'c this file\n')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
787
        t.move('c', 'b')
788
        self.failIf(t.has('c'))
789
        self.check_transport_contents('c this file\n', t, 'b')
790
791
        # TODO: Try to write a test for atomicity
792
        # TODO: Test moving into a non-existant subdirectory
793
        # TODO: Test Transport.move_multi
794
795
    def test_copy(self):
796
        t = self.get_transport()
797
798
        if t.is_readonly():
799
            return
800
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
801
        t.put_bytes('a', 'a file\n')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
802
        t.copy('a', 'b')
803
        self.check_transport_contents('a file\n', t, 'b')
804
805
        self.assertRaises(NoSuchFile, t.copy, 'c', 'd')
806
        os.mkdir('c')
807
        # What should the assert be if you try to copy a
808
        # file over a directory?
809
        #self.assertRaises(Something, t.copy, 'a', 'c')
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
810
        t.put_bytes('d', 'text in d\n')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
811
        t.copy('d', 'b')
812
        self.check_transport_contents('text in d\n', t, 'b')
813
814
        # TODO: test copy_multi
815
816
    def test_connection_error(self):
817
        """ConnectionError is raised when connection is impossible"""
1530.1.9 by Robert Collins
Test bogus urls with http in the new infrastructure.
818
        try:
819
            url = self._server.get_bogus_url()
820
        except NotImplementedError:
821
            raise TestSkipped("Transport %s has no bogus URL support." %
822
                              self._server.__class__)
1530.1.3 by Robert Collins
transport implementations now tested consistently.
823
        try:
1711.2.42 by John Arbash Meinel
enable bogus_url support for SFTP tests
824
            t = bzrlib.transport.get_transport(url)
1530.1.3 by Robert Collins
transport implementations now tested consistently.
825
            t.get('.bzr/branch')
826
        except (ConnectionError, NoSuchFile), e:
827
            pass
828
        except (Exception), e:
1786.1.27 by John Arbash Meinel
Fix up the http transports so that tests pass with the new configuration.
829
            self.fail('Wrong exception thrown (%s.%s): %s' 
830
                        % (e.__class__.__module__, e.__class__.__name__, e))
1530.1.3 by Robert Collins
transport implementations now tested consistently.
831
        else:
1707.3.11 by John Arbash Meinel
fixing more tests.
832
            self.fail('Did not get the expected ConnectionError or NoSuchFile.')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
833
834
    def test_stat(self):
835
        # TODO: Test stat, just try once, and if it throws, stop testing
836
        from stat import S_ISDIR, S_ISREG
837
838
        t = self.get_transport()
839
840
        try:
841
            st = t.stat('.')
842
        except TransportNotPossible, e:
843
            # This transport cannot stat
844
            return
845
846
        paths = ['a', 'b/', 'b/c', 'b/d/', 'b/d/e']
847
        sizes = [14, 0, 16, 0, 18] 
1551.2.39 by abentley
Fix line endings in tests
848
        self.build_tree(paths, transport=t, line_endings='binary')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
849
850
        for path, size in zip(paths, sizes):
851
            st = t.stat(path)
852
            if path.endswith('/'):
853
                self.failUnless(S_ISDIR(st.st_mode))
854
                # directory sizes are meaningless
855
            else:
856
                self.failUnless(S_ISREG(st.st_mode))
857
                self.assertEqual(size, st.st_size)
858
859
        remote_stats = list(t.stat_multi(paths))
860
        remote_iter_stats = list(t.stat_multi(iter(paths)))
861
862
        self.assertRaises(NoSuchFile, t.stat, 'q')
863
        self.assertRaises(NoSuchFile, t.stat, 'b/a')
864
865
        self.assertListRaises(NoSuchFile, t.stat_multi, ['a', 'c', 'd'])
866
        self.assertListRaises(NoSuchFile, t.stat_multi, iter(['a', 'c', 'd']))
1534.4.15 by Robert Collins
Remove shutil dependency in upgrade - create a delete_tree method for transports.
867
        self.build_tree(['subdir/', 'subdir/file'], transport=t)
868
        subdir = t.clone('subdir')
869
        subdir.stat('./file')
1534.4.26 by Robert Collins
Move working tree initialisation out from Branch.initialize, deprecated Branch.initialize to Branch.create.
870
        subdir.stat('.')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
871
872
    def test_list_dir(self):
873
        # TODO: Test list_dir, just try once, and if it throws, stop testing
874
        t = self.get_transport()
875
        
876
        if not t.listable():
877
            self.assertRaises(TransportNotPossible, t.list_dir, '.')
878
            return
879
880
        def sorted_list(d):
881
            l = list(t.list_dir(d))
882
            l.sort()
883
            return l
884
885
        # SftpServer creates control files in the working directory
886
        # so lets move down a directory to avoid those.
1534.4.9 by Robert Collins
Add a readonly decorator for transports.
887
        if not t.is_readonly():
888
            t.mkdir('wd')
889
        else:
890
            os.mkdir('wd')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
891
        t = t.clone('wd')
892
1910.7.1 by Andrew Bennetts
Make sure list_dir always returns url-escaped names.
893
        self.assertEqual([], sorted_list('.'))
1534.4.15 by Robert Collins
Remove shutil dependency in upgrade - create a delete_tree method for transports.
894
        # c2 is precisely one letter longer than c here to test that
895
        # suffixing is not confused.
1959.2.1 by John Arbash Meinel
David Allouche: Make transports return escaped paths
896
        # a%25b checks that quoting is done consistently across transports
897
        tree_names = ['a', 'a%25b', 'b', 'c/', 'c/d', 'c/e', 'c2/']
1534.4.9 by Robert Collins
Add a readonly decorator for transports.
898
        if not t.is_readonly():
1959.2.1 by John Arbash Meinel
David Allouche: Make transports return escaped paths
899
            self.build_tree(tree_names, transport=t)
1534.4.9 by Robert Collins
Add a readonly decorator for transports.
900
        else:
1959.2.1 by John Arbash Meinel
David Allouche: Make transports return escaped paths
901
            self.build_tree(['wd/' + name for name in tree_names])
1530.1.3 by Robert Collins
transport implementations now tested consistently.
902
1959.2.1 by John Arbash Meinel
David Allouche: Make transports return escaped paths
903
        self.assertEqual(
1959.2.3 by John Arbash Meinel
Remove some unicode string notations
904
            ['a', 'a%2525b', 'b', 'c', 'c2'], sorted_list('.'))
1910.7.1 by Andrew Bennetts
Make sure list_dir always returns url-escaped names.
905
        self.assertEqual(['d', 'e'], sorted_list('c'))
1530.1.3 by Robert Collins
transport implementations now tested consistently.
906
1534.4.9 by Robert Collins
Add a readonly decorator for transports.
907
        if not t.is_readonly():
908
            t.delete('c/d')
909
            t.delete('b')
910
        else:
911
            os.unlink('wd/c/d')
912
            os.unlink('wd/b')
913
            
1959.2.3 by John Arbash Meinel
Remove some unicode string notations
914
        self.assertEqual(['a', 'a%2525b', 'c', 'c2'], sorted_list('.'))
1910.7.1 by Andrew Bennetts
Make sure list_dir always returns url-escaped names.
915
        self.assertEqual(['e'], sorted_list('c'))
1530.1.3 by Robert Collins
transport implementations now tested consistently.
916
1662.1.12 by Martin Pool
Translate unknown sftp errors to PathError, no NoSuchFile
917
        self.assertListRaises(PathError, t.list_dir, 'q')
918
        self.assertListRaises(PathError, t.list_dir, 'c/f')
919
        self.assertListRaises(PathError, t.list_dir, 'a')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
920
1910.7.1 by Andrew Bennetts
Make sure list_dir always returns url-escaped names.
921
    def test_list_dir_result_is_url_escaped(self):
922
        t = self.get_transport()
923
        if not t.listable():
924
            raise TestSkipped("transport not listable")
925
926
        if not t.is_readonly():
927
            self.build_tree(['a/', 'a/%'], transport=t)
928
        else:
929
            self.build_tree(['a/', 'a/%'])
930
        
1910.7.2 by Andrew Bennetts
Also assert that list_dir returns plain str objects.
931
        names = list(t.list_dir('a'))
932
        self.assertEqual(['%25'], names)
933
        self.assertIsInstance(names[0], str)
1910.7.1 by Andrew Bennetts
Make sure list_dir always returns url-escaped names.
934
1530.1.3 by Robert Collins
transport implementations now tested consistently.
935
    def test_clone(self):
936
        # TODO: Test that clone moves up and down the filesystem
937
        t1 = self.get_transport()
938
939
        self.build_tree(['a', 'b/', 'b/c'], transport=t1)
940
941
        self.failUnless(t1.has('a'))
942
        self.failUnless(t1.has('b/c'))
943
        self.failIf(t1.has('c'))
944
945
        t2 = t1.clone('b')
946
        self.assertEqual(t1.base + 'b/', t2.base)
947
948
        self.failUnless(t2.has('c'))
949
        self.failIf(t2.has('a'))
950
951
        t3 = t2.clone('..')
952
        self.failUnless(t3.has('a'))
953
        self.failIf(t3.has('c'))
954
955
        self.failIf(t1.has('b/d'))
956
        self.failIf(t2.has('d'))
957
        self.failIf(t3.has('b/d'))
958
959
        if t1.is_readonly():
960
            open('b/d', 'wb').write('newfile\n')
961
        else:
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
962
            t2.put_bytes('d', 'newfile\n')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
963
964
        self.failUnless(t1.has('b/d'))
965
        self.failUnless(t2.has('d'))
966
        self.failUnless(t3.has('b/d'))
967
968
    def test_relpath(self):
969
        t = self.get_transport()
970
        self.assertEqual('', t.relpath(t.base))
971
        # base ends with /
972
        self.assertEqual('', t.relpath(t.base[:-1]))
973
        # subdirs which dont exist should still give relpaths.
974
        self.assertEqual('foo', t.relpath(t.base + 'foo'))
975
        # trailing slash should be the same.
976
        self.assertEqual('foo', t.relpath(t.base + 'foo/'))
977
1636.1.1 by Robert Collins
Fix calling relpath() and abspath() on transports at their root.
978
    def test_relpath_at_root(self):
979
        t = self.get_transport()
980
        # clone all the way to the top
981
        new_transport = t.clone('..')
982
        while new_transport.base != t.base:
983
            t = new_transport
984
            new_transport = t.clone('..')
985
        # we must be able to get a relpath below the root
986
        self.assertEqual('', t.relpath(t.base))
987
        # and a deeper one should work too
988
        self.assertEqual('foo/bar', t.relpath(t.base + 'foo/bar'))
989
1530.1.3 by Robert Collins
transport implementations now tested consistently.
990
    def test_abspath(self):
991
        # smoke test for abspath. Corner cases for backends like unix fs's
992
        # that have aliasing problems like symlinks should go in backend
993
        # specific test cases.
994
        transport = self.get_transport()
1540.3.24 by Martin Pool
Add new protocol 'http+pycurl' that always uses PyCurl.
995
        
996
        # disabled because some transports might normalize urls in generating
997
        # the abspath - eg http+pycurl-> just http -- mbp 20060308 
1530.1.3 by Robert Collins
transport implementations now tested consistently.
998
        self.assertEqual(transport.base + 'relpath',
999
                         transport.abspath('relpath'))
1000
1685.1.9 by John Arbash Meinel
Updated LocalTransport so that it's base is now a URL rather than a local path. This helps consistency with all other functions. To do so, I added local_abspath() which returns the local path, and local_path_to/from_url
1001
    def test_local_abspath(self):
1002
        transport = self.get_transport()
1003
        try:
1004
            p = transport.local_abspath('.')
1005
        except TransportNotPossible:
1006
            pass # This is not a local transport
1007
        else:
1008
            self.assertEqual(getcwd(), p)
1009
1636.1.1 by Robert Collins
Fix calling relpath() and abspath() on transports at their root.
1010
    def test_abspath_at_root(self):
1011
        t = self.get_transport()
1012
        # clone all the way to the top
1013
        new_transport = t.clone('..')
1014
        while new_transport.base != t.base:
1015
            t = new_transport
1016
            new_transport = t.clone('..')
1017
        # we must be able to get a abspath of the root when we ask for
1018
        # t.abspath('..') - this due to our choice that clone('..')
1019
        # should return the root from the root, combined with the desire that
1020
        # the url from clone('..') and from abspath('..') should be the same.
1021
        self.assertEqual(t.base, t.abspath('..'))
1022
        # '' should give us the root
1023
        self.assertEqual(t.base, t.abspath(''))
1024
        # and a path should append to the url
1025
        self.assertEqual(t.base + 'foo', t.abspath('foo'))
1026
1530.1.3 by Robert Collins
transport implementations now tested consistently.
1027
    def test_iter_files_recursive(self):
1530.1.4 by Robert Collins
integrate Memory tests into transport interface tests.
1028
        transport = self.get_transport()
1029
        if not transport.listable():
1553.5.13 by Martin Pool
New Transport.rename that mustn't overwrite
1030
            self.assertRaises(TransportNotPossible,
1530.1.4 by Robert Collins
integrate Memory tests into transport interface tests.
1031
                              transport.iter_files_recursive)
1032
            return
1553.5.13 by Martin Pool
New Transport.rename that mustn't overwrite
1033
        self.build_tree(['isolated/',
1530.1.4 by Robert Collins
integrate Memory tests into transport interface tests.
1034
                         'isolated/dir/',
1035
                         'isolated/dir/foo',
1036
                         'isolated/dir/bar',
1959.2.1 by John Arbash Meinel
David Allouche: Make transports return escaped paths
1037
                         'isolated/dir/b%25z', # make sure quoting is correct
1530.1.4 by Robert Collins
integrate Memory tests into transport interface tests.
1038
                         'isolated/bar'],
1039
                        transport=transport)
1530.1.3 by Robert Collins
transport implementations now tested consistently.
1040
        paths = set(transport.iter_files_recursive())
1553.5.13 by Martin Pool
New Transport.rename that mustn't overwrite
1041
        # nb the directories are not converted
1042
        self.assertEqual(paths,
1043
                    set(['isolated/dir/foo',
1044
                         'isolated/dir/bar',
1959.2.1 by John Arbash Meinel
David Allouche: Make transports return escaped paths
1045
                         'isolated/dir/b%2525z',
1553.5.13 by Martin Pool
New Transport.rename that mustn't overwrite
1046
                         'isolated/bar']))
1047
        sub_transport = transport.clone('isolated')
1048
        paths = set(sub_transport.iter_files_recursive())
1959.2.1 by John Arbash Meinel
David Allouche: Make transports return escaped paths
1049
        self.assertEqual(paths,
1050
            set(['dir/foo', 'dir/bar', 'dir/b%2525z', 'bar']))
1051
1052
    def test_copy_tree(self):
1053
        # TODO: test file contents and permissions are preserved. This test was
1054
        # added just to ensure that quoting was handled correctly.
1055
        # -- David Allouche 2006-08-11
1056
        transport = self.get_transport()
1057
        if not transport.listable():
1058
            self.assertRaises(TransportNotPossible,
1059
                              transport.iter_files_recursive)
1060
            return
1061
        if transport.is_readonly():
1062
            return
1063
        self.build_tree(['from/',
1064
                         'from/dir/',
1065
                         'from/dir/foo',
1066
                         'from/dir/bar',
1067
                         'from/dir/b%25z', # make sure quoting is correct
1068
                         'from/bar'],
1069
                        transport=transport)
1070
        transport.copy_tree('from', 'to')
1071
        paths = set(transport.iter_files_recursive())
1072
        self.assertEqual(paths,
1073
                    set(['from/dir/foo',
1074
                         'from/dir/bar',
1075
                         'from/dir/b%2525z',
1076
                         'from/bar',
1077
                         'to/dir/foo',
1078
                         'to/dir/bar',
1079
                         'to/dir/b%2525z',
1080
                         'to/bar',]))
1185.85.76 by John Arbash Meinel
Adding an InvalidURL so transports can report they expect utf-8 quoted paths. Updated tests
1081
1082
    def test_unicode_paths(self):
1685.1.57 by Martin Pool
[broken] Skip unicode blackbox tests if not supported by filesystem
1083
        """Test that we can read/write files with Unicode names."""
1185.85.76 by John Arbash Meinel
Adding an InvalidURL so transports can report they expect utf-8 quoted paths. Updated tests
1084
        t = self.get_transport()
1085
1711.7.36 by John Arbash Meinel
Use different filenames to avoid path collisions on win32 w/ FAT32
1086
        # With FAT32 and certain encodings on win32
1087
        # '\xe5' and '\xe4' actually map to the same file
1088
        # adding a suffix kicks in the 'preserving but insensitive'
1089
        # route, and maintains the right files
1090
        files = [u'\xe5.1', # a w/ circle iso-8859-1
1091
                 u'\xe4.2', # a w/ dots iso-8859-1
1185.85.76 by John Arbash Meinel
Adding an InvalidURL so transports can report they expect utf-8 quoted paths. Updated tests
1092
                 u'\u017d', # Z with umlat iso-8859-2
1093
                 u'\u062c', # Arabic j
1094
                 u'\u0410', # Russian A
1095
                 u'\u65e5', # Kanji person
1096
                ]
1097
1685.1.72 by Wouter van Heyst
StubSFTPServer should use bytestreams rather than unicode
1098
        try:
1711.4.13 by John Arbash Meinel
Use line_endings='binary' for win32
1099
            self.build_tree(files, transport=t, line_endings='binary')
1685.1.72 by Wouter van Heyst
StubSFTPServer should use bytestreams rather than unicode
1100
        except UnicodeError:
1101
            raise TestSkipped("cannot handle unicode paths in current encoding")
1185.85.76 by John Arbash Meinel
Adding an InvalidURL so transports can report they expect utf-8 quoted paths. Updated tests
1102
1103
        # A plain unicode string is not a valid url
1104
        for fname in files:
1105
            self.assertRaises(InvalidURL, t.get, fname)
1106
1107
        for fname in files:
1108
            fname_utf8 = fname.encode('utf-8')
1109
            contents = 'contents of %s\n' % (fname_utf8,)
1685.1.45 by John Arbash Meinel
Moved url functions into bzrlib.urlutils
1110
            self.check_transport_contents(contents, t, urlutils.escape(fname))
1185.85.76 by John Arbash Meinel
Adding an InvalidURL so transports can report they expect utf-8 quoted paths. Updated tests
1111
1534.4.26 by Robert Collins
Move working tree initialisation out from Branch.initialize, deprecated Branch.initialize to Branch.create.
1112
    def test_connect_twice_is_same_content(self):
1113
        # check that our server (whatever it is) is accessable reliably
1114
        # via get_transport and multiple connections share content.
1115
        transport = self.get_transport()
1116
        if transport.is_readonly():
1117
            return
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
1118
        transport.put_bytes('foo', 'bar')
1534.4.26 by Robert Collins
Move working tree initialisation out from Branch.initialize, deprecated Branch.initialize to Branch.create.
1119
        transport2 = self.get_transport()
1120
        self.check_transport_contents('bar', transport2, 'foo')
1121
        # its base should be usable.
1122
        transport2 = bzrlib.transport.get_transport(transport.base)
1123
        self.check_transport_contents('bar', transport2, 'foo')
1124
1125
        # now opening at a relative url should give use a sane result:
1126
        transport.mkdir('newdir')
1127
        transport2 = bzrlib.transport.get_transport(transport.base + "newdir")
1128
        transport2 = transport2.clone('..')
1129
        self.check_transport_contents('bar', transport2, 'foo')
1130
1131
    def test_lock_write(self):
1132
        transport = self.get_transport()
1133
        if transport.is_readonly():
1134
            self.assertRaises(TransportNotPossible, transport.lock_write, 'foo')
1135
            return
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
1136
        transport.put_bytes('lock', '')
1534.4.26 by Robert Collins
Move working tree initialisation out from Branch.initialize, deprecated Branch.initialize to Branch.create.
1137
        lock = transport.lock_write('lock')
1138
        # TODO make this consistent on all platforms:
1139
        # self.assertRaises(LockError, transport.lock_write, 'lock')
1140
        lock.unlock()
1141
1142
    def test_lock_read(self):
1143
        transport = self.get_transport()
1144
        if transport.is_readonly():
1145
            file('lock', 'w').close()
1146
        else:
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
1147
            transport.put_bytes('lock', '')
1534.4.26 by Robert Collins
Move working tree initialisation out from Branch.initialize, deprecated Branch.initialize to Branch.create.
1148
        lock = transport.lock_read('lock')
1149
        # TODO make this consistent on all platforms:
1150
        # self.assertRaises(LockError, transport.lock_read, 'lock')
1151
        lock.unlock()
1185.85.80 by John Arbash Meinel
[merge] jam-integration 1527, including branch-formats, help text, misc bug fixes.
1152
1594.2.5 by Robert Collins
Readv patch from Johan Rydberg giving knits partial download support.
1153
    def test_readv(self):
1154
        transport = self.get_transport()
1155
        if transport.is_readonly():
1156
            file('a', 'w').write('0123456789')
1157
        else:
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
1158
            transport.put_bytes('a', '0123456789')
1185.85.76 by John Arbash Meinel
Adding an InvalidURL so transports can report they expect utf-8 quoted paths. Updated tests
1159
1594.2.17 by Robert Collins
Better readv coalescing, now with test, and progress during knit index reading.
1160
        d = list(transport.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
1594.2.5 by Robert Collins
Readv patch from Johan Rydberg giving knits partial download support.
1161
        self.assertEqual(d[0], (0, '0'))
1594.2.17 by Robert Collins
Better readv coalescing, now with test, and progress during knit index reading.
1162
        self.assertEqual(d[1], (1, '1'))
1163
        self.assertEqual(d[2], (3, '34'))
1164
        self.assertEqual(d[3], (9, '9'))
1786.1.8 by John Arbash Meinel
[merge] Johan Rydberg test updates
1165
1864.5.2 by John Arbash Meinel
always read in sorted order, and return in requested order, but only cache what is currently out of order
1166
    def test_readv_out_of_order(self):
1167
        transport = self.get_transport()
1168
        if transport.is_readonly():
1169
            file('a', 'w').write('0123456789')
1170
        else:
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
1171
            transport.put_bytes('a', '01234567890')
1864.5.2 by John Arbash Meinel
always read in sorted order, and return in requested order, but only cache what is currently out of order
1172
1173
        d = list(transport.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
1174
        self.assertEqual(d[0], (1, '1'))
1175
        self.assertEqual(d[1], (9, '9'))
1176
        self.assertEqual(d[2], (0, '0'))
1177
        self.assertEqual(d[3], (3, '34'))