219
168
def test_clone(self):
220
169
transport = MemoryTransport()
221
170
self.assertTrue(isinstance(transport, MemoryTransport))
222
self.assertEqual("memory:///", transport.clone("/").base)
224
172
def test_abspath(self):
225
173
transport = MemoryTransport()
226
174
self.assertEqual("memory:///relpath", transport.abspath('relpath'))
228
def test_abspath_of_root(self):
229
transport = MemoryTransport()
230
self.assertEqual("memory:///", transport.base)
231
self.assertEqual("memory:///", transport.abspath('/'))
233
def test_abspath_of_relpath_starting_at_root(self):
234
transport = MemoryTransport()
235
self.assertEqual("memory:///foo", transport.abspath('/foo'))
176
def test_relpath(self):
177
transport = MemoryTransport()
237
179
def test_append_and_get(self):
238
180
transport = MemoryTransport()
239
transport.append_bytes('path', 'content')
181
transport.append('path', StringIO('content'))
240
182
self.assertEqual(transport.get('path').read(), 'content')
241
transport.append_file('path', StringIO('content'))
183
transport.append('path', StringIO('content'))
242
184
self.assertEqual(transport.get('path').read(), 'contentcontent')
244
186
def test_put_and_get(self):
245
187
transport = MemoryTransport()
246
transport.put_file('path', StringIO('content'))
188
transport.put('path', StringIO('content'))
247
189
self.assertEqual(transport.get('path').read(), 'content')
248
transport.put_bytes('path', 'content')
190
transport.put('path', StringIO('content'))
249
191
self.assertEqual(transport.get('path').read(), 'content')
251
193
def test_append_without_dir_fails(self):
252
194
transport = MemoryTransport()
253
195
self.assertRaises(NoSuchFile,
254
transport.append_bytes, 'dir/path', 'content')
196
transport.append, 'dir/path', StringIO('content'))
256
198
def test_put_without_dir_fails(self):
257
199
transport = MemoryTransport()
258
200
self.assertRaises(NoSuchFile,
259
transport.put_file, 'dir/path', StringIO('content'))
201
transport.put, 'dir/path', StringIO('content'))
261
203
def test_get_missing(self):
262
204
transport = MemoryTransport()
300
232
def test_parameters(self):
301
233
transport = MemoryTransport()
302
234
self.assertEqual(True, transport.listable())
235
self.assertEqual(False, transport.should_cache())
303
236
self.assertEqual(False, transport.is_readonly())
305
238
def test_iter_files_recursive(self):
306
239
transport = MemoryTransport()
307
240
transport.mkdir('dir')
308
transport.put_bytes('dir/foo', 'content')
309
transport.put_bytes('dir/bar', 'content')
310
transport.put_bytes('bar', 'content')
241
transport.put('dir/foo', StringIO('content'))
242
transport.put('dir/bar', StringIO('content'))
243
transport.put('bar', StringIO('content'))
311
244
paths = set(transport.iter_files_recursive())
312
245
self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
314
247
def test_stat(self):
315
248
transport = MemoryTransport()
316
transport.put_bytes('foo', 'content')
317
transport.put_bytes('bar', 'phowar')
249
transport.put('foo', StringIO('content'))
250
transport.put('bar', StringIO('phowar'))
318
251
self.assertEqual(7, transport.stat('foo').st_size)
319
252
self.assertEqual(6, transport.stat('bar').st_size)
322
class ChrootDecoratorTransportTest(TestCase):
323
"""Chroot decoration specific tests."""
325
def test_abspath(self):
326
# The abspath is always relative to the chroot_url.
327
server = ChrootServer(get_transport('memory:///foo/bar/'))
329
transport = get_transport(server.get_url())
330
self.assertEqual(server.get_url(), transport.abspath('/'))
332
subdir_transport = transport.clone('subdir')
333
self.assertEqual(server.get_url(), subdir_transport.abspath('/'))
336
def test_clone(self):
337
server = ChrootServer(get_transport('memory:///foo/bar/'))
339
transport = get_transport(server.get_url())
340
# relpath from root and root path are the same
341
relpath_cloned = transport.clone('foo')
342
abspath_cloned = transport.clone('/foo')
343
self.assertEqual(server, relpath_cloned.server)
344
self.assertEqual(server, abspath_cloned.server)
347
def test_chroot_url_preserves_chroot(self):
348
"""Calling get_transport on a chroot transport's base should produce a
349
transport with exactly the same behaviour as the original chroot
352
This is so that it is not possible to escape a chroot by doing::
353
url = chroot_transport.base
354
parent_url = urlutils.join(url, '..')
355
new_transport = get_transport(parent_url)
357
server = ChrootServer(get_transport('memory:///path/subpath'))
359
transport = get_transport(server.get_url())
360
new_transport = get_transport(transport.base)
361
self.assertEqual(transport.server, new_transport.server)
362
self.assertEqual(transport.base, new_transport.base)
365
def test_urljoin_preserves_chroot(self):
366
"""Using urlutils.join(url, '..') on a chroot URL should not produce a
367
URL that escapes the intended chroot.
369
This is so that it is not possible to escape a chroot by doing::
370
url = chroot_transport.base
371
parent_url = urlutils.join(url, '..')
372
new_transport = get_transport(parent_url)
374
server = ChrootServer(get_transport('memory:///path/'))
376
transport = get_transport(server.get_url())
378
InvalidURLJoin, urlutils.join, transport.base, '..')
382
class ChrootServerTest(TestCase):
384
def test_construct(self):
385
backing_transport = MemoryTransport()
386
server = ChrootServer(backing_transport)
387
self.assertEqual(backing_transport, server.backing_transport)
389
def test_setUp(self):
390
backing_transport = MemoryTransport()
391
server = ChrootServer(backing_transport)
393
self.assertTrue(server.scheme in _get_protocol_handlers().keys())
395
def test_tearDown(self):
396
backing_transport = MemoryTransport()
397
server = ChrootServer(backing_transport)
400
self.assertFalse(server.scheme in _get_protocol_handlers().keys())
402
def test_get_url(self):
403
backing_transport = MemoryTransport()
404
server = ChrootServer(backing_transport)
406
self.assertEqual('chroot-%d:///' % id(server), server.get_url())
410
255
class ReadonlyDecoratorTransportTest(TestCase):
411
256
"""Readonly decoration specific tests."""
550
399
super(TestTransportImplementation, self).setUp()
551
400
self._server = self.transport_server()
552
401
self._server.setUp()
553
self.addCleanup(self._server.tearDown)
555
def get_transport(self, relpath=None):
556
"""Return a connected transport to the local directory.
558
:param relpath: a path relative to the base url.
404
super(TestTransportImplementation, self).tearDown()
405
self._server.tearDown()
407
def get_transport(self):
408
"""Return a connected transport to the local directory."""
560
409
base_url = self._server.get_url()
561
url = self._adjust_url(base_url, relpath)
562
410
# try getting the transport via the regular interface:
563
t = get_transport(url)
564
# vila--20070607 if the following are commented out the test suite
565
# still pass. Is this really still needed or was it a forgotten
567
if not isinstance(t, self.transport_class):
411
t = get_transport(base_url)
412
if not isinstance(t, self.transport_class):
568
413
# we did not get the correct transport class type. Override the
569
414
# regular connection behaviour by direct construction.
570
t = self.transport_class(url)
415
t = self.transport_class(base_url)
574
class TestLocalTransports(TestCase):
576
def test_get_transport_from_abspath(self):
577
here = osutils.abspath('.')
578
t = get_transport(here)
579
self.assertIsInstance(t, LocalTransport)
580
self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
582
def test_get_transport_from_relpath(self):
583
here = osutils.abspath('.')
584
t = get_transport('.')
585
self.assertIsInstance(t, LocalTransport)
586
self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
588
def test_get_transport_from_local_url(self):
589
here = osutils.abspath('.')
590
here_url = urlutils.local_path_to_url(here) + '/'
591
t = get_transport(here_url)
592
self.assertIsInstance(t, LocalTransport)
593
self.assertEquals(t.base, here_url)
595
def test_local_abspath(self):
596
here = osutils.abspath('.')
597
t = get_transport(here)
598
self.assertEquals(t.local_abspath(''), here)
601
class TestWin32LocalTransport(TestCase):
603
def test_unc_clone_to_root(self):
604
# Win32 UNC path like \\HOST\path
605
# clone to root should stop at least at \\HOST part
607
t = EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
610
self.assertEquals(t.base, 'file://HOST/')
611
# make sure we reach the root
613
self.assertEquals(t.base, 'file://HOST/')
616
class TestConnectedTransport(TestCase):
617
"""Tests for connected to remote server transports"""
619
def test_parse_url(self):
620
t = ConnectedTransport('http://simple.example.com/home/source')
621
self.assertEquals(t._host, 'simple.example.com')
622
self.assertEquals(t._port, 80)
623
self.assertEquals(t._path, '/home/source/')
624
self.failUnless(t._user is None)
625
self.failUnless(t._password is None)
627
self.assertEquals(t.base, 'http://simple.example.com/home/source/')
629
def test_parse_quoted_url(self):
630
t = ConnectedTransport('http://ro%62ey:h%40t@ex%41mple.com:2222/path')
631
self.assertEquals(t._host, 'exAmple.com')
632
self.assertEquals(t._port, 2222)
633
self.assertEquals(t._user, 'robey')
634
self.assertEquals(t._password, 'h@t')
635
self.assertEquals(t._path, '/path/')
637
# Base should not keep track of the password
638
self.assertEquals(t.base, 'http://robey@exAmple.com:2222/path/')
640
def test_parse_invalid_url(self):
641
self.assertRaises(errors.InvalidURL,
643
'sftp://lily.org:~janneke/public/bzr/gub')
645
def test_relpath(self):
646
t = ConnectedTransport('sftp://user@host.com/abs/path')
648
self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
649
self.assertRaises(errors.PathNotChild, t.relpath,
650
'http://user@host.com/abs/path/sub')
651
self.assertRaises(errors.PathNotChild, t.relpath,
652
'sftp://user2@host.com/abs/path/sub')
653
self.assertRaises(errors.PathNotChild, t.relpath,
654
'sftp://user@otherhost.com/abs/path/sub')
655
self.assertRaises(errors.PathNotChild, t.relpath,
656
'sftp://user@host.com:33/abs/path/sub')
657
# Make sure it works when we don't supply a username
658
t = ConnectedTransport('sftp://host.com/abs/path')
659
self.assertEquals(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
661
# Make sure it works when parts of the path will be url encoded
662
t = ConnectedTransport('sftp://host.com/dev/%path')
663
self.assertEquals(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
665
def test_connection_sharing_propagate_credentials(self):
666
t = ConnectedTransport('foo://user@host.com/abs/path')
667
self.assertIs(None, t._get_connection())
668
self.assertIs(None, t._password)
669
c = t.clone('subdir')
670
self.assertEquals(None, c._get_connection())
671
self.assertIs(None, t._password)
673
# Simulate the user entering a password
675
connection = object()
676
t._set_connection(connection, password)
677
self.assertIs(connection, t._get_connection())
678
self.assertIs(password, t._get_credentials())
679
self.assertIs(connection, c._get_connection())
680
self.assertIs(password, c._get_credentials())
682
# credentials can be updated
683
new_password = 'even more secret'
684
c._update_credentials(new_password)
685
self.assertIs(connection, t._get_connection())
686
self.assertIs(new_password, t._get_credentials())
687
self.assertIs(connection, c._get_connection())
688
self.assertIs(new_password, c._get_credentials())
691
class TestReusedTransports(TestCase):
692
"""Tests for transport reuse"""
694
def test_reuse_same_transport(self):
695
possible_transports = []
696
t1 = get_transport('http://foo/',
697
possible_transports=possible_transports)
698
self.assertEqual([t1], possible_transports)
699
t2 = get_transport('http://foo/', possible_transports=[t1])
700
self.assertIs(t1, t2)
702
# Also check that final '/' are handled correctly
703
t3 = get_transport('http://foo/path/')
704
t4 = get_transport('http://foo/path', possible_transports=[t3])
705
self.assertIs(t3, t4)
707
t5 = get_transport('http://foo/path')
708
t6 = get_transport('http://foo/path/', possible_transports=[t5])
709
self.assertIs(t5, t6)
711
def test_don_t_reuse_different_transport(self):
712
t1 = get_transport('http://foo/path')
713
t2 = get_transport('http://bar/path', possible_transports=[t1])
714
self.assertIsNot(t1, t2)
717
class TestRemoteTCPTransport(TestCase):
718
"""Tests for bzr:// transport (RemoteTCPTransport)."""
720
def test_relpath_with_implicit_port(self):
721
"""Connected transports with the same URL are the same, even if the
724
So t.relpath(url) should always be '' if t.base is the same as url, or
725
if the only difference is that one explicitly specifies the default
726
port and the other doesn't specify a port.
728
t_implicit_port = RemoteTCPTransport('bzr://host.com/')
729
self.assertEquals('', t_implicit_port.relpath('bzr://host.com/'))
730
self.assertEquals('', t_implicit_port.relpath('bzr://host.com:4155/'))
731
t_explicit_port = RemoteTCPTransport('bzr://host.com:4155/')
732
self.assertEquals('', t_explicit_port.relpath('bzr://host.com/'))
733
self.assertEquals('', t_explicit_port.relpath('bzr://host.com:4155/'))
735
def test_construct_uses_default_port(self):
736
"""If no port is specified, then RemoteTCPTransport uses
739
t = get_transport('bzr://host.com/')
740
self.assertEquals(BZR_DEFAULT_PORT, t._port)
742
def test_url_omits_default_port(self):
743
"""If a RemoteTCPTransport uses the default port, then its base URL
746
This is like how ":80" is omitted from "http://example.com/".
748
t = get_transport('bzr://host.com:4155/')
749
self.assertEquals('bzr://host.com/', t.base)
751
def test_url_includes_non_default_port(self):
752
"""Non-default ports are included in the transport's URL.
754
Contrast this to `test_url_omits_default_port`.
756
t = get_transport('bzr://host.com:666/')
757
self.assertEquals('bzr://host.com:666/', t.base)
760
class SSHPortTestMixin(object):
761
"""Mixin class for testing SSH-based transports' use of ports in URLs.
763
Unlike other connected transports, SSH-based transports (sftp, bzr+ssh)
764
don't have a default port, because the user may have OpenSSH configured to
765
use a non-standard port.
768
def make_url(self, netloc):
769
"""Make a url for the given netloc, using the scheme defined on the
772
return '%s://%s/' % (self.scheme, netloc)
774
def test_relpath_with_implicit_port(self):
775
"""SSH-based transports with the same URL are the same.
777
Note than an unspecified port number is different to port 22 (because
778
OpenSSH may be configured to use a non-standard port).
780
So t.relpath(url) should always be '' if t.base is the same as url, but
781
raise PathNotChild if the ports in t and url are not both specified (or
784
url_implicit_port = self.make_url('host.com')
785
url_explicit_port = self.make_url('host.com:22')
787
t_implicit_port = get_transport(url_implicit_port)
788
self.assertEquals('', t_implicit_port.relpath(url_implicit_port))
790
PathNotChild, t_implicit_port.relpath, url_explicit_port)
792
t_explicit_port = get_transport(url_explicit_port)
794
PathNotChild, t_explicit_port.relpath, url_implicit_port)
795
self.assertEquals('', t_explicit_port.relpath(url_explicit_port))
797
def test_construct_with_no_port(self):
798
"""If no port is specified, then the SSH-based transport's _port will
801
t = get_transport(self.make_url('host.com'))
802
self.assertEquals(None, t._port)
804
def test_url_with_no_port(self):
805
"""If no port was specified, none is shown in the base URL."""
806
t = get_transport(self.make_url('host.com'))
807
self.assertEquals(self.make_url('host.com'), t.base)
809
def test_url_includes_port(self):
810
"""An SSH-based transport's base will show the port if one was
811
specified, even if that port is 22, because we do not assume 22 is the
814
# 22 is the "standard" port for SFTP.
815
t = get_transport(self.make_url('host.com:22'))
816
self.assertEquals(self.make_url('host.com:22'), t.base)
817
# 666 is not a standard port.
818
t = get_transport(self.make_url('host.com:666'))
819
self.assertEquals(self.make_url('host.com:666'), t.base)
822
class SFTPTransportPortTest(TestCase, SSHPortTestMixin):
823
"""Tests for sftp:// transport (SFTPTransport)."""
828
class BzrSSHTransportPortTest(TestCase, SSHPortTestMixin):
829
"""Tests for bzr+ssh:// transport (RemoteSSHTransport)."""
834
class TestTransportTrace(TestCase):
837
transport = get_transport('trace+memory://')
838
self.assertIsInstance(
839
transport, bzrlib.transport.trace.TransportTraceDecorator)
841
def test_clone_preserves_activity(self):
842
transport = get_transport('trace+memory://')
843
transport2 = transport.clone('.')
844
self.assertTrue(transport is not transport2)
845
self.assertTrue(transport._activity is transport2._activity)
847
# the following specific tests are for the operations that have made use of
848
# logging in tests; we could test every single operation but doing that
849
# still won't cause a test failure when the top level Transport API
850
# changes; so there is little return doing that.
852
transport = get_transport('trace+memory:///')
853
transport.put_bytes('foo', 'barish')
856
# put_bytes records the bytes, not the content to avoid memory
858
expected_result.append(('put_bytes', 'foo', 6, None))
859
# get records the file name only.
860
expected_result.append(('get', 'foo'))
861
self.assertEqual(expected_result, transport._activity)
863
def test_readv(self):
864
transport = get_transport('trace+memory:///')
865
transport.put_bytes('foo', 'barish')
866
list(transport.readv('foo', [(0, 1), (3, 2)], adjust_for_latency=True,
869
# put_bytes records the bytes, not the content to avoid memory
871
expected_result.append(('put_bytes', 'foo', 6, None))
872
# readv records the supplied offset request
873
expected_result.append(('readv', 'foo', [(0, 1), (3, 2)], True, 6))
874
self.assertEqual(expected_result, transport._activity)
added
removed
bzrlib/tests/test_transport.py
