← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

  • Committer: Robert Collins
  • Date: 2010-04-08 04:34:03 UTC
  • mfrom: (5138 +trunk)
  • mto: This revision was merged to the branch mainline in revision 5139.
  • Revision ID: robertc@robertcollins.net-20100408043403-56z0d07vdqrx7f3t
Update bugfix for 528114 to trunk.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
25
25
    errors,
26
26
    osutils,
27
27
    tests,
28
 
    transport as _mod_transport,
 
28
    transport,
29
29
    urlutils,
30
30
    )
31
31
from bzrlib.transport import (
 
32
    chroot,
32
33
    fakenfs,
 
34
    local,
33
35
    memory,
 
36
    pathfilter,
34
37
    readonly,
35
38
    )
36
 
from bzrlib.errors import (DependencyNotPresent,
37
 
                           FileExists,
38
 
                           InvalidURLJoin,
39
 
                           NoSuchFile,
40
 
                           PathNotChild,
41
 
                           ReadError,
42
 
                           UnsupportedProtocol,
43
 
                           )
44
 
from bzrlib.tests import features, TestCase, TestCaseInTempDir
45
 
from bzrlib.transport import (_clear_protocol_handlers,
46
 
                              _CoalescedOffset,
47
 
                              ConnectedTransport,
48
 
                              _get_protocol_handlers,
49
 
                              _set_protocol_handlers,
50
 
                              _get_transport_modules,
51
 
                              get_transport,
52
 
                              LateReadError,
53
 
                              register_lazy_transport,
54
 
                              register_transport_proto,
55
 
                              Transport,
56
 
                              )
57
 
from bzrlib.transport.chroot import ChrootServer
58
 
from bzrlib.transport.local import (LocalTransport,
59
 
                                    EmulatedWin32LocalTransport)
60
 
from bzrlib.transport.pathfilter import PathFilteringServer
 
39
from bzrlib.tests import (
 
40
    features,
 
41
    test_server,
 
42
    )
61
43
 
62
44
 
63
45
# TODO: Should possibly split transport-specific tests into their own files.
64
46
 
65
47
 
66
 
class TestTransport(TestCase):
 
48
class TestTransport(tests.TestCase):
67
49
    """Test the non transport-concrete class functionality."""
68
50
 
 
51
    # FIXME: These tests should use addCleanup() and/or overrideAttr() instead
 
52
    # of try/finally -- vila 20100205
 
53
 
69
54
    def test__get_set_protocol_handlers(self):
70
 
        handlers = _get_protocol_handlers()
 
55
        handlers = transport._get_protocol_handlers()
71
56
        self.assertNotEqual([], handlers.keys( ))
72
57
        try:
73
 
            _clear_protocol_handlers()
74
 
            self.assertEqual([], _get_protocol_handlers().keys())
 
58
            transport._clear_protocol_handlers()
 
59
            self.assertEqual([], transport._get_protocol_handlers().keys())
75
60
        finally:
76
 
            _set_protocol_handlers(handlers)
 
61
            transport._set_protocol_handlers(handlers)
77
62
 
78
63
    def test_get_transport_modules(self):
79
 
        handlers = _get_protocol_handlers()
 
64
        handlers = transport._get_protocol_handlers()
80
65
        # don't pollute the current handlers
81
 
        _clear_protocol_handlers()
 
66
        transport._clear_protocol_handlers()
82
67
        class SampleHandler(object):
83
68
            """I exist, isnt that enough?"""
84
69
        try:
85
 
            _clear_protocol_handlers()
86
 
            register_transport_proto('foo')
87
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
88
 
                                    'TestTransport.SampleHandler')
89
 
            register_transport_proto('bar')
90
 
            register_lazy_transport('bar', 'bzrlib.tests.test_transport',
91
 
                                    'TestTransport.SampleHandler')
 
70
            transport._clear_protocol_handlers()
 
71
            transport.register_transport_proto('foo')
 
72
            transport.register_lazy_transport('foo',
 
73
                                              'bzrlib.tests.test_transport',
 
74
                                              'TestTransport.SampleHandler')
 
75
            transport.register_transport_proto('bar')
 
76
            transport.register_lazy_transport('bar',
 
77
                                              'bzrlib.tests.test_transport',
 
78
                                              'TestTransport.SampleHandler')
92
79
            self.assertEqual([SampleHandler.__module__,
93
80
                              'bzrlib.transport.chroot',
94
81
                              'bzrlib.transport.pathfilter'],
95
 
                             _get_transport_modules())
 
82
                             transport._get_transport_modules())
96
83
        finally:
97
 
            _set_protocol_handlers(handlers)
 
84
            transport._set_protocol_handlers(handlers)
98
85
 
99
86
    def test_transport_dependency(self):
100
87
        """Transport with missing dependency causes no error"""
101
 
        saved_handlers = _get_protocol_handlers()
 
88
        saved_handlers = transport._get_protocol_handlers()
102
89
        # don't pollute the current handlers
103
 
        _clear_protocol_handlers()
 
90
        transport._clear_protocol_handlers()
104
91
        try:
105
 
            register_transport_proto('foo')
106
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
107
 
                    'BadTransportHandler')
 
92
            transport.register_transport_proto('foo')
 
93
            transport.register_lazy_transport(
 
94
                'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
108
95
            try:
109
 
                get_transport('foo://fooserver/foo')
110
 
            except UnsupportedProtocol, e:
 
96
                transport.get_transport('foo://fooserver/foo')
 
97
            except errors.UnsupportedProtocol, e:
111
98
                e_str = str(e)
112
99
                self.assertEquals('Unsupported protocol'
113
100
                                  ' for url "foo://fooserver/foo":'
117
104
                self.fail('Did not raise UnsupportedProtocol')
118
105
        finally:
119
106
            # restore original values
120
 
            _set_protocol_handlers(saved_handlers)
 
107
            transport._set_protocol_handlers(saved_handlers)
121
108
 
122
109
    def test_transport_fallback(self):
123
110
        """Transport with missing dependency causes no error"""
124
 
        saved_handlers = _get_protocol_handlers()
 
111
        saved_handlers = transport._get_protocol_handlers()
125
112
        try:
126
 
            _clear_protocol_handlers()
127
 
            register_transport_proto('foo')
128
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
129
 
                    'BackupTransportHandler')
130
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
131
 
                    'BadTransportHandler')
132
 
            t = get_transport('foo://fooserver/foo')
 
113
            transport._clear_protocol_handlers()
 
114
            transport.register_transport_proto('foo')
 
115
            transport.register_lazy_transport(
 
116
                'foo', 'bzrlib.tests.test_transport', 'BackupTransportHandler')
 
117
            transport.register_lazy_transport(
 
118
                'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
 
119
            t = transport.get_transport('foo://fooserver/foo')
133
120
            self.assertTrue(isinstance(t, BackupTransportHandler))
134
121
        finally:
135
 
            _set_protocol_handlers(saved_handlers)
 
122
            transport._set_protocol_handlers(saved_handlers)
136
123
 
137
124
    def test_ssh_hints(self):
138
125
        """Transport ssh:// should raise an error pointing out bzr+ssh://"""
139
126
        try:
140
 
            get_transport('ssh://fooserver/foo')
141
 
        except UnsupportedProtocol, e:
 
127
            transport.get_transport('ssh://fooserver/foo')
 
128
        except errors.UnsupportedProtocol, e:
142
129
            e_str = str(e)
143
130
            self.assertEquals('Unsupported protocol'
144
131
                              ' for url "ssh://fooserver/foo":'
145
 
                              ' bzr supports bzr+ssh to operate over ssh, use "bzr+ssh://fooserver/foo".',
 
132
                              ' bzr supports bzr+ssh to operate over ssh,'
 
133
                              ' use "bzr+ssh://fooserver/foo".',
146
134
                              str(e))
147
135
        else:
148
136
            self.fail('Did not raise UnsupportedProtocol')
149
137
 
150
138
    def test_LateReadError(self):
151
139
        """The LateReadError helper should raise on read()."""
152
 
        a_file = LateReadError('a path')
 
140
        a_file = transport.LateReadError('a path')
153
141
        try:
154
142
            a_file.read()
155
 
        except ReadError, error:
 
143
        except errors.ReadError, error:
156
144
            self.assertEqual('a path', error.path)
157
 
        self.assertRaises(ReadError, a_file.read, 40)
 
145
        self.assertRaises(errors.ReadError, a_file.read, 40)
158
146
        a_file.close()
159
147
 
160
148
    def test__combine_paths(self):
161
 
        t = Transport('/')
 
149
        t = transport.Transport('/')
162
150
        self.assertEqual('/home/sarah/project/foo',
163
151
                         t._combine_paths('/home/sarah', 'project/foo'))
164
152
        self.assertEqual('/etc',
175
163
        self.assertEqual('memory:///t is not a local path.', str(e))
176
164
 
177
165
 
178
 
class TestCoalesceOffsets(TestCase):
 
166
class TestCoalesceOffsets(tests.TestCase):
179
167
 
180
168
    def check(self, expected, offsets, limit=0, max_size=0, fudge=0):
181
 
        coalesce = Transport._coalesce_offsets
182
 
        exp = [_CoalescedOffset(*x) for x in expected]
 
169
        coalesce = transport.Transport._coalesce_offsets
 
170
        exp = [transport._CoalescedOffset(*x) for x in expected]
183
171
        out = list(coalesce(offsets, limit=limit, fudge_factor=fudge,
184
172
                            max_size=max_size))
185
173
        self.assertEqual(exp, out)
260
248
                   max_size=1*1024*1024*1024)
261
249
 
262
250
 
263
 
class TestMemoryServer(TestCase):
 
251
class TestMemoryServer(tests.TestCase):
264
252
 
265
253
    def test_create_server(self):
266
254
        server = memory.MemoryServer()
267
255
        server.start_server()
268
256
        url = server.get_url()
269
 
        self.assertTrue(url in _mod_transport.transport_list_registry)
270
 
        t = _mod_transport.get_transport(url)
 
257
        self.assertTrue(url in transport.transport_list_registry)
 
258
        t = transport.get_transport(url)
271
259
        del t
272
260
        server.stop_server()
273
 
        self.assertFalse(url in _mod_transport.transport_list_registry)
 
261
        self.assertFalse(url in transport.transport_list_registry)
274
262
        self.assertRaises(errors.UnsupportedProtocol,
275
 
                          _mod_transport.get_transport, url)
276
 
 
277
 
 
278
 
class TestMemoryTransport(TestCase):
 
263
                          transport.get_transport, url)
 
264
 
 
265
 
 
266
class TestMemoryTransport(tests.TestCase):
279
267
 
280
268
    def test_get_transport(self):
281
269
        memory.MemoryTransport()
282
270
 
283
271
    def test_clone(self):
284
 
        transport = memory.MemoryTransport()
285
 
        self.assertTrue(isinstance(transport, memory.MemoryTransport))
286
 
        self.assertEqual("memory:///", transport.clone("/").base)
 
272
        t = memory.MemoryTransport()
 
273
        self.assertTrue(isinstance(t, memory.MemoryTransport))
 
274
        self.assertEqual("memory:///", t.clone("/").base)
287
275
 
288
276
    def test_abspath(self):
289
 
        transport = memory.MemoryTransport()
290
 
        self.assertEqual("memory:///relpath", transport.abspath('relpath'))
 
277
        t = memory.MemoryTransport()
 
278
        self.assertEqual("memory:///relpath", t.abspath('relpath'))
291
279
 
292
280
    def test_abspath_of_root(self):
293
 
        transport = memory.MemoryTransport()
294
 
        self.assertEqual("memory:///", transport.base)
295
 
        self.assertEqual("memory:///", transport.abspath('/'))
 
281
        t = memory.MemoryTransport()
 
282
        self.assertEqual("memory:///", t.base)
 
283
        self.assertEqual("memory:///", t.abspath('/'))
296
284
 
297
285
    def test_abspath_of_relpath_starting_at_root(self):
298
 
        transport = memory.MemoryTransport()
299
 
        self.assertEqual("memory:///foo", transport.abspath('/foo'))
 
286
        t = memory.MemoryTransport()
 
287
        self.assertEqual("memory:///foo", t.abspath('/foo'))
300
288
 
301
289
    def test_append_and_get(self):
302
 
        transport = memory.MemoryTransport()
303
 
        transport.append_bytes('path', 'content')
304
 
        self.assertEqual(transport.get('path').read(), 'content')
305
 
        transport.append_file('path', StringIO('content'))
306
 
        self.assertEqual(transport.get('path').read(), 'contentcontent')
 
290
        t = memory.MemoryTransport()
 
291
        t.append_bytes('path', 'content')
 
292
        self.assertEqual(t.get('path').read(), 'content')
 
293
        t.append_file('path', StringIO('content'))
 
294
        self.assertEqual(t.get('path').read(), 'contentcontent')
307
295
 
308
296
    def test_put_and_get(self):
309
 
        transport = memory.MemoryTransport()
310
 
        transport.put_file('path', StringIO('content'))
311
 
        self.assertEqual(transport.get('path').read(), 'content')
312
 
        transport.put_bytes('path', 'content')
313
 
        self.assertEqual(transport.get('path').read(), 'content')
 
297
        t = memory.MemoryTransport()
 
298
        t.put_file('path', StringIO('content'))
 
299
        self.assertEqual(t.get('path').read(), 'content')
 
300
        t.put_bytes('path', 'content')
 
301
        self.assertEqual(t.get('path').read(), 'content')
314
302
 
315
303
    def test_append_without_dir_fails(self):
316
 
        transport = memory.MemoryTransport()
317
 
        self.assertRaises(NoSuchFile,
318
 
                          transport.append_bytes, 'dir/path', 'content')
 
304
        t = memory.MemoryTransport()
 
305
        self.assertRaises(errors.NoSuchFile,
 
306
                          t.append_bytes, 'dir/path', 'content')
319
307
 
320
308
    def test_put_without_dir_fails(self):
321
 
        transport = memory.MemoryTransport()
322
 
        self.assertRaises(NoSuchFile,
323
 
                          transport.put_file, 'dir/path', StringIO('content'))
 
309
        t = memory.MemoryTransport()
 
310
        self.assertRaises(errors.NoSuchFile,
 
311
                          t.put_file, 'dir/path', StringIO('content'))
324
312
 
325
313
    def test_get_missing(self):
326
314
        transport = memory.MemoryTransport()
327
 
        self.assertRaises(NoSuchFile, transport.get, 'foo')
 
315
        self.assertRaises(errors.NoSuchFile, transport.get, 'foo')
328
316
 
329
317
    def test_has_missing(self):
330
 
        transport = memory.MemoryTransport()
331
 
        self.assertEquals(False, transport.has('foo'))
 
318
        t = memory.MemoryTransport()
 
319
        self.assertEquals(False, t.has('foo'))
332
320
 
333
321
    def test_has_present(self):
334
 
        transport = memory.MemoryTransport()
335
 
        transport.append_bytes('foo', 'content')
336
 
        self.assertEquals(True, transport.has('foo'))
 
322
        t = memory.MemoryTransport()
 
323
        t.append_bytes('foo', 'content')
 
324
        self.assertEquals(True, t.has('foo'))
337
325
 
338
326
    def test_list_dir(self):
339
 
        transport = memory.MemoryTransport()
340
 
        transport.put_bytes('foo', 'content')
341
 
        transport.mkdir('dir')
342
 
        transport.put_bytes('dir/subfoo', 'content')
343
 
        transport.put_bytes('dirlike', 'content')
 
327
        t = memory.MemoryTransport()
 
328
        t.put_bytes('foo', 'content')
 
329
        t.mkdir('dir')
 
330
        t.put_bytes('dir/subfoo', 'content')
 
331
        t.put_bytes('dirlike', 'content')
344
332
 
345
 
        self.assertEquals(['dir', 'dirlike', 'foo'], sorted(transport.list_dir('.')))
346
 
        self.assertEquals(['subfoo'], sorted(transport.list_dir('dir')))
 
333
        self.assertEquals(['dir', 'dirlike', 'foo'], sorted(t.list_dir('.')))
 
334
        self.assertEquals(['subfoo'], sorted(t.list_dir('dir')))
347
335
 
348
336
    def test_mkdir(self):
349
 
        transport = memory.MemoryTransport()
350
 
        transport.mkdir('dir')
351
 
        transport.append_bytes('dir/path', 'content')
352
 
        self.assertEqual(transport.get('dir/path').read(), 'content')
 
337
        t = memory.MemoryTransport()
 
338
        t.mkdir('dir')
 
339
        t.append_bytes('dir/path', 'content')
 
340
        self.assertEqual(t.get('dir/path').read(), 'content')
353
341
 
354
342
    def test_mkdir_missing_parent(self):
355
 
        transport = memory.MemoryTransport()
356
 
        self.assertRaises(NoSuchFile,
357
 
                          transport.mkdir, 'dir/dir')
 
343
        t = memory.MemoryTransport()
 
344
        self.assertRaises(errors.NoSuchFile, t.mkdir, 'dir/dir')
358
345
 
359
346
    def test_mkdir_twice(self):
360
 
        transport = memory.MemoryTransport()
361
 
        transport.mkdir('dir')
362
 
        self.assertRaises(FileExists, transport.mkdir, 'dir')
 
347
        t = memory.MemoryTransport()
 
348
        t.mkdir('dir')
 
349
        self.assertRaises(errors.FileExists, t.mkdir, 'dir')
363
350
 
364
351
    def test_parameters(self):
365
 
        transport = memory.MemoryTransport()
366
 
        self.assertEqual(True, transport.listable())
367
 
        self.assertEqual(False, transport.is_readonly())
 
352
        t = memory.MemoryTransport()
 
353
        self.assertEqual(True, t.listable())
 
354
        self.assertEqual(False, t.is_readonly())
368
355
 
369
356
    def test_iter_files_recursive(self):
370
 
        transport = memory.MemoryTransport()
371
 
        transport.mkdir('dir')
372
 
        transport.put_bytes('dir/foo', 'content')
373
 
        transport.put_bytes('dir/bar', 'content')
374
 
        transport.put_bytes('bar', 'content')
375
 
        paths = set(transport.iter_files_recursive())
 
357
        t = memory.MemoryTransport()
 
358
        t.mkdir('dir')
 
359
        t.put_bytes('dir/foo', 'content')
 
360
        t.put_bytes('dir/bar', 'content')
 
361
        t.put_bytes('bar', 'content')
 
362
        paths = set(t.iter_files_recursive())
376
363
        self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
377
364
 
378
365
    def test_stat(self):
379
 
        transport = memory.MemoryTransport()
380
 
        transport.put_bytes('foo', 'content')
381
 
        transport.put_bytes('bar', 'phowar')
382
 
        self.assertEqual(7, transport.stat('foo').st_size)
383
 
        self.assertEqual(6, transport.stat('bar').st_size)
384
 
 
385
 
 
386
 
class ChrootDecoratorTransportTest(TestCase):
 
366
        t = memory.MemoryTransport()
 
367
        t.put_bytes('foo', 'content')
 
368
        t.put_bytes('bar', 'phowar')
 
369
        self.assertEqual(7, t.stat('foo').st_size)
 
370
        self.assertEqual(6, t.stat('bar').st_size)
 
371
 
 
372
 
 
373
class ChrootDecoratorTransportTest(tests.TestCase):
387
374
    """Chroot decoration specific tests."""
388
375
 
389
376
    def test_abspath(self):
390
377
        # The abspath is always relative to the chroot_url.
391
 
        server = ChrootServer(get_transport('memory:///foo/bar/'))
 
378
        server = chroot.ChrootServer(
 
379
            transport.get_transport('memory:///foo/bar/'))
392
380
        self.start_server(server)
393
 
        transport = get_transport(server.get_url())
394
 
        self.assertEqual(server.get_url(), transport.abspath('/'))
 
381
        t = transport.get_transport(server.get_url())
 
382
        self.assertEqual(server.get_url(), t.abspath('/'))
395
383
 
396
 
        subdir_transport = transport.clone('subdir')
397
 
        self.assertEqual(server.get_url(), subdir_transport.abspath('/'))
 
384
        subdir_t = t.clone('subdir')
 
385
        self.assertEqual(server.get_url(), subdir_t.abspath('/'))
398
386
 
399
387
    def test_clone(self):
400
 
        server = ChrootServer(get_transport('memory:///foo/bar/'))
 
388
        server = chroot.ChrootServer(
 
389
            transport.get_transport('memory:///foo/bar/'))
401
390
        self.start_server(server)
402
 
        transport = get_transport(server.get_url())
 
391
        t = transport.get_transport(server.get_url())
403
392
        # relpath from root and root path are the same
404
 
        relpath_cloned = transport.clone('foo')
405
 
        abspath_cloned = transport.clone('/foo')
 
393
        relpath_cloned = t.clone('foo')
 
394
        abspath_cloned = t.clone('/foo')
406
395
        self.assertEqual(server, relpath_cloned.server)
407
396
        self.assertEqual(server, abspath_cloned.server)
408
397
 
414
403
        This is so that it is not possible to escape a chroot by doing::
415
404
            url = chroot_transport.base
416
405
            parent_url = urlutils.join(url, '..')
417
 
            new_transport = get_transport(parent_url)
 
406
            new_t = transport.get_transport(parent_url)
418
407
        """
419
 
        server = ChrootServer(get_transport('memory:///path/subpath'))
 
408
        server = chroot.ChrootServer(
 
409
            transport.get_transport('memory:///path/subpath'))
420
410
        self.start_server(server)
421
 
        transport = get_transport(server.get_url())
422
 
        new_transport = get_transport(transport.base)
423
 
        self.assertEqual(transport.server, new_transport.server)
424
 
        self.assertEqual(transport.base, new_transport.base)
 
411
        t = transport.get_transport(server.get_url())
 
412
        new_t = transport.get_transport(t.base)
 
413
        self.assertEqual(t.server, new_t.server)
 
414
        self.assertEqual(t.base, new_t.base)
425
415
 
426
416
    def test_urljoin_preserves_chroot(self):
427
417
        """Using urlutils.join(url, '..') on a chroot URL should not produce a
430
420
        This is so that it is not possible to escape a chroot by doing::
431
421
            url = chroot_transport.base
432
422
            parent_url = urlutils.join(url, '..')
433
 
            new_transport = get_transport(parent_url)
 
423
            new_t = transport.get_transport(parent_url)
434
424
        """
435
 
        server = ChrootServer(get_transport('memory:///path/'))
 
425
        server = chroot.ChrootServer(transport.get_transport('memory:///path/'))
436
426
        self.start_server(server)
437
 
        transport = get_transport(server.get_url())
 
427
        t = transport.get_transport(server.get_url())
438
428
        self.assertRaises(
439
 
            InvalidURLJoin, urlutils.join, transport.base, '..')
440
 
 
441
 
 
442
 
class ChrootServerTest(TestCase):
 
429
            errors.InvalidURLJoin, urlutils.join, t.base, '..')
 
430
 
 
431
 
 
432
class TestChrootServer(tests.TestCase):
443
433
 
444
434
    def test_construct(self):
445
435
        backing_transport = memory.MemoryTransport()
446
 
        server = ChrootServer(backing_transport)
 
436
        server = chroot.ChrootServer(backing_transport)
447
437
        self.assertEqual(backing_transport, server.backing_transport)
448
438
 
449
439
    def test_setUp(self):
450
440
        backing_transport = memory.MemoryTransport()
451
 
        server = ChrootServer(backing_transport)
 
441
        server = chroot.ChrootServer(backing_transport)
452
442
        server.start_server()
453
443
        try:
454
 
            self.assertTrue(server.scheme in _get_protocol_handlers().keys())
 
444
            self.assertTrue(server.scheme
 
445
                            in transport._get_protocol_handlers().keys())
455
446
        finally:
456
447
            server.stop_server()
457
448
 
458
449
    def test_stop_server(self):
459
450
        backing_transport = memory.MemoryTransport()
460
 
        server = ChrootServer(backing_transport)
 
451
        server = chroot.ChrootServer(backing_transport)
461
452
        server.start_server()
462
453
        server.stop_server()
463
 
        self.assertFalse(server.scheme in _get_protocol_handlers().keys())
 
454
        self.assertFalse(server.scheme
 
455
                         in transport._get_protocol_handlers().keys())
464
456
 
465
457
    def test_get_url(self):
466
458
        backing_transport = memory.MemoryTransport()
467
 
        server = ChrootServer(backing_transport)
 
459
        server = chroot.ChrootServer(backing_transport)
468
460
        server.start_server()
469
461
        try:
470
462
            self.assertEqual('chroot-%d:///' % id(server), server.get_url())
472
464
            server.stop_server()
473
465
 
474
466
 
475
 
class PathFilteringDecoratorTransportTest(TestCase):
 
467
class PathFilteringDecoratorTransportTest(tests.TestCase):
476
468
    """Pathfilter decoration specific tests."""
477
469
 
478
470
    def test_abspath(self):
479
471
        # The abspath is always relative to the base of the backing transport.
480
 
        server = PathFilteringServer(get_transport('memory:///foo/bar/'),
 
472
        server = pathfilter.PathFilteringServer(
 
473
            transport.get_transport('memory:///foo/bar/'),
481
474
            lambda x: x)
482
475
        server.start_server()
483
 
        transport = get_transport(server.get_url())
484
 
        self.assertEqual(server.get_url(), transport.abspath('/'))
 
476
        t = transport.get_transport(server.get_url())
 
477
        self.assertEqual(server.get_url(), t.abspath('/'))
485
478
 
486
 
        subdir_transport = transport.clone('subdir')
487
 
        self.assertEqual(server.get_url(), subdir_transport.abspath('/'))
 
479
        subdir_t = t.clone('subdir')
 
480
        self.assertEqual(server.get_url(), subdir_t.abspath('/'))
488
481
        server.stop_server()
489
482
 
490
483
    def make_pf_transport(self, filter_func=None):
494
487
            parameter to override it."""
495
488
        if filter_func is None:
496
489
            filter_func = lambda x: x
497
 
        server = PathFilteringServer(
498
 
            get_transport('memory:///foo/bar/'), filter_func)
 
490
        server = pathfilter.PathFilteringServer(
 
491
            transport.get_transport('memory:///foo/bar/'), filter_func)
499
492
        server.start_server()
500
493
        self.addCleanup(server.stop_server)
501
 
        return get_transport(server.get_url())
 
494
        return transport.get_transport(server.get_url())
502
495
 
503
496
    def test__filter(self):
504
497
        # _filter (with an identity func as filter_func) always returns
505
498
        # paths relative to the base of the backing transport.
506
 
        transport = self.make_pf_transport()
507
 
        self.assertEqual('foo', transport._filter('foo'))
508
 
        self.assertEqual('foo/bar', transport._filter('foo/bar'))
509
 
        self.assertEqual('', transport._filter('..'))
510
 
        self.assertEqual('', transport._filter('/'))
 
499
        t = self.make_pf_transport()
 
500
        self.assertEqual('foo', t._filter('foo'))
 
501
        self.assertEqual('foo/bar', t._filter('foo/bar'))
 
502
        self.assertEqual('', t._filter('..'))
 
503
        self.assertEqual('', t._filter('/'))
511
504
        # The base of the pathfiltering transport is taken into account too.
512
 
        transport = transport.clone('subdir1/subdir2')
513
 
        self.assertEqual('subdir1/subdir2/foo', transport._filter('foo'))
514
 
        self.assertEqual(
515
 
            'subdir1/subdir2/foo/bar', transport._filter('foo/bar'))
516
 
        self.assertEqual('subdir1', transport._filter('..'))
517
 
        self.assertEqual('', transport._filter('/'))
 
505
        t = t.clone('subdir1/subdir2')
 
506
        self.assertEqual('subdir1/subdir2/foo', t._filter('foo'))
 
507
        self.assertEqual('subdir1/subdir2/foo/bar', t._filter('foo/bar'))
 
508
        self.assertEqual('subdir1', t._filter('..'))
 
509
        self.assertEqual('', t._filter('/'))
518
510
 
519
511
    def test_filter_invocation(self):
520
512
        filter_log = []
521
513
        def filter(path):
522
514
            filter_log.append(path)
523
515
            return path
524
 
        transport = self.make_pf_transport(filter)
525
 
        transport.has('abc')
 
516
        t = self.make_pf_transport(filter)
 
517
        t.has('abc')
526
518
        self.assertEqual(['abc'], filter_log)
527
519
        del filter_log[:]
528
 
        transport.clone('abc').has('xyz')
 
520
        t.clone('abc').has('xyz')
529
521
        self.assertEqual(['abc/xyz'], filter_log)
530
522
        del filter_log[:]
531
 
        transport.has('/abc')
 
523
        t.has('/abc')
532
524
        self.assertEqual(['abc'], filter_log)
533
525
 
534
526
    def test_clone(self):
535
 
        transport = self.make_pf_transport()
 
527
        t = self.make_pf_transport()
536
528
        # relpath from root and root path are the same
537
 
        relpath_cloned = transport.clone('foo')
538
 
        abspath_cloned = transport.clone('/foo')
539
 
        self.assertEqual(transport.server, relpath_cloned.server)
540
 
        self.assertEqual(transport.server, abspath_cloned.server)
 
529
        relpath_cloned = t.clone('foo')
 
530
        abspath_cloned = t.clone('/foo')
 
531
        self.assertEqual(t.server, relpath_cloned.server)
 
532
        self.assertEqual(t.server, abspath_cloned.server)
541
533
 
542
534
    def test_url_preserves_pathfiltering(self):
543
535
        """Calling get_transport on a pathfiltered transport's base should
548
540
        otherwise) the filtering by doing::
549
541
            url = filtered_transport.base
550
542
            parent_url = urlutils.join(url, '..')
551
 
            new_transport = get_transport(parent_url)
 
543
            new_t = transport.get_transport(parent_url)
552
544
        """
553
 
        transport = self.make_pf_transport()
554
 
        new_transport = get_transport(transport.base)
555
 
        self.assertEqual(transport.server, new_transport.server)
556
 
        self.assertEqual(transport.base, new_transport.base)
557
 
 
558
 
 
559
 
class ReadonlyDecoratorTransportTest(TestCase):
 
545
        t = self.make_pf_transport()
 
546
        new_t = transport.get_transport(t.base)
 
547
        self.assertEqual(t.server, new_t.server)
 
548
        self.assertEqual(t.base, new_t.base)
 
549
 
 
550
 
 
551
class ReadonlyDecoratorTransportTest(tests.TestCase):
560
552
    """Readonly decoration specific tests."""
561
553
 
562
554
    def test_local_parameters(self):
563
555
        # connect to . in readonly mode
564
 
        transport = readonly.ReadonlyTransportDecorator('readonly+.')
565
 
        self.assertEqual(True, transport.listable())
566
 
        self.assertEqual(True, transport.is_readonly())
 
556
        t = readonly.ReadonlyTransportDecorator('readonly+.')
 
557
        self.assertEqual(True, t.listable())
 
558
        self.assertEqual(True, t.is_readonly())
567
559
 
568
560
    def test_http_parameters(self):
569
561
        from bzrlib.tests.http_server import HttpServer
570
562
        # connect to '.' via http which is not listable
571
563
        server = HttpServer()
572
564
        self.start_server(server)
573
 
        transport = get_transport('readonly+' + server.get_url())
574
 
        self.failUnless(isinstance(transport,
575
 
                                   readonly.ReadonlyTransportDecorator))
576
 
        self.assertEqual(False, transport.listable())
577
 
        self.assertEqual(True, transport.is_readonly())
578
 
 
579
 
 
580
 
class FakeNFSDecoratorTests(TestCaseInTempDir):
 
565
        t = transport.get_transport('readonly+' + server.get_url())
 
566
        self.failUnless(isinstance(t, readonly.ReadonlyTransportDecorator))
 
567
        self.assertEqual(False, t.listable())
 
568
        self.assertEqual(True, t.is_readonly())
 
569
 
 
570
 
 
571
class FakeNFSDecoratorTests(tests.TestCaseInTempDir):
581
572
    """NFS decorator specific tests."""
582
573
 
583
574
    def get_nfs_transport(self, url):
587
578
    def test_local_parameters(self):
588
579
        # the listable and is_readonly parameters
589
580
        # are not changed by the fakenfs decorator
590
 
        transport = self.get_nfs_transport('.')
591
 
        self.assertEqual(True, transport.listable())
592
 
        self.assertEqual(False, transport.is_readonly())
 
581
        t = self.get_nfs_transport('.')
 
582
        self.assertEqual(True, t.listable())
 
583
        self.assertEqual(False, t.is_readonly())
593
584
 
594
585
    def test_http_parameters(self):
595
586
        # the listable and is_readonly parameters
598
589
        # connect to '.' via http which is not listable
599
590
        server = HttpServer()
600
591
        self.start_server(server)
601
 
        transport = self.get_nfs_transport(server.get_url())
602
 
        self.assertIsInstance(
603
 
            transport, fakenfs.FakeNFSTransportDecorator)
604
 
        self.assertEqual(False, transport.listable())
605
 
        self.assertEqual(True, transport.is_readonly())
 
592
        t = self.get_nfs_transport(server.get_url())
 
593
        self.assertIsInstance(t, fakenfs.FakeNFSTransportDecorator)
 
594
        self.assertEqual(False, t.listable())
 
595
        self.assertEqual(True, t.is_readonly())
606
596
 
607
597
    def test_fakenfs_server_default(self):
608
598
        # a FakeNFSServer() should bring up a local relpath server for itself
609
 
        server = fakenfs.FakeNFSServer()
 
599
        server = test_server.FakeNFSServer()
610
600
        self.start_server(server)
611
601
        # the url should be decorated appropriately
612
602
        self.assertStartsWith(server.get_url(), 'fakenfs+')
613
603
        # and we should be able to get a transport for it
614
 
        transport = get_transport(server.get_url())
 
604
        t = transport.get_transport(server.get_url())
615
605
        # which must be a FakeNFSTransportDecorator instance.
616
 
        self.assertIsInstance(transport, fakenfs.FakeNFSTransportDecorator)
 
606
        self.assertIsInstance(t, fakenfs.FakeNFSTransportDecorator)
617
607
 
618
608
    def test_fakenfs_rename_semantics(self):
619
609
        # a FakeNFS transport must mangle the way rename errors occur to
620
610
        # look like NFS problems.
621
 
        transport = self.get_nfs_transport('.')
 
611
        t = self.get_nfs_transport('.')
622
612
        self.build_tree(['from/', 'from/foo', 'to/', 'to/bar'],
623
 
                        transport=transport)
624
 
        self.assertRaises(errors.ResourceBusy,
625
 
                          transport.rename, 'from', 'to')
626
 
 
627
 
 
628
 
class FakeVFATDecoratorTests(TestCaseInTempDir):
 
613
                        transport=t)
 
614
        self.assertRaises(errors.ResourceBusy, t.rename, 'from', 'to')
 
615
 
 
616
 
 
617
class FakeVFATDecoratorTests(tests.TestCaseInTempDir):
629
618
    """Tests for simulation of VFAT restrictions"""
630
619
 
631
620
    def get_vfat_transport(self, url):
635
624
 
636
625
    def test_transport_creation(self):
637
626
        from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
638
 
        transport = self.get_vfat_transport('.')
639
 
        self.assertIsInstance(transport, FakeVFATTransportDecorator)
 
627
        t = self.get_vfat_transport('.')
 
628
        self.assertIsInstance(t, FakeVFATTransportDecorator)
640
629
 
641
630
    def test_transport_mkdir(self):
642
 
        transport = self.get_vfat_transport('.')
643
 
        transport.mkdir('HELLO')
644
 
        self.assertTrue(transport.has('hello'))
645
 
        self.assertTrue(transport.has('Hello'))
 
631
        t = self.get_vfat_transport('.')
 
632
        t.mkdir('HELLO')
 
633
        self.assertTrue(t.has('hello'))
 
634
        self.assertTrue(t.has('Hello'))
646
635
 
647
636
    def test_forbidden_chars(self):
648
 
        transport = self.get_vfat_transport('.')
649
 
        self.assertRaises(ValueError, transport.has, "<NU>")
650
 
 
651
 
 
652
 
class BadTransportHandler(Transport):
 
637
        t = self.get_vfat_transport('.')
 
638
        self.assertRaises(ValueError, t.has, "<NU>")
 
639
 
 
640
 
 
641
class BadTransportHandler(transport.Transport):
653
642
    def __init__(self, base_url):
654
 
        raise DependencyNotPresent('some_lib', 'testing missing dependency')
655
 
 
656
 
 
657
 
class BackupTransportHandler(Transport):
 
643
        raise errors.DependencyNotPresent('some_lib',
 
644
                                          'testing missing dependency')
 
645
 
 
646
 
 
647
class BackupTransportHandler(transport.Transport):
658
648
    """Test transport that works as a backup for the BadTransportHandler"""
659
649
    pass
660
650
 
661
651
 
662
 
class TestTransportImplementation(TestCaseInTempDir):
 
652
class TestTransportImplementation(tests.TestCaseInTempDir):
663
653
    """Implementation verification for transports.
664
654
 
665
655
    To verify a transport we need a server factory, which is a callable
694
684
        base_url = self._server.get_url()
695
685
        url = self._adjust_url(base_url, relpath)
696
686
        # try getting the transport via the regular interface:
697
 
        t = get_transport(url)
 
687
        t = transport.get_transport(url)
698
688
        # vila--20070607 if the following are commented out the test suite
699
689
        # still pass. Is this really still needed or was it a forgotten
700
690
        # temporary fix ?
705
695
        return t
706
696
 
707
697
 
708
 
class TestLocalTransports(TestCase):
 
698
class TestLocalTransports(tests.TestCase):
709
699
 
710
700
    def test_get_transport_from_abspath(self):
711
701
        here = osutils.abspath('.')
712
 
        t = get_transport(here)
713
 
        self.assertIsInstance(t, LocalTransport)
 
702
        t = transport.get_transport(here)
 
703
        self.assertIsInstance(t, local.LocalTransport)
714
704
        self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
715
705
 
716
706
    def test_get_transport_from_relpath(self):
717
707
        here = osutils.abspath('.')
718
 
        t = get_transport('.')
719
 
        self.assertIsInstance(t, LocalTransport)
 
708
        t = transport.get_transport('.')
 
709
        self.assertIsInstance(t, local.LocalTransport)
720
710
        self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
721
711
 
722
712
    def test_get_transport_from_local_url(self):
723
713
        here = osutils.abspath('.')
724
714
        here_url = urlutils.local_path_to_url(here) + '/'
725
 
        t = get_transport(here_url)
726
 
        self.assertIsInstance(t, LocalTransport)
 
715
        t = transport.get_transport(here_url)
 
716
        self.assertIsInstance(t, local.LocalTransport)
727
717
        self.assertEquals(t.base, here_url)
728
718
 
729
719
    def test_local_abspath(self):
730
720
        here = osutils.abspath('.')
731
 
        t = get_transport(here)
 
721
        t = transport.get_transport(here)
732
722
        self.assertEquals(t.local_abspath(''), here)
733
723
 
734
724
 
735
 
class TestWin32LocalTransport(TestCase):
 
725
class TestWin32LocalTransport(tests.TestCase):
736
726
 
737
727
    def test_unc_clone_to_root(self):
738
728
        # Win32 UNC path like \\HOST\path
739
729
        # clone to root should stop at least at \\HOST part
740
730
        # not on \\
741
 
        t = EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
 
731
        t = local.EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
742
732
        for i in xrange(4):
743
733
            t = t.clone('..')
744
734
        self.assertEquals(t.base, 'file://HOST/')
747
737
        self.assertEquals(t.base, 'file://HOST/')
748
738
 
749
739
 
750
 
class TestConnectedTransport(TestCase):
 
740
class TestConnectedTransport(tests.TestCase):
751
741
    """Tests for connected to remote server transports"""
752
742
 
753
743
    def test_parse_url(self):
754
 
        t = ConnectedTransport('http://simple.example.com/home/source')
 
744
        t = transport.ConnectedTransport(
 
745
            'http://simple.example.com/home/source')
755
746
        self.assertEquals(t._host, 'simple.example.com')
756
747
        self.assertEquals(t._port, None)
757
748
        self.assertEquals(t._path, '/home/source/')
762
753
 
763
754
    def test_parse_url_with_at_in_user(self):
764
755
        # Bug 228058
765
 
        t = ConnectedTransport('ftp://user@host.com@www.host.com/')
 
756
        t = transport.ConnectedTransport('ftp://user@host.com@www.host.com/')
766
757
        self.assertEquals(t._user, 'user@host.com')
767
758
 
768
759
    def test_parse_quoted_url(self):
769
 
        t = ConnectedTransport('http://ro%62ey:h%40t@ex%41mple.com:2222/path')
 
760
        t = transport.ConnectedTransport(
 
761
            'http://ro%62ey:h%40t@ex%41mple.com:2222/path')
770
762
        self.assertEquals(t._host, 'exAmple.com')
771
763
        self.assertEquals(t._port, 2222)
772
764
        self.assertEquals(t._user, 'robey')
778
770
 
779
771
    def test_parse_invalid_url(self):
780
772
        self.assertRaises(errors.InvalidURL,
781
 
                          ConnectedTransport,
 
773
                          transport.ConnectedTransport,
782
774
                          'sftp://lily.org:~janneke/public/bzr/gub')
783
775
 
784
776
    def test_relpath(self):
785
 
        t = ConnectedTransport('sftp://user@host.com/abs/path')
 
777
        t = transport.ConnectedTransport('sftp://user@host.com/abs/path')
786
778
 
787
779
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
788
780
        self.assertRaises(errors.PathNotChild, t.relpath,
794
786
        self.assertRaises(errors.PathNotChild, t.relpath,
795
787
                          'sftp://user@host.com:33/abs/path/sub')
796
788
        # Make sure it works when we don't supply a username
797
 
        t = ConnectedTransport('sftp://host.com/abs/path')
 
789
        t = transport.ConnectedTransport('sftp://host.com/abs/path')
798
790
        self.assertEquals(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
799
791
 
800
792
        # Make sure it works when parts of the path will be url encoded
801
 
        t = ConnectedTransport('sftp://host.com/dev/%path')
 
793
        t = transport.ConnectedTransport('sftp://host.com/dev/%path')
802
794
        self.assertEquals(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
803
795
 
804
796
    def test_connection_sharing_propagate_credentials(self):
805
 
        t = ConnectedTransport('ftp://user@host.com/abs/path')
 
797
        t = transport.ConnectedTransport('ftp://user@host.com/abs/path')
806
798
        self.assertEquals('user', t._user)
807
799
        self.assertEquals('host.com', t._host)
808
800
        self.assertIs(None, t._get_connection())
829
821
        self.assertIs(new_password, c._get_credentials())
830
822
 
831
823
 
832
 
class TestReusedTransports(TestCase):
 
824
class TestReusedTransports(tests.TestCase):
833
825
    """Tests for transport reuse"""
834
826
 
835
827
    def test_reuse_same_transport(self):
836
828
        possible_transports = []
837
 
        t1 = get_transport('http://foo/',
838
 
                           possible_transports=possible_transports)
 
829
        t1 = transport.get_transport('http://foo/',
 
830
                                     possible_transports=possible_transports)
839
831
        self.assertEqual([t1], possible_transports)
840
 
        t2 = get_transport('http://foo/', possible_transports=[t1])
 
832
        t2 = transport.get_transport('http://foo/',
 
833
                                     possible_transports=[t1])
841
834
        self.assertIs(t1, t2)
842
835
 
843
836
        # Also check that final '/' are handled correctly
844
 
        t3 = get_transport('http://foo/path/')
845
 
        t4 = get_transport('http://foo/path', possible_transports=[t3])
 
837
        t3 = transport.get_transport('http://foo/path/')
 
838
        t4 = transport.get_transport('http://foo/path',
 
839
                                     possible_transports=[t3])
846
840
        self.assertIs(t3, t4)
847
841
 
848
 
        t5 = get_transport('http://foo/path')
849
 
        t6 = get_transport('http://foo/path/', possible_transports=[t5])
 
842
        t5 = transport.get_transport('http://foo/path')
 
843
        t6 = transport.get_transport('http://foo/path/',
 
844
                                     possible_transports=[t5])
850
845
        self.assertIs(t5, t6)
851
846
 
852
847
    def test_don_t_reuse_different_transport(self):
853
 
        t1 = get_transport('http://foo/path')
854
 
        t2 = get_transport('http://bar/path', possible_transports=[t1])
 
848
        t1 = transport.get_transport('http://foo/path')
 
849
        t2 = transport.get_transport('http://bar/path',
 
850
                                     possible_transports=[t1])
855
851
        self.assertIsNot(t1, t2)
856
852
 
857
853
 
858
 
class TestTransportTrace(TestCase):
 
854
class TestTransportTrace(tests.TestCase):
859
855
 
860
856
    def test_get(self):
861
 
        transport = get_transport('trace+memory://')
862
 
        self.assertIsInstance(
863
 
            transport, bzrlib.transport.trace.TransportTraceDecorator)
 
857
        t = transport.get_transport('trace+memory://')
 
858
        self.assertIsInstance(t, bzrlib.transport.trace.TransportTraceDecorator)
864
859
 
865
860
    def test_clone_preserves_activity(self):
866
 
        transport = get_transport('trace+memory://')
867
 
        transport2 = transport.clone('.')
868
 
        self.assertTrue(transport is not transport2)
869
 
        self.assertTrue(transport._activity is transport2._activity)
 
861
        t = transport.get_transport('trace+memory://')
 
862
        t2 = t.clone('.')
 
863
        self.assertTrue(t is not t2)
 
864
        self.assertTrue(t._activity is t2._activity)
870
865
 
871
866
    # the following specific tests are for the operations that have made use of
872
867
    # logging in tests; we could test every single operation but doing that
873
868
    # still won't cause a test failure when the top level Transport API
874
869
    # changes; so there is little return doing that.
875
870
    def test_get(self):
876
 
        transport = get_transport('trace+memory:///')
877
 
        transport.put_bytes('foo', 'barish')
878
 
        transport.get('foo')
 
871
        t = transport.get_transport('trace+memory:///')
 
872
        t.put_bytes('foo', 'barish')
 
873
        t.get('foo')
879
874
        expected_result = []
880
875
        # put_bytes records the bytes, not the content to avoid memory
881
876
        # pressure.
882
877
        expected_result.append(('put_bytes', 'foo', 6, None))
883
878
        # get records the file name only.
884
879
        expected_result.append(('get', 'foo'))
885
 
        self.assertEqual(expected_result, transport._activity)
 
880
        self.assertEqual(expected_result, t._activity)
886
881
 
887
882
    def test_readv(self):
888
 
        transport = get_transport('trace+memory:///')
889
 
        transport.put_bytes('foo', 'barish')
890
 
        list(transport.readv('foo', [(0, 1), (3, 2)], adjust_for_latency=True,
891
 
            upper_limit=6))
 
883
        t = transport.get_transport('trace+memory:///')
 
884
        t.put_bytes('foo', 'barish')
 
885
        list(t.readv('foo', [(0, 1), (3, 2)],
 
886
                     adjust_for_latency=True, upper_limit=6))
892
887
        expected_result = []
893
888
        # put_bytes records the bytes, not the content to avoid memory
894
889
        # pressure.
895
890
        expected_result.append(('put_bytes', 'foo', 6, None))
896
891
        # readv records the supplied offset request
897
892
        expected_result.append(('readv', 'foo', [(0, 1), (3, 2)], True, 6))
898
 
        self.assertEqual(expected_result, transport._activity)
 
893
        self.assertEqual(expected_result, t._activity)
899
894
 
900
895
 
901
896
class TestSSHConnections(tests.TestCaseWithTransport):
977
972
            # prefix a '/' to get the right path.
978
973
            path_to_branch = '/' + path_to_branch
979
974
        url = 'bzr+ssh://fred:secret@localhost:%d%s' % (port, path_to_branch)
980
 
        t = get_transport(url)
 
975
        t = transport.get_transport(url)
981
976
        self.permit_url(t.base)
982
977
        t.mkdir('foo')
983
978