59
56
def test__get_set_protocol_handlers(self):
60
57
handlers = _get_protocol_handlers()
61
self.assertNotEqual([], handlers.keys( ))
58
self.assertNotEqual({}, handlers)
63
_clear_protocol_handlers()
64
self.assertEqual([], _get_protocol_handlers().keys())
60
_set_protocol_handlers({})
61
self.assertEqual({}, _get_protocol_handlers())
66
63
_set_protocol_handlers(handlers)
68
65
def test_get_transport_modules(self):
69
66
handlers = _get_protocol_handlers()
70
# don't pollute the current handlers
71
_clear_protocol_handlers()
72
67
class SampleHandler(object):
73
68
"""I exist, isnt that enough?"""
75
_clear_protocol_handlers()
76
register_transport_proto('foo')
77
register_lazy_transport('foo', 'bzrlib.tests.test_transport',
78
'TestTransport.SampleHandler')
79
register_transport_proto('bar')
80
register_lazy_transport('bar', 'bzrlib.tests.test_transport',
81
'TestTransport.SampleHandler')
82
self.assertEqual([SampleHandler.__module__,
83
'bzrlib.transport.chroot'],
71
_set_protocol_handlers(my_handlers)
72
register_lazy_transport('foo', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
73
register_lazy_transport('bar', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
74
self.assertEqual([SampleHandler.__module__, 'bzrlib.transport.chroot'],
84
75
_get_transport_modules())
86
77
_set_protocol_handlers(handlers)
210
179
], [(10, 10), (30, 10), (100, 10)],
213
def test_coalesce_max_size(self):
214
self.check([(10, 20, [(0, 10), (10, 10)]),
216
# If one range is above max_size, it gets its own coalesced
218
(100, 80, [(0, 80),]),],
219
[(10, 10), (20, 10), (30, 50), (100, 80)],
223
def test_coalesce_no_max_size(self):
224
self.check([(10, 170, [(0, 10), (10, 10), (20, 50), (70, 100)]),],
225
[(10, 10), (20, 10), (30, 50), (80, 100)],
228
def test_coalesce_default_limit(self):
229
# 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)]),
232
(10*ten_mb, ten_mb, [(0, ten_mb)])],
233
[(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)],
236
max_size=1*1024*1024*1024)
239
184
class TestMemoryTransport(TestCase):
577
527
self._server.setUp()
578
528
self.addCleanup(self._server.tearDown)
580
def get_transport(self, relpath=None):
581
"""Return a connected transport to the local directory.
583
:param relpath: a path relative to the base url.
530
def get_transport(self):
531
"""Return a connected transport to the local directory."""
585
532
base_url = self._server.get_url()
586
url = self._adjust_url(base_url, relpath)
587
533
# try getting the transport via the regular interface:
588
t = get_transport(url)
589
# vila--20070607 if the following are commented out the test suite
590
# still pass. Is this really still needed or was it a forgotten
534
t = get_transport(base_url)
592
535
if not isinstance(t, self.transport_class):
593
536
# we did not get the correct transport class type. Override the
594
537
# regular connection behaviour by direct construction.
595
t = self.transport_class(url)
538
t = self.transport_class(base_url)
599
542
class TestLocalTransports(TestCase):
601
544
def test_get_transport_from_abspath(self):
602
here = osutils.abspath('.')
545
here = os.path.abspath('.')
603
546
t = get_transport(here)
604
547
self.assertIsInstance(t, LocalTransport)
605
548
self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
607
550
def test_get_transport_from_relpath(self):
608
here = osutils.abspath('.')
551
here = os.path.abspath('.')
609
552
t = get_transport('.')
610
553
self.assertIsInstance(t, LocalTransport)
611
554
self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
613
556
def test_get_transport_from_local_url(self):
614
here = osutils.abspath('.')
557
here = os.path.abspath('.')
615
558
here_url = urlutils.local_path_to_url(here) + '/'
616
559
t = get_transport(here_url)
617
560
self.assertIsInstance(t, LocalTransport)
618
561
self.assertEquals(t.base, here_url)
620
def test_local_abspath(self):
621
here = osutils.abspath('.')
622
t = get_transport(here)
623
self.assertEquals(t.local_abspath(''), here)
626
564
class TestWin32LocalTransport(TestCase):
636
574
# make sure we reach the root
637
575
t = t.clone('..')
638
576
self.assertEquals(t.base, 'file://HOST/')
641
class TestConnectedTransport(TestCase):
642
"""Tests for connected to remote server transports"""
644
def test_parse_url(self):
645
t = ConnectedTransport('http://simple.example.com/home/source')
646
self.assertEquals(t._host, 'simple.example.com')
647
self.assertEquals(t._port, None)
648
self.assertEquals(t._path, '/home/source/')
649
self.failUnless(t._user is None)
650
self.failUnless(t._password is None)
652
self.assertEquals(t.base, 'http://simple.example.com/home/source/')
654
def test_parse_url_with_at_in_user(self):
656
t = ConnectedTransport('ftp://user@host.com@www.host.com/')
657
self.assertEquals(t._user, 'user@host.com')
659
def test_parse_quoted_url(self):
660
t = ConnectedTransport('http://ro%62ey:h%40t@ex%41mple.com:2222/path')
661
self.assertEquals(t._host, 'exAmple.com')
662
self.assertEquals(t._port, 2222)
663
self.assertEquals(t._user, 'robey')
664
self.assertEquals(t._password, 'h@t')
665
self.assertEquals(t._path, '/path/')
667
# Base should not keep track of the password
668
self.assertEquals(t.base, 'http://robey@exAmple.com:2222/path/')
670
def test_parse_invalid_url(self):
671
self.assertRaises(errors.InvalidURL,
673
'sftp://lily.org:~janneke/public/bzr/gub')
675
def test_relpath(self):
676
t = ConnectedTransport('sftp://user@host.com/abs/path')
678
self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
679
self.assertRaises(errors.PathNotChild, t.relpath,
680
'http://user@host.com/abs/path/sub')
681
self.assertRaises(errors.PathNotChild, t.relpath,
682
'sftp://user2@host.com/abs/path/sub')
683
self.assertRaises(errors.PathNotChild, t.relpath,
684
'sftp://user@otherhost.com/abs/path/sub')
685
self.assertRaises(errors.PathNotChild, t.relpath,
686
'sftp://user@host.com:33/abs/path/sub')
687
# Make sure it works when we don't supply a username
688
t = ConnectedTransport('sftp://host.com/abs/path')
689
self.assertEquals(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
691
# Make sure it works when parts of the path will be url encoded
692
t = ConnectedTransport('sftp://host.com/dev/%path')
693
self.assertEquals(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
695
def test_connection_sharing_propagate_credentials(self):
696
t = ConnectedTransport('ftp://user@host.com/abs/path')
697
self.assertEquals('user', t._user)
698
self.assertEquals('host.com', t._host)
699
self.assertIs(None, t._get_connection())
700
self.assertIs(None, t._password)
701
c = t.clone('subdir')
702
self.assertIs(None, c._get_connection())
703
self.assertIs(None, t._password)
705
# Simulate the user entering a password
707
connection = object()
708
t._set_connection(connection, password)
709
self.assertIs(connection, t._get_connection())
710
self.assertIs(password, t._get_credentials())
711
self.assertIs(connection, c._get_connection())
712
self.assertIs(password, c._get_credentials())
714
# credentials can be updated
715
new_password = 'even more secret'
716
c._update_credentials(new_password)
717
self.assertIs(connection, t._get_connection())
718
self.assertIs(new_password, t._get_credentials())
719
self.assertIs(connection, c._get_connection())
720
self.assertIs(new_password, c._get_credentials())
723
class TestReusedTransports(TestCase):
724
"""Tests for transport reuse"""
726
def test_reuse_same_transport(self):
727
possible_transports = []
728
t1 = get_transport('http://foo/',
729
possible_transports=possible_transports)
730
self.assertEqual([t1], possible_transports)
731
t2 = get_transport('http://foo/', possible_transports=[t1])
732
self.assertIs(t1, t2)
734
# Also check that final '/' are handled correctly
735
t3 = get_transport('http://foo/path/')
736
t4 = get_transport('http://foo/path', possible_transports=[t3])
737
self.assertIs(t3, t4)
739
t5 = get_transport('http://foo/path')
740
t6 = get_transport('http://foo/path/', possible_transports=[t5])
741
self.assertIs(t5, t6)
743
def test_don_t_reuse_different_transport(self):
744
t1 = get_transport('http://foo/path')
745
t2 = get_transport('http://bar/path', possible_transports=[t1])
746
self.assertIsNot(t1, t2)
749
class TestTransportTrace(TestCase):
752
transport = get_transport('trace+memory://')
753
self.assertIsInstance(
754
transport, bzrlib.transport.trace.TransportTraceDecorator)
756
def test_clone_preserves_activity(self):
757
transport = get_transport('trace+memory://')
758
transport2 = transport.clone('.')
759
self.assertTrue(transport is not transport2)
760
self.assertTrue(transport._activity is transport2._activity)
762
# the following specific tests are for the operations that have made use of
763
# logging in tests; we could test every single operation but doing that
764
# still won't cause a test failure when the top level Transport API
765
# changes; so there is little return doing that.
767
transport = get_transport('trace+memory:///')
768
transport.put_bytes('foo', 'barish')
771
# put_bytes records the bytes, not the content to avoid memory
773
expected_result.append(('put_bytes', 'foo', 6, None))
774
# get records the file name only.
775
expected_result.append(('get', 'foo'))
776
self.assertEqual(expected_result, transport._activity)
778
def test_readv(self):
779
transport = get_transport('trace+memory:///')
780
transport.put_bytes('foo', 'barish')
781
list(transport.readv('foo', [(0, 1), (3, 2)], adjust_for_latency=True,
784
# put_bytes records the bytes, not the content to avoid memory
786
expected_result.append(('put_bytes', 'foo', 6, None))
787
# readv records the supplied offset request
788
expected_result.append(('readv', 'foo', [(0, 1), (3, 2)], True, 6))
789
self.assertEqual(expected_result, transport._activity)
added
removed
bzrlib/tests/test_transport.py
