48
52
class TestTransport(tests.TestCase):
49
53
"""Test the non transport-concrete class functionality."""
51
# FIXME: These tests should use addCleanup() and/or overrideAttr() instead
52
# of try/finally -- vila 20100205
54
55
def test__get_set_protocol_handlers(self):
55
56
handlers = transport._get_protocol_handlers()
56
self.assertNotEqual([], handlers.keys( ))
58
transport._clear_protocol_handlers()
59
self.assertEqual([], transport._get_protocol_handlers().keys())
61
transport._set_protocol_handlers(handlers)
57
self.assertNotEqual([], handlers.keys())
58
transport._clear_protocol_handlers()
59
self.addCleanup(transport._set_protocol_handlers, handlers)
60
self.assertEqual([], transport._get_protocol_handlers().keys())
63
62
def test_get_transport_modules(self):
64
63
handlers = transport._get_protocol_handlers()
64
self.addCleanup(transport._set_protocol_handlers, handlers)
65
65
# don't pollute the current handlers
66
66
transport._clear_protocol_handlers()
67
68
class SampleHandler(object):
68
69
"""I exist, isnt that enough?"""
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())
84
transport._set_protocol_handlers(handlers)
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())
86
84
def test_transport_dependency(self):
87
85
"""Transport with missing dependency causes no error"""
88
86
saved_handlers = transport._get_protocol_handlers()
87
self.addCleanup(transport._set_protocol_handlers, saved_handlers)
89
88
# don't pollute the current handlers
90
89
transport._clear_protocol_handlers()
90
transport.register_transport_proto('foo')
91
transport.register_lazy_transport(
92
'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
92
transport.register_transport_proto('foo')
93
transport.register_lazy_transport(
94
'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
96
transport.get_transport('foo://fooserver/foo')
97
except errors.UnsupportedProtocol, 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))
104
self.fail('Did not raise UnsupportedProtocol')
106
# restore original values
107
transport._set_protocol_handlers(saved_handlers)
94
transport.get_transport_from_url('foo://fooserver/foo')
95
except errors.UnsupportedProtocol, e:
97
self.assertEquals('Unsupported protocol'
98
' for url "foo://fooserver/foo":'
99
' Unable to import library "some_lib":'
100
' testing missing dependency', str(e))
102
self.fail('Did not raise UnsupportedProtocol')
109
104
def test_transport_fallback(self):
110
105
"""Transport with missing dependency causes no error"""
111
106
saved_handlers = transport._get_protocol_handlers()
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))
122
transport._set_protocol_handlers(saved_handlers)
107
self.addCleanup(transport._set_protocol_handlers, saved_handlers)
108
transport._clear_protocol_handlers()
109
transport.register_transport_proto('foo')
110
transport.register_lazy_transport(
111
'foo', 'bzrlib.tests.test_transport', 'BackupTransportHandler')
112
transport.register_lazy_transport(
113
'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
114
t = transport.get_transport_from_url('foo://fooserver/foo')
115
self.assertTrue(isinstance(t, BackupTransportHandler))
124
117
def test_ssh_hints(self):
125
118
"""Transport ssh:// should raise an error pointing out bzr+ssh://"""
127
transport.get_transport('ssh://fooserver/foo')
120
transport.get_transport_from_url('ssh://fooserver/foo')
128
121
except errors.UnsupportedProtocol, e:
130
123
self.assertEquals('Unsupported protocol'
219
201
def test_coalesce_fudge(self):
220
202
self.check([(10, 30, [(0, 10), (20, 10)]),
221
(100, 10, [(0, 10),]),
203
(100, 10, [(0, 10)]),
222
204
], [(10, 10), (30, 10), (100, 10)],
225
207
def test_coalesce_max_size(self):
226
208
self.check([(10, 20, [(0, 10), (10, 10)]),
227
209
(30, 50, [(0, 50)]),
228
210
# If one range is above max_size, it gets its own coalesced
230
(100, 80, [(0, 80),]),],
212
(100, 80, [(0, 80)]),],
231
213
[(10, 10), (20, 10), (30, 50), (100, 80)],
235
216
def test_coalesce_no_max_size(self):
236
self.check([(10, 170, [(0, 10), (10, 10), (20, 50), (70, 100)]),],
217
self.check([(10, 170, [(0, 10), (10, 10), (20, 50), (70, 100)])],
237
218
[(10, 10), (20, 10), (30, 50), (80, 100)],
240
221
def test_coalesce_default_limit(self):
241
222
# 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)]),
223
ten_mb = 10 * 1024 * 1024
224
self.check([(0, 10 * ten_mb, [(i * ten_mb, ten_mb) for i in range(10)]),
244
225
(10*ten_mb, ten_mb, [(0, ten_mb)])],
245
226
[(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)],
227
self.check([(0, 11 * ten_mb, [(i * ten_mb, ten_mb) for i in range(11)])],
228
[(i * ten_mb, ten_mb) for i in range(11)],
248
229
max_size=1*1024*1024*1024)
677
class TestTransportFromPath(tests.TestCaseInTempDir):
679
def test_with_path(self):
680
t = transport.get_transport_from_path(self.test_dir)
681
self.assertIsInstance(t, local.LocalTransport)
682
self.assertEquals(t.base.rstrip("/"),
683
urlutils.local_path_to_url(self.test_dir))
685
def test_with_url(self):
686
t = transport.get_transport_from_path("file:")
687
self.assertIsInstance(t, local.LocalTransport)
688
self.assertEquals(t.base.rstrip("/"),
689
urlutils.local_path_to_url(os.path.join(self.test_dir, "file:")))
692
class TestTransportFromUrl(tests.TestCaseInTempDir):
694
def test_with_path(self):
695
self.assertRaises(errors.InvalidURL, transport.get_transport_from_url,
698
def test_with_url(self):
699
url = urlutils.local_path_to_url(self.test_dir)
700
t = transport.get_transport_from_url(url)
701
self.assertIsInstance(t, local.LocalTransport)
702
self.assertEquals(t.base.rstrip("/"), url)
698
705
class TestLocalTransports(tests.TestCase):
700
707
def test_get_transport_from_abspath(self):
743
773
def test_parse_url(self):
744
774
t = transport.ConnectedTransport(
745
775
'http://simple.example.com/home/source')
746
self.assertEquals(t._host, 'simple.example.com')
747
self.assertEquals(t._port, None)
748
self.assertEquals(t._path, '/home/source/')
749
self.assertTrue(t._user is None)
750
self.assertTrue(t._password is None)
776
self.assertEquals(t._parsed_url.host, 'simple.example.com')
777
self.assertEquals(t._parsed_url.port, None)
778
self.assertEquals(t._parsed_url.path, '/home/source/')
779
self.assertTrue(t._parsed_url.user is None)
780
self.assertTrue(t._parsed_url.password is None)
752
782
self.assertEquals(t.base, 'http://simple.example.com/home/source/')
754
784
def test_parse_url_with_at_in_user(self):
756
786
t = transport.ConnectedTransport('ftp://user@host.com@www.host.com/')
757
self.assertEquals(t._user, 'user@host.com')
787
self.assertEquals(t._parsed_url.user, 'user@host.com')
759
789
def test_parse_quoted_url(self):
760
790
t = transport.ConnectedTransport(
761
791
'http://ro%62ey:h%40t@ex%41mple.com:2222/path')
762
self.assertEquals(t._host, 'exAmple.com')
763
self.assertEquals(t._port, 2222)
764
self.assertEquals(t._user, 'robey')
765
self.assertEquals(t._password, 'h@t')
766
self.assertEquals(t._path, '/path/')
792
self.assertEquals(t._parsed_url.host, 'exAmple.com')
793
self.assertEquals(t._parsed_url.port, 2222)
794
self.assertEquals(t._parsed_url.user, 'robey')
795
self.assertEquals(t._parsed_url.password, 'h@t')
796
self.assertEquals(t._parsed_url.path, '/path/')
768
798
# Base should not keep track of the password
769
799
self.assertEquals(t.base, 'http://robey@exAmple.com:2222/path/')
827
858
def test_reuse_same_transport(self):
828
859
possible_transports = []
829
t1 = transport.get_transport('http://foo/',
860
t1 = transport.get_transport_from_url('http://foo/',
830
861
possible_transports=possible_transports)
831
862
self.assertEqual([t1], possible_transports)
832
t2 = transport.get_transport('http://foo/',
863
t2 = transport.get_transport_from_url('http://foo/',
833
864
possible_transports=[t1])
834
865
self.assertIs(t1, t2)
836
867
# Also check that final '/' are handled correctly
837
t3 = transport.get_transport('http://foo/path/')
838
t4 = transport.get_transport('http://foo/path',
868
t3 = transport.get_transport_from_url('http://foo/path/')
869
t4 = transport.get_transport_from_url('http://foo/path',
839
870
possible_transports=[t3])
840
871
self.assertIs(t3, t4)
842
t5 = transport.get_transport('http://foo/path')
843
t6 = transport.get_transport('http://foo/path/',
873
t5 = transport.get_transport_from_url('http://foo/path')
874
t6 = transport.get_transport_from_url('http://foo/path/',
844
875
possible_transports=[t5])
845
876
self.assertIs(t5, t6)
847
878
def test_don_t_reuse_different_transport(self):
848
t1 = transport.get_transport('http://foo/path')
849
t2 = transport.get_transport('http://bar/path',
879
t1 = transport.get_transport_from_url('http://foo/path')
880
t2 = transport.get_transport_from_url('http://bar/path',
850
881
possible_transports=[t1])
851
882
self.assertIsNot(t1, t2)
854
885
class TestTransportTrace(tests.TestCase):
857
t = transport.get_transport('trace+memory://')
858
self.assertIsInstance(t, bzrlib.transport.trace.TransportTraceDecorator)
887
def test_decorator(self):
888
t = transport.get_transport_from_url('trace+memory://')
889
self.assertIsInstance(
890
t, bzrlib.transport.trace.TransportTraceDecorator)
860
892
def test_clone_preserves_activity(self):
861
t = transport.get_transport('trace+memory://')
893
t = transport.get_transport_from_url('trace+memory://')
862
894
t2 = t.clone('.')
863
895
self.assertTrue(t is not t2)
864
896
self.assertTrue(t._activity is t2._activity)
1004
1037
result = http.unhtml_roughly(fake_html)
1005
1038
self.assertEquals(len(result), 1000)
1006
1039
self.assertStartsWith(result, " something!")
1042
class SomeDirectory(object):
1044
def look_up(self, name, url):
1048
class TestLocationToUrl(tests.TestCase):
1050
def get_base_location(self):
1051
path = osutils.abspath('/foo/bar')
1052
if path.startswith('/'):
1053
url = 'file://%s' % (path,)
1055
# On Windows, abspaths start with the drive letter, so we have to
1056
# add in the extra '/'
1057
url = 'file:///%s' % (path,)
1060
def test_regular_url(self):
1061
self.assertEquals("file://foo", location_to_url("file://foo"))
1063
def test_directory(self):
1064
directories.register("bar:", SomeDirectory, "Dummy directory")
1065
self.addCleanup(directories.remove, "bar:")
1066
self.assertEquals("http://bar", location_to_url("bar:"))
1068
def test_unicode_url(self):
1069
self.assertRaises(errors.InvalidURL, location_to_url,
1070
"http://fo/\xc3\xaf".decode("utf-8"))
1072
def test_unicode_path(self):
1073
path, url = self.get_base_location()
1074
location = path + "\xc3\xaf".decode("utf-8")
1076
self.assertEquals(url, location_to_url(location))
1078
def test_path(self):
1079
path, url = self.get_base_location()
1080
self.assertEquals(url, location_to_url(path))
1082
def test_relative_file_url(self):
1083
self.assertEquals(urlutils.local_path_to_url(".") + "/bar",
1084
location_to_url("file:bar"))
1086
def test_absolute_file_url(self):
1087
self.assertEquals("file:///bar", location_to_url("file:/bar"))