← 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: Jelmer Vernooij
  • Date: 2011-07-23 16:33:38 UTC
  • mto: This revision was merged to the branch mainline in revision 6045.
  • Revision ID: jelmer@samba.org-20110723163338-mjox55g52kt3u922
Add get_transport_from_url and get_transport_from_path functions.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
27
27
    transport,
28
28
    urlutils,
29
29
    )
 
30
from bzrlib.directory_service import directories
30
31
from bzrlib.transport import (
31
32
    chroot,
32
33
    fakenfs,
33
34
    http,
34
35
    local,
 
36
    location_to_url,
35
37
    memory,
36
38
    pathfilter,
37
39
    readonly,
38
40
    )
 
41
import bzrlib.transport.trace
39
42
from bzrlib.tests import (
40
43
    features,
41
44
    test_server,
48
51
class TestTransport(tests.TestCase):
49
52
    """Test the non transport-concrete class functionality."""
50
53
 
51
 
    # FIXME: These tests should use addCleanup() and/or overrideAttr() instead
52
 
    # of try/finally -- vila 20100205
53
 
 
54
54
    def test__get_set_protocol_handlers(self):
55
55
        handlers = transport._get_protocol_handlers()
56
 
        self.assertNotEqual([], handlers.keys( ))
57
 
        try:
58
 
            transport._clear_protocol_handlers()
59
 
            self.assertEqual([], transport._get_protocol_handlers().keys())
60
 
        finally:
61
 
            transport._set_protocol_handlers(handlers)
 
56
        self.assertNotEqual([], handlers.keys())
 
57
        transport._clear_protocol_handlers()
 
58
        self.addCleanup(transport._set_protocol_handlers, handlers)
 
59
        self.assertEqual([], transport._get_protocol_handlers().keys())
62
60
 
63
61
    def test_get_transport_modules(self):
64
62
        handlers = transport._get_protocol_handlers()
 
63
        self.addCleanup(transport._set_protocol_handlers, handlers)
65
64
        # don't pollute the current handlers
66
65
        transport._clear_protocol_handlers()
 
66
 
67
67
        class SampleHandler(object):
68
68
            """I exist, isnt that enough?"""
69
 
        try:
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')
79
 
            self.assertEqual([SampleHandler.__module__,
80
 
                              'bzrlib.transport.chroot',
81
 
                              'bzrlib.transport.pathfilter'],
82
 
                             transport._get_transport_modules())
83
 
        finally:
84
 
            transport._set_protocol_handlers(handlers)
 
69
        transport._clear_protocol_handlers()
 
70
        transport.register_transport_proto('foo')
 
71
        transport.register_lazy_transport('foo',
 
72
                                            'bzrlib.tests.test_transport',
 
73
                                            'TestTransport.SampleHandler')
 
74
        transport.register_transport_proto('bar')
 
75
        transport.register_lazy_transport('bar',
 
76
                                            'bzrlib.tests.test_transport',
 
77
                                            'TestTransport.SampleHandler')
 
78
        self.assertEqual([SampleHandler.__module__,
 
79
                            'bzrlib.transport.chroot',
 
80
                            'bzrlib.transport.pathfilter'],
 
81
                            transport._get_transport_modules())
85
82
 
86
83
    def test_transport_dependency(self):
87
84
        """Transport with missing dependency causes no error"""
88
85
        saved_handlers = transport._get_protocol_handlers()
 
86
        self.addCleanup(transport._set_protocol_handlers, saved_handlers)
89
87
        # don't pollute the current handlers
90
88
        transport._clear_protocol_handlers()
 
89
        transport.register_transport_proto('foo')
 
90
        transport.register_lazy_transport(
 
91
            'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
91
92
        try:
92
 
            transport.register_transport_proto('foo')
93
 
            transport.register_lazy_transport(
94
 
                'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
95
 
            try:
96
 
                transport.get_transport('foo://fooserver/foo')
97
 
            except errors.UnsupportedProtocol, e:
98
 
                e_str = str(e)
99
 
                self.assertEquals('Unsupported protocol'
100
 
                                  ' for url "foo://fooserver/foo":'
101
 
                                  ' Unable to import library "some_lib":'
102
 
                                  ' testing missing dependency', str(e))
103
 
            else:
104
 
                self.fail('Did not raise UnsupportedProtocol')
105
 
        finally:
106
 
            # restore original values
107
 
            transport._set_protocol_handlers(saved_handlers)
 
93
            transport.get_transport('foo://fooserver/foo')
 
94
        except errors.UnsupportedProtocol, e:
 
95
            e_str = str(e)
 
96
            self.assertEquals('Unsupported protocol'
 
97
                                ' for url "foo://fooserver/foo":'
 
98
                                ' Unable to import library "some_lib":'
 
99
                                ' testing missing dependency', str(e))
 
100
        else:
 
101
            self.fail('Did not raise UnsupportedProtocol')
108
102
 
109
103
    def test_transport_fallback(self):
110
104
        """Transport with missing dependency causes no error"""
111
105
        saved_handlers = transport._get_protocol_handlers()
112
 
        try:
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')
120
 
            self.assertTrue(isinstance(t, BackupTransportHandler))
121
 
        finally:
122
 
            transport._set_protocol_handlers(saved_handlers)
 
106
        self.addCleanup(transport._set_protocol_handlers, saved_handlers)
 
107
        transport._clear_protocol_handlers()
 
108
        transport.register_transport_proto('foo')
 
109
        transport.register_lazy_transport(
 
110
            'foo', 'bzrlib.tests.test_transport', 'BackupTransportHandler')
 
111
        transport.register_lazy_transport(
 
112
            'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
 
113
        t = transport.get_transport('foo://fooserver/foo')
 
114
        self.assertTrue(isinstance(t, BackupTransportHandler))
123
115
 
124
116
    def test_ssh_hints(self):
125
117
        """Transport ssh:// should raise an error pointing out bzr+ssh://"""
218
210
 
219
211
    def test_coalesce_fudge(self):
220
212
        self.check([(10, 30, [(0, 10), (20, 10)]),
221
 
                    (100, 10, [(0, 10),]),
 
213
                    (100, 10, [(0, 10)]),
222
214
                   ], [(10, 10), (30, 10), (100, 10)],
223
 
                   fudge=10
224
 
                  )
 
215
                   fudge=10)
 
216
 
225
217
    def test_coalesce_max_size(self):
226
218
        self.check([(10, 20, [(0, 10), (10, 10)]),
227
219
                    (30, 50, [(0, 50)]),
228
220
                    # If one range is above max_size, it gets its own coalesced
229
221
                    # offset
230
 
                    (100, 80, [(0, 80),]),],
 
222
                    (100, 80, [(0, 80)]),],
231
223
                   [(10, 10), (20, 10), (30, 50), (100, 80)],
232
 
                   max_size=50
233
 
                  )
 
224
                   max_size=50)
234
225
 
235
226
    def test_coalesce_no_max_size(self):
236
 
        self.check([(10, 170, [(0, 10), (10, 10), (20, 50), (70, 100)]),],
 
227
        self.check([(10, 170, [(0, 10), (10, 10), (20, 50), (70, 100)])],
237
228
                   [(10, 10), (20, 10), (30, 50), (80, 100)],
238
229
                  )
239
230
 
240
231
    def test_coalesce_default_limit(self):
241
232
        # By default we use a 100MB max size.
242
 
        ten_mb = 10*1024*1024
243
 
        self.check([(0, 10*ten_mb, [(i*ten_mb, ten_mb) for i in range(10)]),
 
233
        ten_mb = 10 * 1024 * 1024
 
234
        self.check([(0, 10 * ten_mb, [(i * ten_mb, ten_mb) for i in range(10)]),
244
235
                    (10*ten_mb, ten_mb, [(0, ten_mb)])],
245
236
                   [(i*ten_mb, ten_mb) for i in range(11)])
246
 
        self.check([(0, 11*ten_mb, [(i*ten_mb, ten_mb) for i in range(11)]),],
247
 
                   [(i*ten_mb, ten_mb) for i in range(11)],
 
237
        self.check([(0, 11 * ten_mb, [(i * ten_mb, ten_mb) for i in range(11)])],
 
238
                   [(i * ten_mb, ten_mb) for i in range(11)],
248
239
                   max_size=1*1024*1024*1024)
249
240
 
250
241
 
422
413
            parent_url = urlutils.join(url, '..')
423
414
            new_t = transport.get_transport(parent_url)
424
415
        """
425
 
        server = chroot.ChrootServer(transport.get_transport('memory:///path/'))
 
416
        server = chroot.ChrootServer(
 
417
            transport.get_transport('memory:///path/'))
426
418
        self.start_server(server)
427
419
        t = transport.get_transport(server.get_url())
428
420
        self.assertRaises(
440
432
        backing_transport = memory.MemoryTransport()
441
433
        server = chroot.ChrootServer(backing_transport)
442
434
        server.start_server()
443
 
        try:
444
 
            self.assertTrue(server.scheme
445
 
                            in transport._get_protocol_handlers().keys())
446
 
        finally:
447
 
            server.stop_server()
 
435
        self.addCleanup(server.stop_server)
 
436
        self.assertTrue(server.scheme
 
437
                        in transport._get_protocol_handlers().keys())
448
438
 
449
439
    def test_stop_server(self):
450
440
        backing_transport = memory.MemoryTransport()
458
448
        backing_transport = memory.MemoryTransport()
459
449
        server = chroot.ChrootServer(backing_transport)
460
450
        server.start_server()
461
 
        try:
462
 
            self.assertEqual('chroot-%d:///' % id(server), server.get_url())
463
 
        finally:
464
 
            server.stop_server()
 
451
        self.addCleanup(server.stop_server)
 
452
        self.assertEqual('chroot-%d:///' % id(server), server.get_url())
465
453
 
466
454
 
467
455
class PathFilteringDecoratorTransportTest(tests.TestCase):
482
470
 
483
471
    def make_pf_transport(self, filter_func=None):
484
472
        """Make a PathFilteringTransport backed by a MemoryTransport.
485
 
        
 
473
 
486
474
        :param filter_func: by default this will be a no-op function.  Use this
487
475
            parameter to override it."""
488
476
        if filter_func is None:
510
498
 
511
499
    def test_filter_invocation(self):
512
500
        filter_log = []
 
501
 
513
502
        def filter(path):
514
503
            filter_log.append(path)
515
504
            return path
684
673
        base_url = self._server.get_url()
685
674
        url = self._adjust_url(base_url, relpath)
686
675
        # try getting the transport via the regular interface:
687
 
        t = transport.get_transport(url)
 
676
        t = transport.get_transport_from_url(url)
688
677
        # vila--20070607 if the following are commented out the test suite
689
678
        # still pass. Is this really still needed or was it a forgotten
690
679
        # temporary fix ?
776
765
    def test_relpath(self):
777
766
        t = transport.ConnectedTransport('sftp://user@host.com/abs/path')
778
767
 
779
 
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
 
768
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'),
 
769
            'sub')
780
770
        self.assertRaises(errors.PathNotChild, t.relpath,
781
771
                          'http://user@host.com/abs/path/sub')
782
772
        self.assertRaises(errors.PathNotChild, t.relpath,
853
843
 
854
844
class TestTransportTrace(tests.TestCase):
855
845
 
856
 
    def test_get(self):
 
846
    def test_decorator(self):
857
847
        t = transport.get_transport('trace+memory://')
858
 
        self.assertIsInstance(t, bzrlib.transport.trace.TransportTraceDecorator)
 
848
        self.assertIsInstance(
 
849
            t, bzrlib.transport.trace.TransportTraceDecorator)
859
850
 
860
851
    def test_clone_preserves_activity(self):
861
852
        t = transport.get_transport('trace+memory://')
896
887
class TestSSHConnections(tests.TestCaseWithTransport):
897
888
 
898
889
    def test_bzr_connect_to_bzr_ssh(self):
899
 
        """User acceptance that get_transport of a bzr+ssh:// behaves correctly.
 
890
        """get_transport of a bzr+ssh:// behaves correctly.
900
891
 
901
892
        bzr+ssh:// should cause bzr to run a remote bzr smart server over SSH.
902
893
        """
918
909
        # SSH channel ourselves.  Surely this has already been implemented
919
910
        # elsewhere?
920
911
        started = []
 
912
 
921
913
        class StubSSHServer(stub_sftp.StubServer):
922
914
 
923
915
            test = self
929
921
                    stdout=subprocess.PIPE, stderr=subprocess.PIPE)
930
922
 
931
923
                # XXX: horribly inefficient, not to mention ugly.
932
 
                # Start a thread for each of stdin/out/err, and relay bytes from
933
 
                # the subprocess to channel and vice versa.
 
924
                # Start a thread for each of stdin/out/err, and relay bytes
 
925
                # from the subprocess to channel and vice versa.
934
926
                def ferry_bytes(read, write, close):
935
927
                    while True:
936
928
                        bytes = read(1)
1004
996
        result = http.unhtml_roughly(fake_html)
1005
997
        self.assertEquals(len(result), 1000)
1006
998
        self.assertStartsWith(result, " something!")
 
999
 
 
1000
 
 
1001
class SomeDirectory(object):
 
1002
 
 
1003
    def look_up(self, name, url):
 
1004
        return "http://bar"
 
1005
 
 
1006
 
 
1007
class TestLocationToUrl(tests.TestCase):
 
1008
 
 
1009
    def test_regular_url(self):
 
1010
        self.assertEquals("file://foo", location_to_url("file://foo"))
 
1011
 
 
1012
    def test_directory(self):
 
1013
        directories.register("bar:", SomeDirectory, "Dummy directory")
 
1014
        self.addCleanup(directories.remove, "bar:")
 
1015
        self.assertEquals("http://bar", location_to_url("bar:"))
 
1016
 
 
1017
    def test_unicode_url(self):
 
1018
        self.assertRaises(errors.InvalidURL, location_to_url,
 
1019
            "http://fo/\xc3\xaf".decode("utf-8"))
 
1020
 
 
1021
    def test_unicode_path(self):
 
1022
        self.assertEquals("file:///foo/bar%C3%AF",
 
1023
            location_to_url("/foo/bar\xc3\xaf".decode("utf-8")))
 
1024
 
 
1025
    def test_path(self):
 
1026
        self.assertEquals("file:///foo/bar", location_to_url("/foo/bar"))