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)
223
179
], [(10, 10), (30, 10), (100, 10)],
226
def test_coalesce_max_size(self):
227
self.check([(10, 20, [(0, 10), (10, 10)]),
229
# If one range is above max_size, it gets its own coalesced
231
(100, 80, [(0, 80),]),],
232
[(10, 10), (20, 10), (30, 50), (100, 80)],
236
def test_coalesce_no_max_size(self):
237
self.check([(10, 170, [(0, 10), (10, 10), (20, 50), (70, 100)]),],
238
[(10, 10), (20, 10), (30, 50), (80, 100)],
241
def test_coalesce_default_limit(self):
242
# By default we use a 100MB max size.
243
ten_mb = 10*1024*1024
244
self.check([(0, 10*ten_mb, [(i*ten_mb, ten_mb) for i in range(10)]),
245
(10*ten_mb, ten_mb, [(0, ten_mb)])],
246
[(i*ten_mb, ten_mb) for i in range(11)])
247
self.check([(0, 11*ten_mb, [(i*ten_mb, ten_mb) for i in range(11)]),],
248
[(i*ten_mb, ten_mb) for i in range(11)],
249
max_size=1*1024*1024*1024)
252
184
class TestMemoryTransport(TestCase):
565
502
class TestTransportImplementation(TestCaseInTempDir):
566
503
"""Implementation verification for transports.
568
505
To verify a transport we need a server factory, which is a callable
569
506
that accepts no parameters and returns an implementation of
570
507
bzrlib.transport.Server.
572
509
That Server is then used to construct transport instances and test
573
510
the transport via loopback activity.
575
Currently this assumes that the Transport object is connected to the
576
current working directory. So that whatever is done
577
through the transport, should show up in the working
512
Currently this assumes that the Transport object is connected to the
513
current working directory. So that whatever is done
514
through the transport, should show up in the working
578
515
directory, and vice-versa. This is a bug, because its possible to have
579
URL schemes which provide access to something that may not be
580
result in storage on the local disk, i.e. due to file system limits, or
516
URL schemes which provide access to something that may not be
517
result in storage on the local disk, i.e. due to file system limits, or
581
518
due to it being a database or some other non-filesystem tool.
583
520
This also tests to make sure that the functions work with both
584
521
generators and lists (assuming iter(list) is effectively a generator)
588
525
super(TestTransportImplementation, self).setUp()
589
526
self._server = self.transport_server()
590
527
self._server.setUp()
591
528
self.addCleanup(self._server.tearDown)
593
def get_transport(self, relpath=None):
594
"""Return a connected transport to the local directory.
596
:param relpath: a path relative to the base url.
530
def get_transport(self):
531
"""Return a connected transport to the local directory."""
598
532
base_url = self._server.get_url()
599
url = self._adjust_url(base_url, relpath)
600
533
# try getting the transport via the regular interface:
601
t = get_transport(url)
602
# vila--20070607 if the following are commented out the test suite
603
# still pass. Is this really still needed or was it a forgotten
534
t = get_transport(base_url)
605
535
if not isinstance(t, self.transport_class):
606
536
# we did not get the correct transport class type. Override the
607
537
# regular connection behaviour by direct construction.
608
t = self.transport_class(url)
538
t = self.transport_class(base_url)
612
542
class TestLocalTransports(TestCase):
614
544
def test_get_transport_from_abspath(self):
615
here = osutils.abspath('.')
545
here = os.path.abspath('.')
616
546
t = get_transport(here)
617
547
self.assertIsInstance(t, LocalTransport)
618
548
self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
620
550
def test_get_transport_from_relpath(self):
621
here = osutils.abspath('.')
551
here = os.path.abspath('.')
622
552
t = get_transport('.')
623
553
self.assertIsInstance(t, LocalTransport)
624
554
self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
626
556
def test_get_transport_from_local_url(self):
627
here = osutils.abspath('.')
557
here = os.path.abspath('.')
628
558
here_url = urlutils.local_path_to_url(here) + '/'
629
559
t = get_transport(here_url)
630
560
self.assertIsInstance(t, LocalTransport)
631
561
self.assertEquals(t.base, here_url)
633
def test_local_abspath(self):
634
here = osutils.abspath('.')
635
t = get_transport(here)
636
self.assertEquals(t.local_abspath(''), here)
639
564
class TestWin32LocalTransport(TestCase):
649
574
# make sure we reach the root
650
575
t = t.clone('..')
651
576
self.assertEquals(t.base, 'file://HOST/')
654
class TestConnectedTransport(TestCase):
655
"""Tests for connected to remote server transports"""
657
def test_parse_url(self):
658
t = ConnectedTransport('http://simple.example.com/home/source')
659
self.assertEquals(t._host, 'simple.example.com')
660
self.assertEquals(t._port, None)
661
self.assertEquals(t._path, '/home/source/')
662
self.failUnless(t._user is None)
663
self.failUnless(t._password is None)
665
self.assertEquals(t.base, 'http://simple.example.com/home/source/')
667
def test_parse_url_with_at_in_user(self):
669
t = ConnectedTransport('ftp://user@host.com@www.host.com/')
670
self.assertEquals(t._user, 'user@host.com')
672
def test_parse_quoted_url(self):
673
t = ConnectedTransport('http://ro%62ey:h%40t@ex%41mple.com:2222/path')
674
self.assertEquals(t._host, 'exAmple.com')
675
self.assertEquals(t._port, 2222)
676
self.assertEquals(t._user, 'robey')
677
self.assertEquals(t._password, 'h@t')
678
self.assertEquals(t._path, '/path/')
680
# Base should not keep track of the password
681
self.assertEquals(t.base, 'http://robey@exAmple.com:2222/path/')
683
def test_parse_invalid_url(self):
684
self.assertRaises(errors.InvalidURL,
686
'sftp://lily.org:~janneke/public/bzr/gub')
688
def test_relpath(self):
689
t = ConnectedTransport('sftp://user@host.com/abs/path')
691
self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
692
self.assertRaises(errors.PathNotChild, t.relpath,
693
'http://user@host.com/abs/path/sub')
694
self.assertRaises(errors.PathNotChild, t.relpath,
695
'sftp://user2@host.com/abs/path/sub')
696
self.assertRaises(errors.PathNotChild, t.relpath,
697
'sftp://user@otherhost.com/abs/path/sub')
698
self.assertRaises(errors.PathNotChild, t.relpath,
699
'sftp://user@host.com:33/abs/path/sub')
700
# Make sure it works when we don't supply a username
701
t = ConnectedTransport('sftp://host.com/abs/path')
702
self.assertEquals(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
704
# Make sure it works when parts of the path will be url encoded
705
t = ConnectedTransport('sftp://host.com/dev/%path')
706
self.assertEquals(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
708
def test_connection_sharing_propagate_credentials(self):
709
t = ConnectedTransport('ftp://user@host.com/abs/path')
710
self.assertEquals('user', t._user)
711
self.assertEquals('host.com', t._host)
712
self.assertIs(None, t._get_connection())
713
self.assertIs(None, t._password)
714
c = t.clone('subdir')
715
self.assertIs(None, c._get_connection())
716
self.assertIs(None, t._password)
718
# Simulate the user entering a password
720
connection = object()
721
t._set_connection(connection, password)
722
self.assertIs(connection, t._get_connection())
723
self.assertIs(password, t._get_credentials())
724
self.assertIs(connection, c._get_connection())
725
self.assertIs(password, c._get_credentials())
727
# credentials can be updated
728
new_password = 'even more secret'
729
c._update_credentials(new_password)
730
self.assertIs(connection, t._get_connection())
731
self.assertIs(new_password, t._get_credentials())
732
self.assertIs(connection, c._get_connection())
733
self.assertIs(new_password, c._get_credentials())
736
class TestReusedTransports(TestCase):
737
"""Tests for transport reuse"""
739
def test_reuse_same_transport(self):
740
possible_transports = []
741
t1 = get_transport('http://foo/',
742
possible_transports=possible_transports)
743
self.assertEqual([t1], possible_transports)
744
t2 = get_transport('http://foo/', possible_transports=[t1])
745
self.assertIs(t1, t2)
747
# Also check that final '/' are handled correctly
748
t3 = get_transport('http://foo/path/')
749
t4 = get_transport('http://foo/path', possible_transports=[t3])
750
self.assertIs(t3, t4)
752
t5 = get_transport('http://foo/path')
753
t6 = get_transport('http://foo/path/', possible_transports=[t5])
754
self.assertIs(t5, t6)
756
def test_don_t_reuse_different_transport(self):
757
t1 = get_transport('http://foo/path')
758
t2 = get_transport('http://bar/path', possible_transports=[t1])
759
self.assertIsNot(t1, t2)
762
class TestTransportTrace(TestCase):
765
transport = get_transport('trace+memory://')
766
self.assertIsInstance(
767
transport, bzrlib.transport.trace.TransportTraceDecorator)
769
def test_clone_preserves_activity(self):
770
transport = get_transport('trace+memory://')
771
transport2 = transport.clone('.')
772
self.assertTrue(transport is not transport2)
773
self.assertTrue(transport._activity is transport2._activity)
775
# the following specific tests are for the operations that have made use of
776
# logging in tests; we could test every single operation but doing that
777
# still won't cause a test failure when the top level Transport API
778
# changes; so there is little return doing that.
780
transport = get_transport('trace+memory:///')
781
transport.put_bytes('foo', 'barish')
784
# put_bytes records the bytes, not the content to avoid memory
786
expected_result.append(('put_bytes', 'foo', 6, None))
787
# get records the file name only.
788
expected_result.append(('get', 'foo'))
789
self.assertEqual(expected_result, transport._activity)
791
def test_readv(self):
792
transport = get_transport('trace+memory:///')
793
transport.put_bytes('foo', 'barish')
794
list(transport.readv('foo', [(0, 1), (3, 2)], adjust_for_latency=True,
797
# put_bytes records the bytes, not the content to avoid memory
799
expected_result.append(('put_bytes', 'foo', 6, None))
800
# readv records the supplied offset request
801
expected_result.append(('readv', 'foo', [(0, 1), (3, 2)], True, 6))
802
self.assertEqual(expected_result, transport._activity)
added
removed
bzrlib/tests/test_transport.py
