48
48
class TestTransport(tests.TestCase):
49
49
"""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
51
def test__get_set_protocol_handlers(self):
55
52
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)
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())
63
58
def test_get_transport_modules(self):
64
59
handlers = transport._get_protocol_handlers()
60
self.addCleanup(transport._set_protocol_handlers, handlers)
65
61
# don't pollute the current handlers
66
62
transport._clear_protocol_handlers()
67
64
class SampleHandler(object):
68
65
"""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)
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())
86
80
def test_transport_dependency(self):
87
81
"""Transport with missing dependency causes no error"""
88
82
saved_handlers = transport._get_protocol_handlers()
83
self.addCleanup(transport._set_protocol_handlers, saved_handlers)
89
84
# don't pollute the current handlers
90
85
transport._clear_protocol_handlers()
86
transport.register_transport_proto('foo')
87
transport.register_lazy_transport(
88
'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)
90
transport.get_transport('foo://fooserver/foo')
91
except errors.UnsupportedProtocol, 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))
98
self.fail('Did not raise UnsupportedProtocol')
109
100
def test_transport_fallback(self):
110
101
"""Transport with missing dependency causes no error"""
111
102
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)
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))
124
113
def test_ssh_hints(self):
125
114
"""Transport ssh:// should raise an error pointing out bzr+ssh://"""
219
208
def test_coalesce_fudge(self):
220
209
self.check([(10, 30, [(0, 10), (20, 10)]),
221
(100, 10, [(0, 10),]),
210
(100, 10, [(0, 10)]),
222
211
], [(10, 10), (30, 10), (100, 10)],
225
214
def test_coalesce_max_size(self):
226
215
self.check([(10, 20, [(0, 10), (10, 10)]),
227
216
(30, 50, [(0, 50)]),
228
217
# If one range is above max_size, it gets its own coalesced
230
(100, 80, [(0, 80),]),],
219
(100, 80, [(0, 80)]),],
231
220
[(10, 10), (20, 10), (30, 50), (100, 80)],
235
223
def test_coalesce_no_max_size(self):
236
self.check([(10, 170, [(0, 10), (10, 10), (20, 50), (70, 100)]),],
224
self.check([(10, 170, [(0, 10), (10, 10), (20, 50), (70, 100)])],
237
225
[(10, 10), (20, 10), (30, 50), (80, 100)],
240
228
def test_coalesce_default_limit(self):
241
229
# 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)]),
230
ten_mb = 10 * 1024 * 1024
231
self.check([(0, 10 * ten_mb, [(i * ten_mb, ten_mb) for i in range(10)]),
244
232
(10*ten_mb, ten_mb, [(0, ten_mb)])],
245
233
[(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)],
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)],
248
236
max_size=1*1024*1024*1024)