← 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-05-16 13:39:39 UTC
  • mto: (5923.1.1 trunk)
  • mto: This revision was merged to the branch mainline in revision 5925.
  • Revision ID: jelmer@samba.org-20110516133939-8u1pc9utas3uw1lt
https://launchpad.net/bugs/592083
Require a unicode prompt to be passed into all methods that prompt.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
48
48
class TestTransport(tests.TestCase):
49
49
    """Test the non transport-concrete class functionality."""
50
50
 
 
51
    # FIXME: These tests should use addCleanup() and/or overrideAttr() instead
 
52
    # of try/finally -- vila 20100205
 
53
 
51
54
    def test__get_set_protocol_handlers(self):
52
55
        handlers = transport._get_protocol_handlers()
53
 
        self.assertNotEqual([], handlers.keys())
54
 
        transport._clear_protocol_handlers()
55
 
        self.addCleanup(transport._set_protocol_handlers, handlers)
56
 
        self.assertEqual([], transport._get_protocol_handlers().keys())
 
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)
57
62
 
58
63
    def test_get_transport_modules(self):
59
64
        handlers = transport._get_protocol_handlers()
60
 
        self.addCleanup(transport._set_protocol_handlers, handlers)
61
65
        # don't pollute the current handlers
62
66
        transport._clear_protocol_handlers()
63
 
 
64
67
        class SampleHandler(object):
65
68
            """I exist, isnt that enough?"""
66
 
        transport._clear_protocol_handlers()
67
 
        transport.register_transport_proto('foo')
68
 
        transport.register_lazy_transport('foo',
69
 
                                            'bzrlib.tests.test_transport',
70
 
                                            'TestTransport.SampleHandler')
71
 
        transport.register_transport_proto('bar')
72
 
        transport.register_lazy_transport('bar',
73
 
                                            'bzrlib.tests.test_transport',
74
 
                                            'TestTransport.SampleHandler')
75
 
        self.assertEqual([SampleHandler.__module__,
76
 
                            'bzrlib.transport.chroot',
77
 
                            'bzrlib.transport.pathfilter'],
78
 
                            transport._get_transport_modules())
 
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)
79
85
 
80
86
    def test_transport_dependency(self):
81
87
        """Transport with missing dependency causes no error"""
82
88
        saved_handlers = transport._get_protocol_handlers()
83
 
        self.addCleanup(transport._set_protocol_handlers, saved_handlers)
84
89
        # don't pollute the current handlers
85
90
        transport._clear_protocol_handlers()
86
 
        transport.register_transport_proto('foo')
87
 
        transport.register_lazy_transport(
88
 
            'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
89
91
        try:
90
 
            transport.get_transport('foo://fooserver/foo')
91
 
        except errors.UnsupportedProtocol, e:
92
 
            e_str = str(e)
93
 
            self.assertEquals('Unsupported protocol'
94
 
                                ' for url "foo://fooserver/foo":'
95
 
                                ' Unable to import library "some_lib":'
96
 
                                ' testing missing dependency', str(e))
97
 
        else:
98
 
            self.fail('Did not raise UnsupportedProtocol')
 
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)
99
108
 
100
109
    def test_transport_fallback(self):
101
110
        """Transport with missing dependency causes no error"""
102
111
        saved_handlers = transport._get_protocol_handlers()
103
 
        self.addCleanup(transport._set_protocol_handlers, saved_handlers)
104
 
        transport._clear_protocol_handlers()
105
 
        transport.register_transport_proto('foo')
106
 
        transport.register_lazy_transport(
107
 
            'foo', 'bzrlib.tests.test_transport', 'BackupTransportHandler')
108
 
        transport.register_lazy_transport(
109
 
            'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
110
 
        t = transport.get_transport('foo://fooserver/foo')
111
 
        self.assertTrue(isinstance(t, BackupTransportHandler))
 
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)
112
123
 
113
124
    def test_ssh_hints(self):
114
125
        """Transport ssh:// should raise an error pointing out bzr+ssh://"""
207
218
 
208
219
    def test_coalesce_fudge(self):
209
220
        self.check([(10, 30, [(0, 10), (20, 10)]),
210
 
                    (100, 10, [(0, 10)]),
 
221
                    (100, 10, [(0, 10),]),
211
222
                   ], [(10, 10), (30, 10), (100, 10)],
212
 
                   fudge=10)
213
 
 
 
223
                   fudge=10
 
224
                  )
214
225
    def test_coalesce_max_size(self):
215
226
        self.check([(10, 20, [(0, 10), (10, 10)]),
216
227
                    (30, 50, [(0, 50)]),
217
228
                    # If one range is above max_size, it gets its own coalesced
218
229
                    # offset
219
 
                    (100, 80, [(0, 80)]),],
 
230
                    (100, 80, [(0, 80),]),],
220
231
                   [(10, 10), (20, 10), (30, 50), (100, 80)],
221
 
                   max_size=50)
 
232
                   max_size=50
 
233
                  )
222
234
 
223
235
    def test_coalesce_no_max_size(self):
224
 
        self.check([(10, 170, [(0, 10), (10, 10), (20, 50), (70, 100)])],
 
236
        self.check([(10, 170, [(0, 10), (10, 10), (20, 50), (70, 100)]),],
225
237
                   [(10, 10), (20, 10), (30, 50), (80, 100)],
226
238
                  )
227
239
 
228
240
    def test_coalesce_default_limit(self):
229
241
        # By default we use a 100MB max size.
230
 
        ten_mb = 10 * 1024 * 1024
231
 
        self.check([(0, 10 * ten_mb, [(i * ten_mb, ten_mb) for i in range(10)]),
 
242
        ten_mb = 10*1024*1024
 
243
        self.check([(0, 10*ten_mb, [(i*ten_mb, ten_mb) for i in range(10)]),
232
244
                    (10*ten_mb, ten_mb, [(0, ten_mb)])],
233
245
                   [(i*ten_mb, ten_mb) for i in range(11)])
234
 
        self.check([(0, 11 * ten_mb, [(i * ten_mb, ten_mb) for i in range(11)])],
235
 
                   [(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)],
236
248
                   max_size=1*1024*1024*1024)
237
249
 
238
250
 
410
422
            parent_url = urlutils.join(url, '..')
411
423
            new_t = transport.get_transport(parent_url)
412
424
        """
413
 
        server = chroot.ChrootServer(
414
 
            transport.get_transport('memory:///path/'))
 
425
        server = chroot.ChrootServer(transport.get_transport('memory:///path/'))
415
426
        self.start_server(server)
416
427
        t = transport.get_transport(server.get_url())
417
428
        self.assertRaises(
429
440
        backing_transport = memory.MemoryTransport()
430
441
        server = chroot.ChrootServer(backing_transport)
431
442
        server.start_server()
432
 
        self.addCleanup(server.stop_server)
433
 
        self.assertTrue(server.scheme
434
 
                        in transport._get_protocol_handlers().keys())
 
443
        try:
 
444
            self.assertTrue(server.scheme
 
445
                            in transport._get_protocol_handlers().keys())
 
446
        finally:
 
447
            server.stop_server()
435
448
 
436
449
    def test_stop_server(self):
437
450
        backing_transport = memory.MemoryTransport()
445
458
        backing_transport = memory.MemoryTransport()
446
459
        server = chroot.ChrootServer(backing_transport)
447
460
        server.start_server()
448
 
        self.addCleanup(server.stop_server)
449
 
        self.assertEqual('chroot-%d:///' % id(server), server.get_url())
 
461
        try:
 
462
            self.assertEqual('chroot-%d:///' % id(server), server.get_url())
 
463
        finally:
 
464
            server.stop_server()
450
465
 
451
466
 
452
467
class PathFilteringDecoratorTransportTest(tests.TestCase):
467
482
 
468
483
    def make_pf_transport(self, filter_func=None):
469
484
        """Make a PathFilteringTransport backed by a MemoryTransport.
470
 
 
 
485
        
471
486
        :param filter_func: by default this will be a no-op function.  Use this
472
487
            parameter to override it."""
473
488
        if filter_func is None:
495
510
 
496
511
    def test_filter_invocation(self):
497
512
        filter_log = []
498
 
 
499
513
        def filter(path):
500
514
            filter_log.append(path)
501
515
            return path
762
776
    def test_relpath(self):
763
777
        t = transport.ConnectedTransport('sftp://user@host.com/abs/path')
764
778
 
765
 
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'),
766
 
            'sub')
 
779
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
767
780
        self.assertRaises(errors.PathNotChild, t.relpath,
768
781
                          'http://user@host.com/abs/path/sub')
769
782
        self.assertRaises(errors.PathNotChild, t.relpath,
842
855
 
843
856
    def test_get(self):
844
857
        t = transport.get_transport('trace+memory://')
845
 
        self.assertIsInstance(
846
 
            t, bzrlib.transport.trace.TransportTraceDecorator)
 
858
        self.assertIsInstance(t, bzrlib.transport.trace.TransportTraceDecorator)
847
859
 
848
860
    def test_clone_preserves_activity(self):
849
861
        t = transport.get_transport('trace+memory://')
884
896
class TestSSHConnections(tests.TestCaseWithTransport):
885
897
 
886
898
    def test_bzr_connect_to_bzr_ssh(self):
887
 
        """get_transport of a bzr+ssh:// behaves correctly.
 
899
        """User acceptance that get_transport of a bzr+ssh:// behaves correctly.
888
900
 
889
901
        bzr+ssh:// should cause bzr to run a remote bzr smart server over SSH.
890
902
        """
906
918
        # SSH channel ourselves.  Surely this has already been implemented
907
919
        # elsewhere?
908
920
        started = []
909
 
 
910
921
        class StubSSHServer(stub_sftp.StubServer):
911
922
 
912
923
            test = self
918
929
                    stdout=subprocess.PIPE, stderr=subprocess.PIPE)
919
930
 
920
931
                # XXX: horribly inefficient, not to mention ugly.
921
 
                # Start a thread for each of stdin/out/err, and relay bytes
922
 
                # from the subprocess to channel and vice versa.
 
932
                # Start a thread for each of stdin/out/err, and relay bytes from
 
933
                # the subprocess to channel and vice versa.
923
934
                def ferry_bytes(read, write, close):
924
935
                    while True:
925
936
                        bytes = read(1)