121
93
self.assertTrue(isinstance(t, BackupTransportHandler))
123
95
_set_protocol_handlers(saved_handlers)
125
def test_LateReadError(self):
126
"""The LateReadError helper should raise on read()."""
127
a_file = LateReadError('a path')
130
except ReadError, error:
131
self.assertEqual('a path', error.path)
132
self.assertRaises(ReadError, a_file.read, 40)
135
def test__combine_paths(self):
137
self.assertEqual('/home/sarah/project/foo',
138
t._combine_paths('/home/sarah', 'project/foo'))
139
self.assertEqual('/etc',
140
t._combine_paths('/home/sarah', '../../etc'))
141
self.assertEqual('/etc',
142
t._combine_paths('/home/sarah', '../../../etc'))
143
self.assertEqual('/etc',
144
t._combine_paths('/home/sarah', '/etc'))
146
def test_local_abspath_non_local_transport(self):
147
# the base implementation should throw
148
t = MemoryTransport()
149
e = self.assertRaises(errors.NotLocalUrl, t.local_abspath, 't')
150
self.assertEqual('memory:///t is not a local path.', str(e))
153
class TestCoalesceOffsets(TestCase):
155
def check(self, expected, offsets, limit=0, fudge=0):
156
coalesce = Transport._coalesce_offsets
157
exp = [_CoalescedOffset(*x) for x in expected]
158
out = list(coalesce(offsets, limit=limit, fudge_factor=fudge))
159
self.assertEqual(exp, out)
161
def test_coalesce_empty(self):
164
def test_coalesce_simple(self):
165
self.check([(0, 10, [(0, 10)])], [(0, 10)])
167
def test_coalesce_unrelated(self):
168
self.check([(0, 10, [(0, 10)]),
170
], [(0, 10), (20, 10)])
172
def test_coalesce_unsorted(self):
173
self.check([(20, 10, [(0, 10)]),
175
], [(20, 10), (0, 10)])
177
def test_coalesce_nearby(self):
178
self.check([(0, 20, [(0, 10), (10, 10)])],
181
def test_coalesce_overlapped(self):
182
self.check([(0, 15, [(0, 10), (5, 10)])],
185
def test_coalesce_limit(self):
186
self.check([(10, 50, [(0, 10), (10, 10), (20, 10),
187
(30, 10), (40, 10)]),
188
(60, 50, [(0, 10), (10, 10), (20, 10),
189
(30, 10), (40, 10)]),
190
], [(10, 10), (20, 10), (30, 10), (40, 10),
191
(50, 10), (60, 10), (70, 10), (80, 10),
192
(90, 10), (100, 10)],
195
def test_coalesce_no_limit(self):
196
self.check([(10, 100, [(0, 10), (10, 10), (20, 10),
197
(30, 10), (40, 10), (50, 10),
198
(60, 10), (70, 10), (80, 10),
200
], [(10, 10), (20, 10), (30, 10), (40, 10),
201
(50, 10), (60, 10), (70, 10), (80, 10),
202
(90, 10), (100, 10)])
204
def test_coalesce_fudge(self):
205
self.check([(10, 30, [(0, 10), (20, 10)]),
206
(100, 10, [(0, 10),]),
207
], [(10, 10), (30, 10), (100, 10)],
212
98
class TestMemoryTransport(TestCase):
217
103
def test_clone(self):
218
104
transport = MemoryTransport()
219
105
self.assertTrue(isinstance(transport, MemoryTransport))
220
self.assertEqual("memory:///", transport.clone("/").base)
222
107
def test_abspath(self):
223
108
transport = MemoryTransport()
224
109
self.assertEqual("memory:///relpath", transport.abspath('relpath'))
226
def test_abspath_of_root(self):
227
transport = MemoryTransport()
228
self.assertEqual("memory:///", transport.base)
229
self.assertEqual("memory:///", transport.abspath('/'))
231
def test_abspath_of_relpath_starting_at_root(self):
232
transport = MemoryTransport()
233
self.assertEqual("memory:///foo", transport.abspath('/foo'))
111
def test_relpath(self):
112
transport = MemoryTransport()
235
114
def test_append_and_get(self):
236
115
transport = MemoryTransport()
237
transport.append_bytes('path', 'content')
116
transport.append('path', StringIO('content'))
238
117
self.assertEqual(transport.get('path').read(), 'content')
239
transport.append_file('path', StringIO('content'))
118
transport.append('path', StringIO('content'))
240
119
self.assertEqual(transport.get('path').read(), 'contentcontent')
242
121
def test_put_and_get(self):
243
122
transport = MemoryTransport()
244
transport.put_file('path', StringIO('content'))
123
transport.put('path', StringIO('content'))
245
124
self.assertEqual(transport.get('path').read(), 'content')
246
transport.put_bytes('path', 'content')
125
transport.put('path', StringIO('content'))
247
126
self.assertEqual(transport.get('path').read(), 'content')
249
128
def test_append_without_dir_fails(self):
250
129
transport = MemoryTransport()
251
130
self.assertRaises(NoSuchFile,
252
transport.append_bytes, 'dir/path', 'content')
131
transport.append, 'dir/path', StringIO('content'))
254
133
def test_put_without_dir_fails(self):
255
134
transport = MemoryTransport()
256
135
self.assertRaises(NoSuchFile,
257
transport.put_file, 'dir/path', StringIO('content'))
136
transport.put, 'dir/path', StringIO('content'))
259
138
def test_get_missing(self):
260
139
transport = MemoryTransport()
304
173
def test_iter_files_recursive(self):
305
174
transport = MemoryTransport()
306
175
transport.mkdir('dir')
307
transport.put_bytes('dir/foo', 'content')
308
transport.put_bytes('dir/bar', 'content')
309
transport.put_bytes('bar', 'content')
176
transport.put('dir/foo', StringIO('content'))
177
transport.put('dir/bar', StringIO('content'))
178
transport.put('bar', StringIO('content'))
310
179
paths = set(transport.iter_files_recursive())
311
180
self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
313
182
def test_stat(self):
314
183
transport = MemoryTransport()
315
transport.put_bytes('foo', 'content')
316
transport.put_bytes('bar', 'phowar')
184
transport.put('foo', StringIO('content'))
185
transport.put('bar', StringIO('phowar'))
317
186
self.assertEqual(7, transport.stat('foo').st_size)
318
187
self.assertEqual(6, transport.stat('bar').st_size)
321
class ChrootDecoratorTransportTest(TestCase):
322
"""Chroot decoration specific tests."""
324
def test_abspath(self):
325
# The abspath is always relative to the chroot_url.
326
server = ChrootServer(get_transport('memory:///foo/bar/'))
328
transport = get_transport(server.get_url())
329
self.assertEqual(server.get_url(), transport.abspath('/'))
331
subdir_transport = transport.clone('subdir')
332
self.assertEqual(server.get_url(), subdir_transport.abspath('/'))
335
def test_clone(self):
336
server = ChrootServer(get_transport('memory:///foo/bar/'))
338
transport = get_transport(server.get_url())
339
# relpath from root and root path are the same
340
relpath_cloned = transport.clone('foo')
341
abspath_cloned = transport.clone('/foo')
342
self.assertEqual(server, relpath_cloned.server)
343
self.assertEqual(server, abspath_cloned.server)
346
def test_chroot_url_preserves_chroot(self):
347
"""Calling get_transport on a chroot transport's base should produce a
348
transport with exactly the same behaviour as the original chroot
351
This is so that it is not possible to escape a chroot by doing::
352
url = chroot_transport.base
353
parent_url = urlutils.join(url, '..')
354
new_transport = get_transport(parent_url)
356
server = ChrootServer(get_transport('memory:///path/subpath'))
358
transport = get_transport(server.get_url())
359
new_transport = get_transport(transport.base)
360
self.assertEqual(transport.server, new_transport.server)
361
self.assertEqual(transport.base, new_transport.base)
364
def test_urljoin_preserves_chroot(self):
365
"""Using urlutils.join(url, '..') on a chroot URL should not produce a
366
URL that escapes the intended chroot.
368
This is so that it is not possible to escape a chroot by doing::
369
url = chroot_transport.base
370
parent_url = urlutils.join(url, '..')
371
new_transport = get_transport(parent_url)
373
server = ChrootServer(get_transport('memory:///path/'))
375
transport = get_transport(server.get_url())
377
InvalidURLJoin, urlutils.join, transport.base, '..')
381
class ChrootServerTest(TestCase):
383
def test_construct(self):
384
backing_transport = MemoryTransport()
385
server = ChrootServer(backing_transport)
386
self.assertEqual(backing_transport, server.backing_transport)
388
def test_setUp(self):
389
backing_transport = MemoryTransport()
390
server = ChrootServer(backing_transport)
392
self.assertTrue(server.scheme in _get_protocol_handlers().keys())
394
def test_tearDown(self):
395
backing_transport = MemoryTransport()
396
server = ChrootServer(backing_transport)
399
self.assertFalse(server.scheme in _get_protocol_handlers().keys())
401
def test_get_url(self):
402
backing_transport = MemoryTransport()
403
server = ChrootServer(backing_transport)
405
self.assertEqual('chroot-%d:///' % id(server), server.get_url())
409
190
class ReadonlyDecoratorTransportTest(TestCase):
410
191
"""Readonly decoration specific tests."""
525
306
class BackupTransportHandler(Transport):
526
307
"""Test transport that works as a backup for the BadTransportHandler"""
530
class TestTransportImplementation(TestCaseInTempDir):
531
"""Implementation verification for transports.
533
To verify a transport we need a server factory, which is a callable
534
that accepts no parameters and returns an implementation of
535
bzrlib.transport.Server.
537
That Server is then used to construct transport instances and test
538
the transport via loopback activity.
540
Currently this assumes that the Transport object is connected to the
541
current working directory. So that whatever is done
542
through the transport, should show up in the working
543
directory, and vice-versa. This is a bug, because its possible to have
544
URL schemes which provide access to something that may not be
545
result in storage on the local disk, i.e. due to file system limits, or
546
due to it being a database or some other non-filesystem tool.
548
This also tests to make sure that the functions work with both
549
generators and lists (assuming iter(list) is effectively a generator)
553
super(TestTransportImplementation, self).setUp()
554
self._server = self.transport_server()
556
self.addCleanup(self._server.tearDown)
558
def get_transport(self, relpath=None):
559
"""Return a connected transport to the local directory.
561
:param relpath: a path relative to the base url.
563
base_url = self._server.get_url()
564
url = self._adjust_url(base_url, relpath)
565
# try getting the transport via the regular interface:
566
t = get_transport(url)
567
# vila--20070607 if the following are commented out the test suite
568
# still pass. Is this really still needed or was it a forgotten
570
if not isinstance(t, self.transport_class):
571
# we did not get the correct transport class type. Override the
572
# regular connection behaviour by direct construction.
573
t = self.transport_class(url)
577
class TestLocalTransports(TestCase):
579
def test_get_transport_from_abspath(self):
580
here = os.path.abspath('.')
581
t = get_transport(here)
582
self.assertIsInstance(t, LocalTransport)
583
self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
585
def test_get_transport_from_relpath(self):
586
here = os.path.abspath('.')
587
t = get_transport('.')
588
self.assertIsInstance(t, LocalTransport)
589
self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
591
def test_get_transport_from_local_url(self):
592
here = os.path.abspath('.')
593
here_url = urlutils.local_path_to_url(here) + '/'
594
t = get_transport(here_url)
595
self.assertIsInstance(t, LocalTransport)
596
self.assertEquals(t.base, here_url)
598
def test_local_abspath(self):
599
here = os.path.abspath('.')
600
t = get_transport(here)
601
self.assertEquals(t.local_abspath(''), here)
604
class TestWin32LocalTransport(TestCase):
606
def test_unc_clone_to_root(self):
607
# Win32 UNC path like \\HOST\path
608
# clone to root should stop at least at \\HOST part
610
t = EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
613
self.assertEquals(t.base, 'file://HOST/')
614
# make sure we reach the root
616
self.assertEquals(t.base, 'file://HOST/')
618
class TestConnectedTransport(TestCase):
619
"""Tests for connected to remote server transports"""
621
def test_parse_url(self):
622
t = ConnectedTransport('sftp://simple.example.com/home/source')
623
self.assertEquals(t._host, 'simple.example.com')
624
self.assertEquals(t._port, None)
625
self.assertEquals(t._path, '/home/source/')
626
self.failUnless(t._user is None)
627
self.failUnless(t._password is None)
629
self.assertEquals(t.base, 'sftp://simple.example.com/home/source/')
631
def test_parse_quoted_url(self):
632
t = ConnectedTransport('http://ro%62ey:h%40t@ex%41mple.com:2222/path')
633
self.assertEquals(t._host, 'exAmple.com')
634
self.assertEquals(t._port, 2222)
635
self.assertEquals(t._user, 'robey')
636
self.assertEquals(t._password, 'h@t')
637
self.assertEquals(t._path, '/path/')
639
# Base should not keep track of the password
640
self.assertEquals(t.base, 'http://robey@exAmple.com:2222/path/')
642
def test_parse_invalid_url(self):
643
self.assertRaises(errors.InvalidURL,
645
'sftp://lily.org:~janneke/public/bzr/gub')
647
def test_relpath(self):
648
t = ConnectedTransport('sftp://user@host.com/abs/path')
650
self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
651
self.assertRaises(errors.PathNotChild, t.relpath,
652
'http://user@host.com/abs/path/sub')
653
self.assertRaises(errors.PathNotChild, t.relpath,
654
'sftp://user2@host.com/abs/path/sub')
655
self.assertRaises(errors.PathNotChild, t.relpath,
656
'sftp://user@otherhost.com/abs/path/sub')
657
self.assertRaises(errors.PathNotChild, t.relpath,
658
'sftp://user@host.com:33/abs/path/sub')
659
# Make sure it works when we don't supply a username
660
t = ConnectedTransport('sftp://host.com/abs/path')
661
self.assertEquals(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
663
# Make sure it works when parts of the path will be url encoded
664
t = ConnectedTransport('sftp://host.com/dev/%path')
665
self.assertEquals(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
667
def test_connection_sharing_propagate_credentials(self):
668
t = ConnectedTransport('foo://user@host.com/abs/path')
669
self.assertIs(None, t._get_connection())
670
self.assertIs(None, t._password)
671
c = t.clone('subdir')
672
self.assertEquals(None, c._get_connection())
673
self.assertIs(None, t._password)
675
# Simulate the user entering a password
677
connection = object()
678
t._set_connection(connection, password)
679
self.assertIs(connection, t._get_connection())
680
self.assertIs(password, t._get_credentials())
681
self.assertIs(connection, c._get_connection())
682
self.assertIs(password, c._get_credentials())
684
# credentials can be updated
685
new_password = 'even more secret'
686
c._update_credentials(new_password)
687
self.assertIs(connection, t._get_connection())
688
self.assertIs(new_password, t._get_credentials())
689
self.assertIs(connection, c._get_connection())
690
self.assertIs(new_password, c._get_credentials())
693
class TestReusedTransports(TestCase):
694
"""Tests for transport reuse"""
696
def test_reuse_same_transport(self):
697
t1 = get_transport('http://foo/')
698
t2 = get_transport('http://foo/', possible_transports=[t1])
699
self.assertIs(t1, t2)
701
# Also check that final '/' are handled correctly
702
t3 = get_transport('http://foo/path/')
703
t4 = get_transport('http://foo/path', possible_transports=[t3])
704
self.assertIs(t3, t4)
706
t5 = get_transport('http://foo/path')
707
t6 = get_transport('http://foo/path/', possible_transports=[t5])
708
self.assertIs(t5, t6)
710
def test_don_t_reuse_different_transport(self):
711
t1 = get_transport('http://foo/path')
712
t2 = get_transport('http://bar/path', possible_transports=[t1])
713
self.assertIsNot(t1, t2)
716
def get_test_permutations():
717
"""Return transport permutations to be used in testing.
719
This module registers some transports, but they're only for testing
720
registration. We don't really want to run all the transport tests against