← 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: Samuel Bronson
  • Date: 2012-08-30 20:36:18 UTC
  • mto: (6015.57.3 2.4)
  • mto: This revision was merged to the branch mainline in revision 6558.
  • Revision ID: naesten@gmail.com-20120830203618-y2dzw91nqpvpgxvx
Update INSTALL for switch to Python 2.6 and up.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
29
29
    transport,
30
30
    urlutils,
31
31
    )
32
 
from bzrlib.directory_service import directories
33
32
from bzrlib.transport import (
34
33
    chroot,
35
34
    fakenfs,
36
35
    http,
37
36
    local,
38
 
    location_to_url,
39
37
    memory,
40
38
    pathfilter,
41
39
    readonly,
42
40
    )
43
 
import bzrlib.transport.trace
44
41
from bzrlib.tests import (
45
42
    features,
46
43
    test_server,
53
50
class TestTransport(tests.TestCase):
54
51
    """Test the non transport-concrete class functionality."""
55
52
 
 
53
    # FIXME: These tests should use addCleanup() and/or overrideAttr() instead
 
54
    # of try/finally -- vila 20100205
 
55
 
56
56
    def test__get_set_protocol_handlers(self):
57
57
        handlers = transport._get_protocol_handlers()
58
 
        self.assertNotEqual([], handlers.keys())
59
 
        transport._clear_protocol_handlers()
60
 
        self.addCleanup(transport._set_protocol_handlers, handlers)
61
 
        self.assertEqual([], transport._get_protocol_handlers().keys())
 
58
        self.assertNotEqual([], handlers.keys( ))
 
59
        try:
 
60
            transport._clear_protocol_handlers()
 
61
            self.assertEqual([], transport._get_protocol_handlers().keys())
 
62
        finally:
 
63
            transport._set_protocol_handlers(handlers)
62
64
 
63
65
    def test_get_transport_modules(self):
64
66
        handlers = transport._get_protocol_handlers()
65
 
        self.addCleanup(transport._set_protocol_handlers, handlers)
66
67
        # don't pollute the current handlers
67
68
        transport._clear_protocol_handlers()
68
 
 
69
69
        class SampleHandler(object):
70
70
            """I exist, isnt that enough?"""
71
 
        transport._clear_protocol_handlers()
72
 
        transport.register_transport_proto('foo')
73
 
        transport.register_lazy_transport('foo',
74
 
                                            'bzrlib.tests.test_transport',
75
 
                                            'TestTransport.SampleHandler')
76
 
        transport.register_transport_proto('bar')
77
 
        transport.register_lazy_transport('bar',
78
 
                                            'bzrlib.tests.test_transport',
79
 
                                            'TestTransport.SampleHandler')
80
 
        self.assertEqual([SampleHandler.__module__,
81
 
                            'bzrlib.transport.chroot',
82
 
                            'bzrlib.transport.pathfilter'],
83
 
                            transport._get_transport_modules())
 
71
        try:
 
72
            transport._clear_protocol_handlers()
 
73
            transport.register_transport_proto('foo')
 
74
            transport.register_lazy_transport('foo',
 
75
                                              'bzrlib.tests.test_transport',
 
76
                                              'TestTransport.SampleHandler')
 
77
            transport.register_transport_proto('bar')
 
78
            transport.register_lazy_transport('bar',
 
79
                                              'bzrlib.tests.test_transport',
 
80
                                              'TestTransport.SampleHandler')
 
81
            self.assertEqual([SampleHandler.__module__,
 
82
                              'bzrlib.transport.chroot',
 
83
                              'bzrlib.transport.pathfilter'],
 
84
                             transport._get_transport_modules())
 
85
        finally:
 
86
            transport._set_protocol_handlers(handlers)
84
87
 
85
88
    def test_transport_dependency(self):
86
89
        """Transport with missing dependency causes no error"""
87
90
        saved_handlers = transport._get_protocol_handlers()
88
 
        self.addCleanup(transport._set_protocol_handlers, saved_handlers)
89
91
        # don't pollute the current handlers
90
92
        transport._clear_protocol_handlers()
91
 
        transport.register_transport_proto('foo')
92
 
        transport.register_lazy_transport(
93
 
            'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
94
93
        try:
95
 
            transport.get_transport_from_url('foo://fooserver/foo')
96
 
        except errors.UnsupportedProtocol, e:
97
 
            e_str = str(e)
98
 
            self.assertEquals('Unsupported protocol'
99
 
                                ' for url "foo://fooserver/foo":'
100
 
                                ' Unable to import library "some_lib":'
101
 
                                ' testing missing dependency', str(e))
102
 
        else:
103
 
            self.fail('Did not raise UnsupportedProtocol')
 
94
            transport.register_transport_proto('foo')
 
95
            transport.register_lazy_transport(
 
96
                'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
 
97
            try:
 
98
                transport.get_transport('foo://fooserver/foo')
 
99
            except errors.UnsupportedProtocol, e:
 
100
                e_str = str(e)
 
101
                self.assertEquals('Unsupported protocol'
 
102
                                  ' for url "foo://fooserver/foo":'
 
103
                                  ' Unable to import library "some_lib":'
 
104
                                  ' testing missing dependency', str(e))
 
105
            else:
 
106
                self.fail('Did not raise UnsupportedProtocol')
 
107
        finally:
 
108
            # restore original values
 
109
            transport._set_protocol_handlers(saved_handlers)
104
110
 
105
111
    def test_transport_fallback(self):
106
112
        """Transport with missing dependency causes no error"""
107
113
        saved_handlers = transport._get_protocol_handlers()
108
 
        self.addCleanup(transport._set_protocol_handlers, saved_handlers)
109
 
        transport._clear_protocol_handlers()
110
 
        transport.register_transport_proto('foo')
111
 
        transport.register_lazy_transport(
112
 
            'foo', 'bzrlib.tests.test_transport', 'BackupTransportHandler')
113
 
        transport.register_lazy_transport(
114
 
            'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
115
 
        t = transport.get_transport_from_url('foo://fooserver/foo')
116
 
        self.assertTrue(isinstance(t, BackupTransportHandler))
 
114
        try:
 
115
            transport._clear_protocol_handlers()
 
116
            transport.register_transport_proto('foo')
 
117
            transport.register_lazy_transport(
 
118
                'foo', 'bzrlib.tests.test_transport', 'BackupTransportHandler')
 
119
            transport.register_lazy_transport(
 
120
                'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
 
121
            t = transport.get_transport('foo://fooserver/foo')
 
122
            self.assertTrue(isinstance(t, BackupTransportHandler))
 
123
        finally:
 
124
            transport._set_protocol_handlers(saved_handlers)
117
125
 
118
126
    def test_ssh_hints(self):
119
127
        """Transport ssh:// should raise an error pointing out bzr+ssh://"""
120
128
        try:
121
 
            transport.get_transport_from_url('ssh://fooserver/foo')
 
129
            transport.get_transport('ssh://fooserver/foo')
122
130
        except errors.UnsupportedProtocol, e:
123
131
            e_str = str(e)
124
132
            self.assertEquals('Unsupported protocol'
139
147
        self.assertRaises(errors.ReadError, a_file.read, 40)
140
148
        a_file.close()
141
149
 
 
150
    def test__combine_paths(self):
 
151
        t = transport.Transport('/')
 
152
        self.assertEqual('/home/sarah/project/foo',
 
153
                         t._combine_paths('/home/sarah', 'project/foo'))
 
154
        self.assertEqual('/etc',
 
155
                         t._combine_paths('/home/sarah', '../../etc'))
 
156
        self.assertEqual('/etc',
 
157
                         t._combine_paths('/home/sarah', '../../../etc'))
 
158
        self.assertEqual('/etc',
 
159
                         t._combine_paths('/home/sarah', '/etc'))
 
160
 
142
161
    def test_local_abspath_non_local_transport(self):
143
162
        # the base implementation should throw
144
163
        t = memory.MemoryTransport()
201
220
 
202
221
    def test_coalesce_fudge(self):
203
222
        self.check([(10, 30, [(0, 10), (20, 10)]),
204
 
                    (100, 10, [(0, 10)]),
 
223
                    (100, 10, [(0, 10),]),
205
224
                   ], [(10, 10), (30, 10), (100, 10)],
206
 
                   fudge=10)
207
 
 
 
225
                   fudge=10
 
226
                  )
208
227
    def test_coalesce_max_size(self):
209
228
        self.check([(10, 20, [(0, 10), (10, 10)]),
210
229
                    (30, 50, [(0, 50)]),
211
230
                    # If one range is above max_size, it gets its own coalesced
212
231
                    # offset
213
 
                    (100, 80, [(0, 80)]),],
 
232
                    (100, 80, [(0, 80),]),],
214
233
                   [(10, 10), (20, 10), (30, 50), (100, 80)],
215
 
                   max_size=50)
 
234
                   max_size=50
 
235
                  )
216
236
 
217
237
    def test_coalesce_no_max_size(self):
218
 
        self.check([(10, 170, [(0, 10), (10, 10), (20, 50), (70, 100)])],
 
238
        self.check([(10, 170, [(0, 10), (10, 10), (20, 50), (70, 100)]),],
219
239
                   [(10, 10), (20, 10), (30, 50), (80, 100)],
220
240
                  )
221
241
 
222
242
    def test_coalesce_default_limit(self):
223
243
        # By default we use a 100MB max size.
224
 
        ten_mb = 10 * 1024 * 1024
225
 
        self.check([(0, 10 * ten_mb, [(i * ten_mb, ten_mb) for i in range(10)]),
 
244
        ten_mb = 10*1024*1024
 
245
        self.check([(0, 10*ten_mb, [(i*ten_mb, ten_mb) for i in range(10)]),
226
246
                    (10*ten_mb, ten_mb, [(0, ten_mb)])],
227
247
                   [(i*ten_mb, ten_mb) for i in range(11)])
228
 
        self.check([(0, 11 * ten_mb, [(i * ten_mb, ten_mb) for i in range(11)])],
229
 
                   [(i * ten_mb, ten_mb) for i in range(11)],
 
248
        self.check([(0, 11*ten_mb, [(i*ten_mb, ten_mb) for i in range(11)]),],
 
249
                   [(i*ten_mb, ten_mb) for i in range(11)],
230
250
                   max_size=1*1024*1024*1024)
231
251
 
232
252
 
237
257
        server.start_server()
238
258
        url = server.get_url()
239
259
        self.assertTrue(url in transport.transport_list_registry)
240
 
        t = transport.get_transport_from_url(url)
 
260
        t = transport.get_transport(url)
241
261
        del t
242
262
        server.stop_server()
243
263
        self.assertFalse(url in transport.transport_list_registry)
358
378
    def test_abspath(self):
359
379
        # The abspath is always relative to the chroot_url.
360
380
        server = chroot.ChrootServer(
361
 
            transport.get_transport_from_url('memory:///foo/bar/'))
 
381
            transport.get_transport('memory:///foo/bar/'))
362
382
        self.start_server(server)
363
 
        t = transport.get_transport_from_url(server.get_url())
 
383
        t = transport.get_transport(server.get_url())
364
384
        self.assertEqual(server.get_url(), t.abspath('/'))
365
385
 
366
386
        subdir_t = t.clone('subdir')
368
388
 
369
389
    def test_clone(self):
370
390
        server = chroot.ChrootServer(
371
 
            transport.get_transport_from_url('memory:///foo/bar/'))
 
391
            transport.get_transport('memory:///foo/bar/'))
372
392
        self.start_server(server)
373
 
        t = transport.get_transport_from_url(server.get_url())
 
393
        t = transport.get_transport(server.get_url())
374
394
        # relpath from root and root path are the same
375
395
        relpath_cloned = t.clone('foo')
376
396
        abspath_cloned = t.clone('/foo')
385
405
        This is so that it is not possible to escape a chroot by doing::
386
406
            url = chroot_transport.base
387
407
            parent_url = urlutils.join(url, '..')
388
 
            new_t = transport.get_transport_from_url(parent_url)
 
408
            new_t = transport.get_transport(parent_url)
389
409
        """
390
410
        server = chroot.ChrootServer(
391
 
            transport.get_transport_from_url('memory:///path/subpath'))
 
411
            transport.get_transport('memory:///path/subpath'))
392
412
        self.start_server(server)
393
 
        t = transport.get_transport_from_url(server.get_url())
394
 
        new_t = transport.get_transport_from_url(t.base)
 
413
        t = transport.get_transport(server.get_url())
 
414
        new_t = transport.get_transport(t.base)
395
415
        self.assertEqual(t.server, new_t.server)
396
416
        self.assertEqual(t.base, new_t.base)
397
417
 
402
422
        This is so that it is not possible to escape a chroot by doing::
403
423
            url = chroot_transport.base
404
424
            parent_url = urlutils.join(url, '..')
405
 
            new_t = transport.get_transport_from_url(parent_url)
 
425
            new_t = transport.get_transport(parent_url)
406
426
        """
407
 
        server = chroot.ChrootServer(
408
 
            transport.get_transport_from_url('memory:///path/'))
 
427
        server = chroot.ChrootServer(transport.get_transport('memory:///path/'))
409
428
        self.start_server(server)
410
 
        t = transport.get_transport_from_url(server.get_url())
 
429
        t = transport.get_transport(server.get_url())
411
430
        self.assertRaises(
412
431
            errors.InvalidURLJoin, urlutils.join, t.base, '..')
413
432
 
423
442
        backing_transport = memory.MemoryTransport()
424
443
        server = chroot.ChrootServer(backing_transport)
425
444
        server.start_server()
426
 
        self.addCleanup(server.stop_server)
427
 
        self.assertTrue(server.scheme
428
 
                        in transport._get_protocol_handlers().keys())
 
445
        try:
 
446
            self.assertTrue(server.scheme
 
447
                            in transport._get_protocol_handlers().keys())
 
448
        finally:
 
449
            server.stop_server()
429
450
 
430
451
    def test_stop_server(self):
431
452
        backing_transport = memory.MemoryTransport()
439
460
        backing_transport = memory.MemoryTransport()
440
461
        server = chroot.ChrootServer(backing_transport)
441
462
        server.start_server()
442
 
        self.addCleanup(server.stop_server)
443
 
        self.assertEqual('chroot-%d:///' % id(server), server.get_url())
444
 
 
445
 
 
446
 
class TestHooks(tests.TestCase):
447
 
    """Basic tests for transport hooks"""
448
 
 
449
 
    def _get_connected_transport(self):
450
 
        return transport.ConnectedTransport("bogus:nowhere")
451
 
 
452
 
    def test_transporthooks_initialisation(self):
453
 
        """Check all expected transport hook points are set up"""
454
 
        hookpoint = transport.TransportHooks()
455
 
        self.assertTrue("post_connect" in hookpoint,
456
 
            "post_connect not in %s" % (hookpoint,))
457
 
 
458
 
    def test_post_connect(self):
459
 
        """Ensure the post_connect hook is called when _set_transport is"""
460
 
        calls = []
461
 
        transport.Transport.hooks.install_named_hook("post_connect",
462
 
            calls.append, None)
463
 
        t = self._get_connected_transport()
464
 
        self.assertLength(0, calls)
465
 
        t._set_connection("connection", "auth")
466
 
        self.assertEqual(calls, [t])
 
463
        try:
 
464
            self.assertEqual('chroot-%d:///' % id(server), server.get_url())
 
465
        finally:
 
466
            server.stop_server()
467
467
 
468
468
 
469
469
class PathFilteringDecoratorTransportTest(tests.TestCase):
472
472
    def test_abspath(self):
473
473
        # The abspath is always relative to the base of the backing transport.
474
474
        server = pathfilter.PathFilteringServer(
475
 
            transport.get_transport_from_url('memory:///foo/bar/'),
 
475
            transport.get_transport('memory:///foo/bar/'),
476
476
            lambda x: x)
477
477
        server.start_server()
478
 
        t = transport.get_transport_from_url(server.get_url())
 
478
        t = transport.get_transport(server.get_url())
479
479
        self.assertEqual(server.get_url(), t.abspath('/'))
480
480
 
481
481
        subdir_t = t.clone('subdir')
484
484
 
485
485
    def make_pf_transport(self, filter_func=None):
486
486
        """Make a PathFilteringTransport backed by a MemoryTransport.
487
 
 
 
487
        
488
488
        :param filter_func: by default this will be a no-op function.  Use this
489
489
            parameter to override it."""
490
490
        if filter_func is None:
491
491
            filter_func = lambda x: x
492
492
        server = pathfilter.PathFilteringServer(
493
 
            transport.get_transport_from_url('memory:///foo/bar/'), filter_func)
 
493
            transport.get_transport('memory:///foo/bar/'), filter_func)
494
494
        server.start_server()
495
495
        self.addCleanup(server.stop_server)
496
 
        return transport.get_transport_from_url(server.get_url())
 
496
        return transport.get_transport(server.get_url())
497
497
 
498
498
    def test__filter(self):
499
499
        # _filter (with an identity func as filter_func) always returns
512
512
 
513
513
    def test_filter_invocation(self):
514
514
        filter_log = []
515
 
 
516
515
        def filter(path):
517
516
            filter_log.append(path)
518
517
            return path
543
542
        otherwise) the filtering by doing::
544
543
            url = filtered_transport.base
545
544
            parent_url = urlutils.join(url, '..')
546
 
            new_t = transport.get_transport_from_url(parent_url)
 
545
            new_t = transport.get_transport(parent_url)
547
546
        """
548
547
        t = self.make_pf_transport()
549
 
        new_t = transport.get_transport_from_url(t.base)
 
548
        new_t = transport.get_transport(t.base)
550
549
        self.assertEqual(t.server, new_t.server)
551
550
        self.assertEqual(t.base, new_t.base)
552
551
 
565
564
        # connect to '.' via http which is not listable
566
565
        server = HttpServer()
567
566
        self.start_server(server)
568
 
        t = transport.get_transport_from_url('readonly+' + server.get_url())
 
567
        t = transport.get_transport('readonly+' + server.get_url())
569
568
        self.assertIsInstance(t, readonly.ReadonlyTransportDecorator)
570
569
        self.assertEqual(False, t.listable())
571
570
        self.assertEqual(True, t.is_readonly())
604
603
        # the url should be decorated appropriately
605
604
        self.assertStartsWith(server.get_url(), 'fakenfs+')
606
605
        # and we should be able to get a transport for it
607
 
        t = transport.get_transport_from_url(server.get_url())
 
606
        t = transport.get_transport(server.get_url())
608
607
        # which must be a FakeNFSTransportDecorator instance.
609
608
        self.assertIsInstance(t, fakenfs.FakeNFSTransportDecorator)
610
609
 
687
686
        base_url = self._server.get_url()
688
687
        url = self._adjust_url(base_url, relpath)
689
688
        # try getting the transport via the regular interface:
690
 
        t = transport.get_transport_from_url(url)
 
689
        t = transport.get_transport(url)
691
690
        # vila--20070607 if the following are commented out the test suite
692
691
        # still pass. Is this really still needed or was it a forgotten
693
692
        # temporary fix ?
698
697
        return t
699
698
 
700
699
 
701
 
class TestTransportFromPath(tests.TestCaseInTempDir):
702
 
 
703
 
    def test_with_path(self):
704
 
        t = transport.get_transport_from_path(self.test_dir)
705
 
        self.assertIsInstance(t, local.LocalTransport)
706
 
        self.assertEquals(t.base.rstrip("/"),
707
 
            urlutils.local_path_to_url(self.test_dir))
708
 
 
709
 
    def test_with_url(self):
710
 
        t = transport.get_transport_from_path("file:")
711
 
        self.assertIsInstance(t, local.LocalTransport)
712
 
        self.assertEquals(t.base.rstrip("/"),
713
 
            urlutils.local_path_to_url(os.path.join(self.test_dir, "file:")))
714
 
 
715
 
 
716
 
class TestTransportFromUrl(tests.TestCaseInTempDir):
717
 
 
718
 
    def test_with_path(self):
719
 
        self.assertRaises(errors.InvalidURL, transport.get_transport_from_url,
720
 
            self.test_dir)
721
 
 
722
 
    def test_with_url(self):
723
 
        url = urlutils.local_path_to_url(self.test_dir)
724
 
        t = transport.get_transport_from_url(url)
725
 
        self.assertIsInstance(t, local.LocalTransport)
726
 
        self.assertEquals(t.base.rstrip("/"), url)
727
 
 
728
 
    def test_with_url_and_segment_parameters(self):
729
 
        url = urlutils.local_path_to_url(self.test_dir)+",branch=foo"
730
 
        t = transport.get_transport_from_url(url)
731
 
        self.assertIsInstance(t, local.LocalTransport)
732
 
        self.assertEquals(t.base.rstrip("/"), url)
733
 
        with open(os.path.join(self.test_dir, "afile"), 'w') as f:
734
 
            f.write("data")
735
 
        self.assertTrue(t.has("afile"))
736
 
 
737
 
 
738
700
class TestLocalTransports(tests.TestCase):
739
701
 
740
702
    def test_get_transport_from_abspath(self):
793
755
        We can't easily observe the external effect but we can at least see
794
756
        it's called.
795
757
        """
796
 
        sentinel = object()
797
 
        fdatasync = getattr(os, 'fdatasync', sentinel)
798
 
        if fdatasync is sentinel:
799
 
            raise tests.TestNotApplicable('fdatasync not supported')
800
758
        t = self.get_transport('.')
801
759
        calls = self.recordCalls(os, 'fdatasync')
802
760
        w = t.open_write_stream('out')
807
765
            self.assertEquals(f.read(), 'foo')
808
766
        self.assertEquals(len(calls), 1, calls)
809
767
 
810
 
    def test_missing_directory(self):
811
 
        t = self.get_transport('.')
812
 
        self.assertRaises(errors.NoSuchFile, t.open_write_stream, 'dir/foo')
813
 
 
814
768
 
815
769
class TestWin32LocalTransport(tests.TestCase):
816
770
 
833
787
    def test_parse_url(self):
834
788
        t = transport.ConnectedTransport(
835
789
            'http://simple.example.com/home/source')
836
 
        self.assertEquals(t._parsed_url.host, 'simple.example.com')
837
 
        self.assertEquals(t._parsed_url.port, None)
838
 
        self.assertEquals(t._parsed_url.path, '/home/source/')
839
 
        self.assertTrue(t._parsed_url.user is None)
840
 
        self.assertTrue(t._parsed_url.password is None)
 
790
        self.assertEquals(t._host, 'simple.example.com')
 
791
        self.assertEquals(t._port, None)
 
792
        self.assertEquals(t._path, '/home/source/')
 
793
        self.assertTrue(t._user is None)
 
794
        self.assertTrue(t._password is None)
841
795
 
842
796
        self.assertEquals(t.base, 'http://simple.example.com/home/source/')
843
797
 
844
798
    def test_parse_url_with_at_in_user(self):
845
799
        # Bug 228058
846
800
        t = transport.ConnectedTransport('ftp://user@host.com@www.host.com/')
847
 
        self.assertEquals(t._parsed_url.user, 'user@host.com')
 
801
        self.assertEquals(t._user, 'user@host.com')
848
802
 
849
803
    def test_parse_quoted_url(self):
850
804
        t = transport.ConnectedTransport(
851
805
            'http://ro%62ey:h%40t@ex%41mple.com:2222/path')
852
 
        self.assertEquals(t._parsed_url.host, 'exAmple.com')
853
 
        self.assertEquals(t._parsed_url.port, 2222)
854
 
        self.assertEquals(t._parsed_url.user, 'robey')
855
 
        self.assertEquals(t._parsed_url.password, 'h@t')
856
 
        self.assertEquals(t._parsed_url.path, '/path/')
 
806
        self.assertEquals(t._host, 'exAmple.com')
 
807
        self.assertEquals(t._port, 2222)
 
808
        self.assertEquals(t._user, 'robey')
 
809
        self.assertEquals(t._password, 'h@t')
 
810
        self.assertEquals(t._path, '/path/')
857
811
 
858
812
        # Base should not keep track of the password
859
 
        self.assertEquals(t.base, 'http://ro%62ey@ex%41mple.com:2222/path/')
 
813
        self.assertEquals(t.base, 'http://robey@exAmple.com:2222/path/')
860
814
 
861
815
    def test_parse_invalid_url(self):
862
816
        self.assertRaises(errors.InvalidURL,
866
820
    def test_relpath(self):
867
821
        t = transport.ConnectedTransport('sftp://user@host.com/abs/path')
868
822
 
869
 
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'),
870
 
            'sub')
 
823
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
871
824
        self.assertRaises(errors.PathNotChild, t.relpath,
872
825
                          'http://user@host.com/abs/path/sub')
873
826
        self.assertRaises(errors.PathNotChild, t.relpath,
886
839
 
887
840
    def test_connection_sharing_propagate_credentials(self):
888
841
        t = transport.ConnectedTransport('ftp://user@host.com/abs/path')
889
 
        self.assertEquals('user', t._parsed_url.user)
890
 
        self.assertEquals('host.com', t._parsed_url.host)
 
842
        self.assertEquals('user', t._user)
 
843
        self.assertEquals('host.com', t._host)
891
844
        self.assertIs(None, t._get_connection())
892
 
        self.assertIs(None, t._parsed_url.password)
 
845
        self.assertIs(None, t._password)
893
846
        c = t.clone('subdir')
894
847
        self.assertIs(None, c._get_connection())
895
 
        self.assertIs(None, t._parsed_url.password)
 
848
        self.assertIs(None, t._password)
896
849
 
897
850
        # Simulate the user entering a password
898
851
        password = 'secret'
917
870
 
918
871
    def test_reuse_same_transport(self):
919
872
        possible_transports = []
920
 
        t1 = transport.get_transport_from_url('http://foo/',
 
873
        t1 = transport.get_transport('http://foo/',
921
874
                                     possible_transports=possible_transports)
922
875
        self.assertEqual([t1], possible_transports)
923
 
        t2 = transport.get_transport_from_url('http://foo/',
 
876
        t2 = transport.get_transport('http://foo/',
924
877
                                     possible_transports=[t1])
925
878
        self.assertIs(t1, t2)
926
879
 
927
880
        # Also check that final '/' are handled correctly
928
 
        t3 = transport.get_transport_from_url('http://foo/path/')
929
 
        t4 = transport.get_transport_from_url('http://foo/path',
 
881
        t3 = transport.get_transport('http://foo/path/')
 
882
        t4 = transport.get_transport('http://foo/path',
930
883
                                     possible_transports=[t3])
931
884
        self.assertIs(t3, t4)
932
885
 
933
 
        t5 = transport.get_transport_from_url('http://foo/path')
934
 
        t6 = transport.get_transport_from_url('http://foo/path/',
 
886
        t5 = transport.get_transport('http://foo/path')
 
887
        t6 = transport.get_transport('http://foo/path/',
935
888
                                     possible_transports=[t5])
936
889
        self.assertIs(t5, t6)
937
890
 
938
891
    def test_don_t_reuse_different_transport(self):
939
 
        t1 = transport.get_transport_from_url('http://foo/path')
940
 
        t2 = transport.get_transport_from_url('http://bar/path',
 
892
        t1 = transport.get_transport('http://foo/path')
 
893
        t2 = transport.get_transport('http://bar/path',
941
894
                                     possible_transports=[t1])
942
895
        self.assertIsNot(t1, t2)
943
896
 
944
897
 
945
898
class TestTransportTrace(tests.TestCase):
946
899
 
947
 
    def test_decorator(self):
948
 
        t = transport.get_transport_from_url('trace+memory://')
949
 
        self.assertIsInstance(
950
 
            t, bzrlib.transport.trace.TransportTraceDecorator)
 
900
    def test_get(self):
 
901
        t = transport.get_transport('trace+memory://')
 
902
        self.assertIsInstance(t, bzrlib.transport.trace.TransportTraceDecorator)
951
903
 
952
904
    def test_clone_preserves_activity(self):
953
 
        t = transport.get_transport_from_url('trace+memory://')
 
905
        t = transport.get_transport('trace+memory://')
954
906
        t2 = t.clone('.')
955
907
        self.assertTrue(t is not t2)
956
908
        self.assertTrue(t._activity is t2._activity)
960
912
    # still won't cause a test failure when the top level Transport API
961
913
    # changes; so there is little return doing that.
962
914
    def test_get(self):
963
 
        t = transport.get_transport_from_url('trace+memory:///')
 
915
        t = transport.get_transport('trace+memory:///')
964
916
        t.put_bytes('foo', 'barish')
965
917
        t.get('foo')
966
918
        expected_result = []
972
924
        self.assertEqual(expected_result, t._activity)
973
925
 
974
926
    def test_readv(self):
975
 
        t = transport.get_transport_from_url('trace+memory:///')
 
927
        t = transport.get_transport('trace+memory:///')
976
928
        t.put_bytes('foo', 'barish')
977
929
        list(t.readv('foo', [(0, 1), (3, 2)],
978
930
                     adjust_for_latency=True, upper_limit=6))
988
940
class TestSSHConnections(tests.TestCaseWithTransport):
989
941
 
990
942
    def test_bzr_connect_to_bzr_ssh(self):
991
 
        """get_transport of a bzr+ssh:// behaves correctly.
 
943
        """User acceptance that get_transport of a bzr+ssh:// behaves correctly.
992
944
 
993
945
        bzr+ssh:// should cause bzr to run a remote bzr smart server over SSH.
994
946
        """
1010
962
        # SSH channel ourselves.  Surely this has already been implemented
1011
963
        # elsewhere?
1012
964
        started = []
1013
 
 
1014
965
        class StubSSHServer(stub_sftp.StubServer):
1015
966
 
1016
967
            test = self
1022
973
                    stdout=subprocess.PIPE, stderr=subprocess.PIPE)
1023
974
 
1024
975
                # XXX: horribly inefficient, not to mention ugly.
1025
 
                # Start a thread for each of stdin/out/err, and relay bytes
1026
 
                # from the subprocess to channel and vice versa.
 
976
                # Start a thread for each of stdin/out/err, and relay bytes from
 
977
                # the subprocess to channel and vice versa.
1027
978
                def ferry_bytes(read, write, close):
1028
979
                    while True:
1029
980
                        bytes = read(1)
1097
1048
        result = http.unhtml_roughly(fake_html)
1098
1049
        self.assertEquals(len(result), 1000)
1099
1050
        self.assertStartsWith(result, " something!")
1100
 
 
1101
 
 
1102
 
class SomeDirectory(object):
1103
 
 
1104
 
    def look_up(self, name, url):
1105
 
        return "http://bar"
1106
 
 
1107
 
 
1108
 
class TestLocationToUrl(tests.TestCase):
1109
 
 
1110
 
    def get_base_location(self):
1111
 
        path = osutils.abspath('/foo/bar')
1112
 
        if path.startswith('/'):
1113
 
            url = 'file://%s' % (path,)
1114
 
        else:
1115
 
            # On Windows, abspaths start with the drive letter, so we have to
1116
 
            # add in the extra '/'
1117
 
            url = 'file:///%s' % (path,)
1118
 
        return path, url
1119
 
 
1120
 
    def test_regular_url(self):
1121
 
        self.assertEquals("file://foo", location_to_url("file://foo"))
1122
 
 
1123
 
    def test_directory(self):
1124
 
        directories.register("bar:", SomeDirectory, "Dummy directory")
1125
 
        self.addCleanup(directories.remove, "bar:")
1126
 
        self.assertEquals("http://bar", location_to_url("bar:"))
1127
 
 
1128
 
    def test_unicode_url(self):
1129
 
        self.assertRaises(errors.InvalidURL, location_to_url,
1130
 
            "http://fo/\xc3\xaf".decode("utf-8"))
1131
 
 
1132
 
    def test_unicode_path(self):
1133
 
        path, url = self.get_base_location()
1134
 
        location = path + "\xc3\xaf".decode("utf-8")
1135
 
        url += '%C3%AF'
1136
 
        self.assertEquals(url, location_to_url(location))
1137
 
 
1138
 
    def test_path(self):
1139
 
        path, url = self.get_base_location()
1140
 
        self.assertEquals(url, location_to_url(path))
1141
 
 
1142
 
    def test_relative_file_url(self):
1143
 
        self.assertEquals(urlutils.local_path_to_url(".") + "/bar",
1144
 
            location_to_url("file:bar"))
1145
 
 
1146
 
    def test_absolute_file_url(self):
1147
 
        self.assertEquals("file:///bar", location_to_url("file:/bar"))