~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

  • Committer: Aaron Bentley
  • Date: 2006-02-22 14:39:42 UTC
  • mto: (2027.1.2 revert-subpath-56549)
  • mto: This revision was merged to the branch mainline in revision 1570.
  • Revision ID: abentley@panoramicfeedback.com-20060222143942-ae72299f2de66767
Fixed build_tree with symlinks

Show diffs side-by-side

added added

removed removed

Lines of Context:
22
22
 
23
23
from bzrlib.errors import (NoSuchFile, FileExists,
24
24
                           TransportNotPossible, ConnectionError)
25
 
from bzrlib.tests import TestCase, TestCaseInTempDir
26
 
from bzrlib.tests.HTTPTestUtil import TestCaseWithWebserver
27
 
from bzrlib.transport import memory, urlescape
28
 
from bzrlib.osutils import pathjoin
29
 
 
30
 
 
31
 
def _append(fn, txt):
32
 
    """Append the given text (file-like object) to the supplied filename."""
33
 
    f = open(fn, 'ab')
34
 
    f.write(txt)
35
 
    f.flush()
36
 
    f.close()
37
 
    del f
38
 
 
39
 
 
40
 
if sys.platform != 'win32':
41
 
    def check_mode(test, path, mode):
42
 
        """Check that a particular path has the correct mode."""
43
 
        actual_mode = stat.S_IMODE(os.stat(path).st_mode)
44
 
        test.assertEqual(mode, actual_mode,
45
 
            'mode of %r incorrect (%o != %o)' % (path, mode, actual_mode))
46
 
else:
47
 
    def check_mode(test, path, mode):
48
 
        """On win32 chmod doesn't have any effect, 
49
 
        so don't actually check anything
50
 
        """
51
 
        return
 
25
from bzrlib.tests import TestCase
 
26
from bzrlib.transport import (_get_protocol_handlers,
 
27
                              _get_transport_modules,
 
28
                              get_transport,
 
29
                              register_lazy_transport,
 
30
                              _set_protocol_handlers,
 
31
                              urlescape,
 
32
                              )
52
33
 
53
34
 
54
35
class TestTransport(TestCase):
57
38
    def test_urlescape(self):
58
39
        self.assertEqual('%25', urlescape('%'))
59
40
 
60
 
 
61
 
class TestTransportMixIn(object):
62
 
    """Subclass this, and it will provide a series of tests for a Transport.
63
 
    It assumes that the Transport object is connected to the 
64
 
    current working directory.  So that whatever is done 
65
 
    through the transport, should show up in the working 
66
 
    directory, and vice-versa.
67
 
 
68
 
    This also tests to make sure that the functions work with both
69
 
    generators and lists (assuming iter(list) is effectively a generator)
70
 
    """
71
 
    readonly = False
72
 
    def get_transport(self):
73
 
        """Children should override this to return the Transport object.
74
 
        """
75
 
        raise NotImplementedError
76
 
 
77
 
    def assertListRaises(self, excClass, func, *args, **kwargs):
78
 
        """Many transport functions can return generators this makes sure
79
 
        to wrap them in a list() call to make sure the whole generator
80
 
        is run, and that the proper exception is raised.
81
 
        """
82
 
        try:
83
 
            list(func(*args, **kwargs))
84
 
        except excClass:
85
 
            return
86
 
        else:
87
 
            if hasattr(excClass,'__name__'): excName = excClass.__name__
88
 
            else: excName = str(excClass)
89
 
            raise self.failureException, "%s not raised" % excName
90
 
 
91
 
    def test_has(self):
92
 
        t = self.get_transport()
93
 
 
94
 
        files = ['a', 'b', 'e', 'g', '%']
95
 
        self.build_tree(files)
96
 
        self.assertEqual(True, t.has('a'))
97
 
        self.assertEqual(False, t.has('c'))
98
 
        self.assertEqual(True, t.has(urlescape('%')))
99
 
        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
100
 
                [True, True, False, False, True, False, True, False])
101
 
        self.assertEqual(True, t.has_any(['a', 'b', 'c']))
102
 
        self.assertEqual(False, t.has_any(['c', 'd', 'f', urlescape('%%')]))
103
 
        self.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']))),
104
 
                [True, True, False, False, True, False, True, False])
105
 
        self.assertEqual(False, t.has_any(['c', 'c', 'c']))
106
 
        self.assertEqual(True, t.has_any(['b', 'b', 'b']))
107
 
 
108
 
    def test_get(self):
109
 
        t = self.get_transport()
110
 
 
111
 
        files = ['a', 'b', 'e', 'g']
112
 
        self.build_tree(files)
113
 
        self.assertEqual(open('a', 'rb').read(), t.get('a').read())
114
 
        content_f = t.get_multi(files)
115
 
        for path,f in zip(files, content_f):
116
 
            self.assertEqual(f.read(), open(path, 'rb').read())
117
 
 
118
 
        content_f = t.get_multi(iter(files))
119
 
        for path,f in zip(files, content_f):
120
 
            self.assertEqual(f.read(), open(path, 'rb').read())
121
 
 
122
 
        self.assertRaises(NoSuchFile, t.get, 'c')
123
 
        self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])
124
 
        self.assertListRaises(NoSuchFile, t.get_multi, iter(['a', 'b', 'c']))
125
 
 
126
 
    def test_put(self):
127
 
        t = self.get_transport()
128
 
 
129
 
        # TODO: jam 20051215 No need to do anything if the test is readonly
130
 
        #                    origininally it was thought that it would give
131
 
        #                    more of a workout to readonly tests. By now the
132
 
        #                    suite is probably thorough enough without testing
133
 
        #                    readonly protocols in write sections
134
 
        #                    The only thing that needs to be tested is that the
135
 
        #                    right error is raised
136
 
 
137
 
        if self.readonly:
138
 
            self.assertRaises(TransportNotPossible,
139
 
                    t.put, 'a', 'some text for a\n')
140
 
            open('a', 'wb').write('some text for a\n')
141
 
        else:
142
 
            t.put('a', 'some text for a\n')
143
 
        self.assert_(os.path.exists('a'))
144
 
        self.check_file_contents('a', 'some text for a\n')
145
 
        self.assertEqual(t.get('a').read(), 'some text for a\n')
146
 
        # Make sure 'has' is updated
147
 
        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
148
 
                [True, False, False, False, False])
149
 
        if self.readonly:
150
 
            self.assertRaises(TransportNotPossible,
151
 
                    t.put_multi,
152
 
                    [('a', 'new\ncontents for\na\n'),
153
 
                        ('d', 'contents\nfor d\n')])
154
 
            open('a', 'wb').write('new\ncontents for\na\n')
155
 
            open('d', 'wb').write('contents\nfor d\n')
156
 
        else:
157
 
            # Put also replaces contents
158
 
            self.assertEqual(t.put_multi([('a', 'new\ncontents for\na\n'),
159
 
                                          ('d', 'contents\nfor d\n')]),
160
 
                             2)
161
 
        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
162
 
                [True, False, False, True, False])
163
 
        self.check_file_contents('a', 'new\ncontents for\na\n')
164
 
        self.check_file_contents('d', 'contents\nfor d\n')
165
 
 
166
 
        if self.readonly:
167
 
            self.assertRaises(TransportNotPossible,
168
 
                t.put_multi, iter([('a', 'diff\ncontents for\na\n'),
169
 
                                  ('d', 'another contents\nfor d\n')]))
170
 
            open('a', 'wb').write('diff\ncontents for\na\n')
171
 
            open('d', 'wb').write('another contents\nfor d\n')
172
 
        else:
173
 
            self.assertEqual(
174
 
                t.put_multi(iter([('a', 'diff\ncontents for\na\n'),
175
 
                                  ('d', 'another contents\nfor d\n')]))
176
 
                             , 2)
177
 
        self.check_file_contents('a', 'diff\ncontents for\na\n')
178
 
        self.check_file_contents('d', 'another contents\nfor d\n')
179
 
 
180
 
        if self.readonly:
181
 
            self.assertRaises(TransportNotPossible,
182
 
                    t.put, 'path/doesnt/exist/c', 'contents')
183
 
        else:
184
 
            self.assertRaises(NoSuchFile,
185
 
                    t.put, 'path/doesnt/exist/c', 'contents')
186
 
 
187
 
        if not self.readonly:
188
 
            t.put('mode644', 'test text\n', mode=0644)
189
 
            check_mode(self, 'mode644', 0644)
190
 
 
191
 
            t.put('mode666', 'test text\n', mode=0666)
192
 
            check_mode(self, 'mode666', 0666)
193
 
 
194
 
            t.put('mode600', 'test text\n', mode=0600)
195
 
            check_mode(self, 'mode600', 0600)
196
 
 
197
 
            # Yes, you can put a file such that it becomes readonly
198
 
            t.put('mode400', 'test text\n', mode=0400)
199
 
            check_mode(self, 'mode400', 0400)
200
 
 
201
 
            t.put_multi([('mmode644', 'text\n')], mode=0644)
202
 
            check_mode(self, 'mmode644', 0644)
203
 
 
204
 
        # TODO: jam 20051215 test put_multi with a mode. I didn't bother because
205
 
        #                    it seems most people don't like the _multi functions
206
 
 
207
 
    def test_put_file(self):
208
 
        t = self.get_transport()
209
 
 
210
 
        # Test that StringIO can be used as a file-like object with put
211
 
        f1 = StringIO('this is a string\nand some more stuff\n')
212
 
        if self.readonly:
213
 
            open('f1', 'wb').write(f1.read())
214
 
        else:
215
 
            t.put('f1', f1)
216
 
 
217
 
        del f1
218
 
 
219
 
        self.check_file_contents('f1', 
220
 
                'this is a string\nand some more stuff\n')
221
 
 
222
 
        f2 = StringIO('here is some text\nand a bit more\n')
223
 
        f3 = StringIO('some text for the\nthird file created\n')
224
 
 
225
 
        if self.readonly:
226
 
            open('f2', 'wb').write(f2.read())
227
 
            open('f3', 'wb').write(f3.read())
228
 
        else:
229
 
            t.put_multi([('f2', f2), ('f3', f3)])
230
 
 
231
 
        del f2, f3
232
 
 
233
 
        self.check_file_contents('f2', 'here is some text\nand a bit more\n')
234
 
        self.check_file_contents('f3', 'some text for the\nthird file created\n')
235
 
 
236
 
        # Test that an actual file object can be used with put
237
 
        f4 = open('f1', 'rb')
238
 
        if self.readonly:
239
 
            open('f4', 'wb').write(f4.read())
240
 
        else:
241
 
            t.put('f4', f4)
242
 
 
243
 
        del f4
244
 
 
245
 
        self.check_file_contents('f4', 
246
 
                'this is a string\nand some more stuff\n')
247
 
 
248
 
        f5 = open('f2', 'rb')
249
 
        f6 = open('f3', 'rb')
250
 
        if self.readonly:
251
 
            open('f5', 'wb').write(f5.read())
252
 
            open('f6', 'wb').write(f6.read())
253
 
        else:
254
 
            t.put_multi([('f5', f5), ('f6', f6)])
255
 
 
256
 
        del f5, f6
257
 
 
258
 
        self.check_file_contents('f5', 'here is some text\nand a bit more\n')
259
 
        self.check_file_contents('f6', 'some text for the\nthird file created\n')
260
 
 
261
 
        if not self.readonly:
262
 
            sio = StringIO('test text\n')
263
 
            t.put('mode644', sio, mode=0644)
264
 
            check_mode(self, 'mode644', 0644)
265
 
 
266
 
            a = open('mode644', 'rb')
267
 
            t.put('mode666', a, mode=0666)
268
 
            check_mode(self, 'mode666', 0666)
269
 
 
270
 
            a = open('mode644', 'rb')
271
 
            t.put('mode600', a, mode=0600)
272
 
            check_mode(self, 'mode600', 0600)
273
 
 
274
 
            # Yes, you can put a file such that it becomes readonly
275
 
            a = open('mode644', 'rb')
276
 
            t.put('mode400', a, mode=0400)
277
 
            check_mode(self, 'mode400', 0400)
278
 
 
279
 
    def test_mkdir(self):
280
 
        t = self.get_transport()
281
 
 
282
 
        # Test mkdir
283
 
        os.mkdir('dir_a')
284
 
        self.assertEqual(t.has('dir_a'), True)
285
 
        self.assertEqual(t.has('dir_b'), False)
286
 
 
287
 
        if self.readonly:
288
 
            self.assertRaises(TransportNotPossible,
289
 
                    t.mkdir, 'dir_b')
290
 
            os.mkdir('dir_b')
291
 
        else:
292
 
            t.mkdir('dir_b')
293
 
        self.assertEqual(t.has('dir_b'), True)
294
 
        self.assert_(os.path.isdir('dir_b'))
295
 
 
296
 
        if self.readonly:
297
 
            self.assertRaises(TransportNotPossible,
298
 
                    t.mkdir_multi, ['dir_c', 'dir_d'])
299
 
            os.mkdir('dir_c')
300
 
            os.mkdir('dir_d')
301
 
        else:
302
 
            t.mkdir_multi(['dir_c', 'dir_d'])
303
 
 
304
 
        if self.readonly:
305
 
            self.assertRaises(TransportNotPossible,
306
 
                    t.mkdir_multi, iter(['dir_e', 'dir_f']))
307
 
            os.mkdir('dir_e')
308
 
            os.mkdir('dir_f')
309
 
        else:
310
 
            t.mkdir_multi(iter(['dir_e', 'dir_f']))
311
 
        self.assertEqual(list(t.has_multi(
312
 
            ['dir_a', 'dir_b', 'dir_c', 'dir_q',
313
 
             'dir_d', 'dir_e', 'dir_f', 'dir_b'])),
314
 
            [True, True, True, False,
315
 
             True, True, True, True])
316
 
        for d in ['dir_a', 'dir_b', 'dir_c', 'dir_d', 'dir_e', 'dir_f']:
317
 
            self.assert_(os.path.isdir(d))
318
 
 
319
 
        if not self.readonly:
320
 
            self.assertRaises(NoSuchFile, t.mkdir, 'path/doesnt/exist')
321
 
            self.assertRaises(FileExists, t.mkdir, 'dir_a') # Creating a directory again should fail
322
 
 
323
 
        # Make sure the transport recognizes when a
324
 
        # directory is created by other means
325
 
        # Caching Transports will fail, because dir_e was already seen not
326
 
        # to exist. So instead, we will search for a new directory
327
 
        #os.mkdir('dir_e')
328
 
        #if not self.readonly:
329
 
        #    self.assertRaises(FileExists, t.mkdir, 'dir_e')
330
 
 
331
 
        os.mkdir('dir_g')
332
 
        if not self.readonly:
333
 
            self.assertRaises(FileExists, t.mkdir, 'dir_g')
334
 
 
335
 
        # Test get/put in sub-directories
336
 
        if self.readonly:
337
 
            open('dir_a/a', 'wb').write('contents of dir_a/a')
338
 
            open('dir_b/b', 'wb').write('contents of dir_b/b')
339
 
        else:
340
 
            self.assertEqual(
341
 
                t.put_multi([('dir_a/a', 'contents of dir_a/a'),
342
 
                             ('dir_b/b', 'contents of dir_b/b')])
343
 
                          , 2)
344
 
        for f in ('dir_a/a', 'dir_b/b'):
345
 
            self.assertEqual(t.get(f).read(), open(f, 'rb').read())
346
 
 
347
 
        if not self.readonly:
348
 
            # Test mkdir with a mode
349
 
            t.mkdir('dmode755', mode=0755)
350
 
            check_mode(self, 'dmode755', 0755)
351
 
 
352
 
            t.mkdir('dmode555', mode=0555)
353
 
            check_mode(self, 'dmode555', 0555)
354
 
 
355
 
            t.mkdir('dmode777', mode=0777)
356
 
            check_mode(self, 'dmode777', 0777)
357
 
 
358
 
            t.mkdir('dmode700', mode=0700)
359
 
            check_mode(self, 'dmode700', 0700)
360
 
 
361
 
            # TODO: jam 20051215 test mkdir_multi with a mode
362
 
            t.mkdir_multi(['mdmode755'], mode=0755)
363
 
            check_mode(self, 'mdmode755', 0755)
364
 
 
365
 
 
366
 
    def test_copy_to(self):
367
 
        import tempfile
368
 
        from bzrlib.transport.local import LocalTransport
369
 
 
370
 
        t = self.get_transport()
371
 
 
372
 
        files = ['a', 'b', 'c', 'd']
373
 
        self.build_tree(files)
374
 
 
375
 
        def get_temp_local():
376
 
            dtmp = tempfile.mkdtemp(dir=u'.', prefix='test-transport-')
377
 
            dtmp_base = os.path.basename(dtmp)
378
 
            return dtmp_base, LocalTransport(dtmp)
379
 
        dtmp_base, local_t = get_temp_local()
380
 
 
381
 
        t.copy_to(files, local_t)
382
 
        for f in files:
383
 
            self.assertEquals(open(f, 'rb').read(),
384
 
                    open(pathjoin(dtmp_base, f), 'rb').read())
385
 
 
386
 
        # Test that copying into a missing directory raises
387
 
        # NoSuchFile
388
 
        os.mkdir('e')
389
 
        open('e/f', 'wb').write('contents of e')
390
 
        self.assertRaises(NoSuchFile, t.copy_to, ['e/f'], local_t)
391
 
 
392
 
        os.mkdir(pathjoin(dtmp_base, 'e'))
393
 
        t.copy_to(['e/f'], local_t)
394
 
 
395
 
        del dtmp_base, local_t
396
 
 
397
 
        dtmp_base, local_t = get_temp_local()
398
 
 
399
 
        files = ['a', 'b', 'c', 'd']
400
 
        t.copy_to(iter(files), local_t)
401
 
        for f in files:
402
 
            self.assertEquals(open(f, 'rb').read(),
403
 
                    open(pathjoin(dtmp_base, f), 'rb').read())
404
 
 
405
 
        del dtmp_base, local_t
406
 
 
407
 
        for mode in (0666, 0644, 0600, 0400):
408
 
            dtmp_base, local_t = get_temp_local()
409
 
            t.copy_to(files, local_t, mode=mode)
410
 
            for f in files:
411
 
                check_mode(self, os.path.join(dtmp_base, f), mode)
412
 
 
413
 
    def test_append(self):
414
 
        t = self.get_transport()
415
 
 
416
 
        if self.readonly:
417
 
            open('a', 'wb').write('diff\ncontents for\na\n')
418
 
            open('b', 'wb').write('contents\nfor b\n')
419
 
        else:
420
 
            t.put_multi([
421
 
                    ('a', 'diff\ncontents for\na\n'),
422
 
                    ('b', 'contents\nfor b\n')
423
 
                    ])
424
 
 
425
 
        if self.readonly:
426
 
            self.assertRaises(TransportNotPossible,
427
 
                    t.append, 'a', 'add\nsome\nmore\ncontents\n')
428
 
            _append('a', 'add\nsome\nmore\ncontents\n')
429
 
        else:
430
 
            t.append('a', 'add\nsome\nmore\ncontents\n')
431
 
 
432
 
        self.check_file_contents('a', 
433
 
            'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n')
434
 
 
435
 
        if self.readonly:
436
 
            self.assertRaises(TransportNotPossible,
437
 
                    t.append_multi,
438
 
                        [('a', 'and\nthen\nsome\nmore\n'),
439
 
                         ('b', 'some\nmore\nfor\nb\n')])
440
 
            _append('a', 'and\nthen\nsome\nmore\n')
441
 
            _append('b', 'some\nmore\nfor\nb\n')
442
 
        else:
443
 
            t.append_multi([('a', 'and\nthen\nsome\nmore\n'),
444
 
                    ('b', 'some\nmore\nfor\nb\n')])
445
 
        self.check_file_contents('a', 
446
 
            'diff\ncontents for\na\n'
447
 
            'add\nsome\nmore\ncontents\n'
448
 
            'and\nthen\nsome\nmore\n')
449
 
        self.check_file_contents('b', 
450
 
                'contents\nfor b\n'
451
 
                'some\nmore\nfor\nb\n')
452
 
 
453
 
        if self.readonly:
454
 
            _append('a', 'a little bit more\n')
455
 
            _append('b', 'from an iterator\n')
456
 
        else:
457
 
            t.append_multi(iter([('a', 'a little bit more\n'),
458
 
                    ('b', 'from an iterator\n')]))
459
 
        self.check_file_contents('a', 
460
 
            'diff\ncontents for\na\n'
461
 
            'add\nsome\nmore\ncontents\n'
462
 
            'and\nthen\nsome\nmore\n'
463
 
            'a little bit more\n')
464
 
        self.check_file_contents('b', 
465
 
                'contents\nfor b\n'
466
 
                'some\nmore\nfor\nb\n'
467
 
                'from an iterator\n')
468
 
 
469
 
        if self.readonly:
470
 
            _append('c', 'some text\nfor a missing file\n')
471
 
            _append('a', 'some text in a\n')
472
 
            _append('d', 'missing file r\n')
473
 
        else:
474
 
            t.append('c', 'some text\nfor a missing file\n')
475
 
            t.append_multi([('a', 'some text in a\n'),
476
 
                            ('d', 'missing file r\n')])
477
 
        self.check_file_contents('a', 
478
 
            'diff\ncontents for\na\n'
479
 
            'add\nsome\nmore\ncontents\n'
480
 
            'and\nthen\nsome\nmore\n'
481
 
            'a little bit more\n'
482
 
            'some text in a\n')
483
 
        self.check_file_contents('c', 'some text\nfor a missing file\n')
484
 
        self.check_file_contents('d', 'missing file r\n')
485
 
 
486
 
    def test_append_file(self):
487
 
        t = self.get_transport()
488
 
 
489
 
        contents = [
490
 
            ('f1', 'this is a string\nand some more stuff\n'),
491
 
            ('f2', 'here is some text\nand a bit more\n'),
492
 
            ('f3', 'some text for the\nthird file created\n'),
493
 
            ('f4', 'this is a string\nand some more stuff\n'),
494
 
            ('f5', 'here is some text\nand a bit more\n'),
495
 
            ('f6', 'some text for the\nthird file created\n')
496
 
        ]
497
 
        
498
 
        if self.readonly:
499
 
            for f, val in contents:
500
 
                open(f, 'wb').write(val)
501
 
        else:
502
 
            t.put_multi(contents)
503
 
 
504
 
        a1 = StringIO('appending to\none\n')
505
 
        if self.readonly:
506
 
            _append('f1', a1.read())
507
 
        else:
508
 
            t.append('f1', a1)
509
 
 
510
 
        del a1
511
 
 
512
 
        self.check_file_contents('f1', 
513
 
                'this is a string\nand some more stuff\n'
514
 
                'appending to\none\n')
515
 
 
516
 
        a2 = StringIO('adding more\ntext to two\n')
517
 
        a3 = StringIO('some garbage\nto put in three\n')
518
 
 
519
 
        if self.readonly:
520
 
            _append('f2', a2.read())
521
 
            _append('f3', a3.read())
522
 
        else:
523
 
            t.append_multi([('f2', a2), ('f3', a3)])
524
 
 
525
 
        del a2, a3
526
 
 
527
 
        self.check_file_contents('f2',
528
 
                'here is some text\nand a bit more\n'
529
 
                'adding more\ntext to two\n')
530
 
        self.check_file_contents('f3', 
531
 
                'some text for the\nthird file created\n'
532
 
                'some garbage\nto put in three\n')
533
 
 
534
 
        # Test that an actual file object can be used with put
535
 
        a4 = open('f1', 'rb')
536
 
        if self.readonly:
537
 
            _append('f4', a4.read())
538
 
        else:
539
 
            t.append('f4', a4)
540
 
 
541
 
        del a4
542
 
 
543
 
        self.check_file_contents('f4', 
544
 
                'this is a string\nand some more stuff\n'
545
 
                'this is a string\nand some more stuff\n'
546
 
                'appending to\none\n')
547
 
 
548
 
        a5 = open('f2', 'rb')
549
 
        a6 = open('f3', 'rb')
550
 
        if self.readonly:
551
 
            _append('f5', a5.read())
552
 
            _append('f6', a6.read())
553
 
        else:
554
 
            t.append_multi([('f5', a5), ('f6', a6)])
555
 
 
556
 
        del a5, a6
557
 
 
558
 
        self.check_file_contents('f5',
559
 
                'here is some text\nand a bit more\n'
560
 
                'here is some text\nand a bit more\n'
561
 
                'adding more\ntext to two\n')
562
 
        self.check_file_contents('f6',
563
 
                'some text for the\nthird file created\n'
564
 
                'some text for the\nthird file created\n'
565
 
                'some garbage\nto put in three\n')
566
 
 
567
 
        a5 = open('f2', 'rb')
568
 
        a6 = open('f2', 'rb')
569
 
        a7 = open('f3', 'rb')
570
 
        if self.readonly:
571
 
            _append('c', a5.read())
572
 
            _append('a', a6.read())
573
 
            _append('d', a7.read())
574
 
        else:
575
 
            t.append('c', a5)
576
 
            t.append_multi([('a', a6), ('d', a7)])
577
 
        del a5, a6, a7
578
 
        self.check_file_contents('c', open('f2', 'rb').read())
579
 
        self.check_file_contents('d', open('f3', 'rb').read())
580
 
 
581
 
 
582
 
    def test_delete(self):
583
 
        # TODO: Test Transport.delete
584
 
        t = self.get_transport()
585
 
 
586
 
        # Not much to do with a readonly transport
587
 
        if self.readonly:
588
 
            return
589
 
 
590
 
        open('a', 'wb').write('a little bit of text\n')
591
 
        self.failUnless(t.has('a'))
592
 
        self.failUnlessExists('a')
593
 
        t.delete('a')
594
 
        self.failIf(os.path.lexists('a'))
595
 
 
596
 
        self.assertRaises(NoSuchFile, t.delete, 'a')
597
 
 
598
 
        open('a', 'wb').write('a text\n')
599
 
        open('b', 'wb').write('b text\n')
600
 
        open('c', 'wb').write('c text\n')
601
 
        self.assertEqual([True, True, True],
602
 
                list(t.has_multi(['a', 'b', 'c'])))
603
 
        t.delete_multi(['a', 'c'])
604
 
        self.assertEqual([False, True, False],
605
 
                list(t.has_multi(['a', 'b', 'c'])))
606
 
        self.failIf(os.path.lexists('a'))
607
 
        self.failUnlessExists('b')
608
 
        self.failIf(os.path.lexists('c'))
609
 
 
610
 
        self.assertRaises(NoSuchFile,
611
 
                t.delete_multi, ['a', 'b', 'c'])
612
 
 
613
 
        self.assertRaises(NoSuchFile,
614
 
                t.delete_multi, iter(['a', 'b', 'c']))
615
 
 
616
 
        open('a', 'wb').write('another a text\n')
617
 
        open('c', 'wb').write('another c text\n')
618
 
        t.delete_multi(iter(['a', 'b', 'c']))
619
 
 
620
 
        # We should have deleted everything
621
 
        # SftpServer creates control files in the
622
 
        # working directory, so we can just do a
623
 
        # plain "listdir".
624
 
        # self.assertEqual([], os.listdir('.'))
625
 
 
626
 
    def test_move(self):
627
 
        t = self.get_transport()
628
 
 
629
 
        if self.readonly:
630
 
            return
631
 
 
632
 
        # TODO: I would like to use os.listdir() to
633
 
        # make sure there are no extra files, but SftpServer
634
 
        # creates control files in the working directory
635
 
        # perhaps all of this could be done in a subdirectory
636
 
 
637
 
        open('a', 'wb').write('a first file\n')
638
 
        self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
639
 
 
640
 
        t.move('a', 'b')
641
 
        self.failUnlessExists('b')
642
 
        self.failIf(os.path.lexists('a'))
643
 
 
644
 
        self.check_file_contents('b', 'a first file\n')
645
 
        self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
646
 
 
647
 
        # Overwrite a file
648
 
        open('c', 'wb').write('c this file\n')
649
 
        t.move('c', 'b')
650
 
        self.failIf(os.path.lexists('c'))
651
 
        self.check_file_contents('b', 'c this file\n')
652
 
 
653
 
        # TODO: Try to write a test for atomicity
654
 
        # TODO: Test moving into a non-existant subdirectory
655
 
        # TODO: Test Transport.move_multi
656
 
 
657
 
    def test_copy(self):
658
 
        t = self.get_transport()
659
 
 
660
 
        if self.readonly:
661
 
            return
662
 
 
663
 
        open('a', 'wb').write('a file\n')
664
 
        t.copy('a', 'b')
665
 
        self.check_file_contents('b', 'a file\n')
666
 
 
667
 
        self.assertRaises(NoSuchFile, t.copy, 'c', 'd')
668
 
        os.mkdir('c')
669
 
        # What should the assert be if you try to copy a
670
 
        # file over a directory?
671
 
        #self.assertRaises(Something, t.copy, 'a', 'c')
672
 
        open('d', 'wb').write('text in d\n')
673
 
        t.copy('d', 'b')
674
 
        self.check_file_contents('b', 'text in d\n')
675
 
 
676
 
        # TODO: test copy_multi
677
 
 
678
 
    def test_connection_error(self):
679
 
        """ConnectionError is raised when connection is impossible"""
680
 
        if not hasattr(self, "get_bogus_transport"):
681
 
            return
682
 
        t = self.get_bogus_transport()
683
 
        try:
684
 
            t.get('.bzr/branch')
685
 
        except (ConnectionError, NoSuchFile), e:
686
 
            pass
687
 
        except (Exception), e:
688
 
            self.failIf(True, 'Wrong exception thrown: %s' % e)
689
 
        else:
690
 
            self.failIf(True, 'Did not get the expected exception.')
691
 
 
692
 
    def test_stat(self):
693
 
        # TODO: Test stat, just try once, and if it throws, stop testing
694
 
        from stat import S_ISDIR, S_ISREG
695
 
 
696
 
        t = self.get_transport()
697
 
 
698
 
        try:
699
 
            st = t.stat('.')
700
 
        except TransportNotPossible, e:
701
 
            # This transport cannot stat
702
 
            return
703
 
 
704
 
        paths = ['a', 'b/', 'b/c', 'b/d/', 'b/d/e']
705
 
        self.build_tree(paths)
706
 
 
707
 
        local_stats = []
708
 
 
709
 
        for p in paths:
710
 
            st = t.stat(p)
711
 
            local_st = os.stat(p)
712
 
            if p.endswith('/'):
713
 
                self.failUnless(S_ISDIR(st.st_mode))
714
 
            else:
715
 
                self.failUnless(S_ISREG(st.st_mode))
716
 
            self.assertEqual(local_st.st_size, st.st_size)
717
 
            self.assertEqual(local_st.st_mode, st.st_mode)
718
 
            local_stats.append(local_st)
719
 
 
720
 
        remote_stats = list(t.stat_multi(paths))
721
 
        remote_iter_stats = list(t.stat_multi(iter(paths)))
722
 
 
723
 
        for local, remote, remote_iter in \
724
 
            zip(local_stats, remote_stats, remote_iter_stats):
725
 
            self.assertEqual(local.st_mode, remote.st_mode)
726
 
            self.assertEqual(local.st_mode, remote_iter.st_mode)
727
 
 
728
 
            self.assertEqual(local.st_size, remote.st_size)
729
 
            self.assertEqual(local.st_size, remote_iter.st_size)
730
 
            # Should we test UID/GID?
731
 
 
732
 
        self.assertRaises(NoSuchFile, t.stat, 'q')
733
 
        self.assertRaises(NoSuchFile, t.stat, 'b/a')
734
 
 
735
 
        self.assertListRaises(NoSuchFile, t.stat_multi, ['a', 'c', 'd'])
736
 
        self.assertListRaises(NoSuchFile, t.stat_multi, iter(['a', 'c', 'd']))
737
 
 
738
 
    def test_list_dir(self):
739
 
        # TODO: Test list_dir, just try once, and if it throws, stop testing
740
 
        t = self.get_transport()
741
 
        
742
 
        if not t.listable():
743
 
            self.assertRaises(TransportNotPossible, t.list_dir, '.')
744
 
            return
745
 
 
746
 
        def sorted_list(d):
747
 
            l = list(t.list_dir(d))
748
 
            l.sort()
749
 
            return l
750
 
 
751
 
        # SftpServer creates control files in the working directory
752
 
        # so lets move down a directory to be safe
753
 
        os.mkdir('wd')
754
 
        os.chdir('wd')
755
 
        t = t.clone('wd')
756
 
 
757
 
        self.assertEqual([], sorted_list(u'.'))
758
 
        self.build_tree(['a', 'b', 'c/', 'c/d', 'c/e'])
759
 
 
760
 
        self.assertEqual([u'a', u'b', u'c'], sorted_list(u'.'))
761
 
        self.assertEqual([u'd', u'e'], sorted_list(u'c'))
762
 
 
763
 
        os.remove('c/d')
764
 
        os.remove('b')
765
 
        self.assertEqual([u'a', u'c'], sorted_list('.'))
766
 
        self.assertEqual([u'e'], sorted_list(u'c'))
767
 
 
768
 
        self.assertListRaises(NoSuchFile, t.list_dir, 'q')
769
 
        self.assertListRaises(NoSuchFile, t.list_dir, 'c/f')
770
 
        self.assertListRaises(NoSuchFile, t.list_dir, 'a')
771
 
 
772
 
    def test_clone(self):
773
 
        # TODO: Test that clone moves up and down the filesystem
774
 
        t1 = self.get_transport()
775
 
 
776
 
        self.build_tree(['a', 'b/', 'b/c'])
777
 
 
778
 
        self.failUnless(t1.has('a'))
779
 
        self.failUnless(t1.has('b/c'))
780
 
        self.failIf(t1.has('c'))
781
 
 
782
 
        t2 = t1.clone('b')
783
 
        self.failUnless(t2.has('c'))
784
 
        self.failIf(t2.has('a'))
785
 
 
786
 
        t3 = t2.clone('..')
787
 
        self.failUnless(t3.has('a'))
788
 
        self.failIf(t3.has('c'))
789
 
 
790
 
        self.failIf(t1.has('b/d'))
791
 
        self.failIf(t2.has('d'))
792
 
        self.failIf(t3.has('b/d'))
793
 
 
794
 
        if self.readonly:
795
 
            open('b/d', 'wb').write('newfile\n')
796
 
        else:
797
 
            t2.put('d', 'newfile\n')
798
 
 
799
 
        self.failUnless(t1.has('b/d'))
800
 
        self.failUnless(t2.has('d'))
801
 
        self.failUnless(t3.has('b/d'))
802
 
 
803
 
        
804
 
class LocalTransportTest(TestCaseInTempDir, TestTransportMixIn):
805
 
    def get_transport(self):
806
 
        from bzrlib.transport.local import LocalTransport
807
 
        return LocalTransport(u'.')
808
 
 
809
 
 
810
 
class HttpTransportTest(TestCaseWithWebserver, TestTransportMixIn):
811
 
 
812
 
    readonly = True
813
 
 
814
 
    def get_transport(self):
815
 
        from bzrlib.transport.http import HttpTransport
816
 
        url = self.get_remote_url(u'.')
817
 
        return HttpTransport(url)
818
 
 
819
 
    def get_bogus_transport(self):
820
 
        from bzrlib.transport.http import HttpTransport
821
 
        return HttpTransport('http://jasldkjsalkdjalksjdkljasd')
822
 
 
823
 
 
824
 
class TestMemoryTransport(TestCase):
825
 
 
826
 
    def test_get_transport(self):
827
 
        memory.MemoryTransport()
828
 
 
829
 
    def test_clone(self):
830
 
        transport = memory.MemoryTransport()
831
 
        self.failUnless(transport.clone() is transport)
832
 
 
833
 
    def test_abspath(self):
834
 
        transport = memory.MemoryTransport()
835
 
        self.assertEqual("in-memory:relpath", transport.abspath('relpath'))
836
 
 
837
 
    def test_relpath(self):
838
 
        transport = memory.MemoryTransport()
839
 
 
840
 
    def test_append_and_get(self):
841
 
        transport = memory.MemoryTransport()
842
 
        transport.append('path', StringIO('content'))
843
 
        self.assertEqual(transport.get('path').read(), 'content')
844
 
        transport.append('path', StringIO('content'))
845
 
        self.assertEqual(transport.get('path').read(), 'contentcontent')
846
 
 
847
 
    def test_put_and_get(self):
848
 
        transport = memory.MemoryTransport()
849
 
        transport.put('path', StringIO('content'))
850
 
        self.assertEqual(transport.get('path').read(), 'content')
851
 
        transport.put('path', StringIO('content'))
852
 
        self.assertEqual(transport.get('path').read(), 'content')
853
 
 
854
 
    def test_append_without_dir_fails(self):
855
 
        transport = memory.MemoryTransport()
856
 
        self.assertRaises(NoSuchFile,
857
 
                          transport.append, 'dir/path', StringIO('content'))
858
 
 
859
 
    def test_put_without_dir_fails(self):
860
 
        transport = memory.MemoryTransport()
861
 
        self.assertRaises(NoSuchFile,
862
 
                          transport.put, 'dir/path', StringIO('content'))
863
 
 
864
 
    def test_get_missing(self):
865
 
        transport = memory.MemoryTransport()
866
 
        self.assertRaises(NoSuchFile, transport.get, 'foo')
867
 
 
868
 
    def test_has_missing(self):
869
 
        transport = memory.MemoryTransport()
870
 
        self.assertEquals(False, transport.has('foo'))
871
 
 
872
 
    def test_has_present(self):
873
 
        transport = memory.MemoryTransport()
874
 
        transport.append('foo', StringIO('content'))
875
 
        self.assertEquals(True, transport.has('foo'))
876
 
 
877
 
    def test_mkdir(self):
878
 
        transport = memory.MemoryTransport()
879
 
        transport.mkdir('dir')
880
 
        transport.append('dir/path', StringIO('content'))
881
 
        self.assertEqual(transport.get('dir/path').read(), 'content')
882
 
 
883
 
    def test_mkdir_missing_parent(self):
884
 
        transport = memory.MemoryTransport()
885
 
        self.assertRaises(NoSuchFile,
886
 
                          transport.mkdir, 'dir/dir')
887
 
 
888
 
    def test_mkdir_twice(self):
889
 
        transport = memory.MemoryTransport()
890
 
        transport.mkdir('dir')
891
 
        self.assertRaises(FileExists, transport.mkdir, 'dir')
 
41
    def test__get_set_protocol_handlers(self):
 
42
        handlers = _get_protocol_handlers()
 
43
        self.assertNotEqual({}, handlers)
 
44
        try:
 
45
            _set_protocol_handlers({})
 
46
            self.assertEqual({}, _get_protocol_handlers())
 
47
        finally:
 
48
            _set_protocol_handlers(handlers)
 
49
 
 
50
    def test_get_transport_modules(self):
 
51
        handlers = _get_protocol_handlers()
 
52
        class SampleHandler(object):
 
53
            """I exist, isnt that enough?"""
 
54
        try:
 
55
            my_handlers = {}
 
56
            _set_protocol_handlers(my_handlers)
 
57
            register_lazy_transport('foo', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
 
58
            register_lazy_transport('bar', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
 
59
            self.assertEqual([SampleHandler.__module__],
 
60
                             _get_transport_modules())
 
61
        finally:
 
62
            _set_protocol_handlers(handlers)
 
63
            
 
64
 
 
65
class MemoryTransportTest(TestCase):
 
66
    """Memory transport specific tests."""
892
67
 
893
68
    def test_parameters(self):
894
 
        transport = memory.MemoryTransport()
895
 
        self.assertEqual(True, transport.listable())
896
 
        self.assertEqual(False, transport.should_cache())
897
 
 
898
 
    def test_iter_files_recursive(self):
899
 
        transport = memory.MemoryTransport()
900
 
        transport.mkdir('dir')
901
 
        transport.put('dir/foo', StringIO('content'))
902
 
        transport.put('dir/bar', StringIO('content'))
903
 
        transport.put('bar', StringIO('content'))
904
 
        paths = set(transport.iter_files_recursive())
905
 
        self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
906
 
 
907
 
    def test_stat(self):
908
 
        transport = memory.MemoryTransport()
909
 
        transport.put('foo', StringIO('content'))
910
 
        transport.put('bar', StringIO('phowar'))
911
 
        self.assertEqual(7, transport.stat('foo').st_size)
912
 
        self.assertEqual(6, transport.stat('bar').st_size)
913
 
 
 
69
        import bzrlib.transport.memory as memory
 
70
        transport = memory.MemoryTransport()
 
71
        self.assertEqual(True, transport.listable())
 
72
        self.assertEqual(False, transport.should_cache())
 
73
        self.assertEqual(False, transport.is_readonly())
 
74
        
 
75
        
 
76
class ReadonlyDecoratorTransportTest(TestCase):
 
77
    """Readonly decoration specific tests."""
 
78
 
 
79
    def test_local_parameters(self):
 
80
        import bzrlib.transport.readonly as readonly
 
81
        # connect to . in readonly mode
 
82
        transport = readonly.ReadonlyTransportDecorator('readonly+.')
 
83
        self.assertEqual(True, transport.listable())
 
84
        self.assertEqual(False, transport.should_cache())
 
85
        self.assertEqual(True, transport.is_readonly())
 
86
 
 
87
    def test_http_parameters(self):
 
88
        import bzrlib.transport.readonly as readonly
 
89
        from bzrlib.transport.http import HttpServer
 
90
        # connect to . via http which is not listable
 
91
        server = HttpServer()
 
92
        server.setUp()
 
93
        try:
 
94
            transport = get_transport('readonly+' + server.get_url())
 
95
            self.failUnless(isinstance(transport,
 
96
                                       readonly.ReadonlyTransportDecorator))
 
97
            self.assertEqual(False, transport.listable())
 
98
            self.assertEqual(True, transport.should_cache())
 
99
            self.assertEqual(True, transport.is_readonly())
 
100
        finally:
 
101
            server.tearDown()