107
112
'foo', 'bzrlib.tests.test_transport', 'BackupTransportHandler')
108
113
transport.register_lazy_transport(
109
114
'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
110
t = transport.get_transport('foo://fooserver/foo')
115
t = transport.get_transport_from_url('foo://fooserver/foo')
111
116
self.assertTrue(isinstance(t, BackupTransportHandler))
113
118
def test_ssh_hints(self):
114
119
"""Transport ssh:// should raise an error pointing out bzr+ssh://"""
116
transport.get_transport('ssh://fooserver/foo')
121
transport.get_transport_from_url('ssh://fooserver/foo')
117
122
except errors.UnsupportedProtocol, e:
119
self.assertEquals('Unsupported protocol'
124
self.assertEqual('Unsupported protocol'
120
125
' for url "ssh://fooserver/foo":'
121
126
' bzr supports bzr+ssh to operate over ssh,'
122
127
' use "bzr+ssh://fooserver/foo".',
449
443
self.assertEqual('chroot-%d:///' % id(server), server.get_url())
446
class TestHooks(tests.TestCase):
447
"""Basic tests for transport hooks"""
449
def _get_connected_transport(self):
450
return transport.ConnectedTransport("bogus:nowhere")
452
def test_transporthooks_initialisation(self):
453
"""Check all expected transport hook points are set up"""
454
hookpoint = transport.TransportHooks()
455
self.assertTrue("post_connect" in hookpoint,
456
"post_connect not in %s" % (hookpoint,))
458
def test_post_connect(self):
459
"""Ensure the post_connect hook is called when _set_transport is"""
461
transport.Transport.hooks.install_named_hook("post_connect",
463
t = self._get_connected_transport()
464
self.assertLength(0, calls)
465
t._set_connection("connection", "auth")
466
self.assertEqual(calls, [t])
452
469
class PathFilteringDecoratorTransportTest(tests.TestCase):
453
470
"""Pathfilter decoration specific tests."""
455
472
def test_abspath(self):
456
473
# The abspath is always relative to the base of the backing transport.
457
474
server = pathfilter.PathFilteringServer(
458
transport.get_transport('memory:///foo/bar/'),
475
transport.get_transport_from_url('memory:///foo/bar/'),
460
477
server.start_server()
461
t = transport.get_transport(server.get_url())
478
t = transport.get_transport_from_url(server.get_url())
462
479
self.assertEqual(server.get_url(), t.abspath('/'))
464
481
subdir_t = t.clone('subdir')
701
class TestTransportFromPath(tests.TestCaseInTempDir):
703
def test_with_path(self):
704
t = transport.get_transport_from_path(self.test_dir)
705
self.assertIsInstance(t, local.LocalTransport)
706
self.assertEqual(t.base.rstrip("/"),
707
urlutils.local_path_to_url(self.test_dir))
709
def test_with_url(self):
710
t = transport.get_transport_from_path("file:")
711
self.assertIsInstance(t, local.LocalTransport)
712
self.assertEqual(t.base.rstrip("/"),
713
urlutils.local_path_to_url(os.path.join(self.test_dir, "file:")))
716
class TestTransportFromUrl(tests.TestCaseInTempDir):
718
def test_with_path(self):
719
self.assertRaises(errors.InvalidURL, transport.get_transport_from_url,
722
def test_with_url(self):
723
url = urlutils.local_path_to_url(self.test_dir)
724
t = transport.get_transport_from_url(url)
725
self.assertIsInstance(t, local.LocalTransport)
726
self.assertEqual(t.base.rstrip("/"), url)
728
def test_with_url_and_segment_parameters(self):
729
url = urlutils.local_path_to_url(self.test_dir)+",branch=foo"
730
t = transport.get_transport_from_url(url)
731
self.assertIsInstance(t, local.LocalTransport)
732
self.assertEqual(t.base.rstrip("/"), url)
733
with open(os.path.join(self.test_dir, "afile"), 'w') as f:
735
self.assertTrue(t.has("afile"))
684
738
class TestLocalTransports(tests.TestCase):
686
740
def test_get_transport_from_abspath(self):
687
741
here = osutils.abspath('.')
688
742
t = transport.get_transport(here)
689
743
self.assertIsInstance(t, local.LocalTransport)
690
self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
744
self.assertEqual(t.base, urlutils.local_path_to_url(here) + '/')
692
746
def test_get_transport_from_relpath(self):
693
747
here = osutils.abspath('.')
694
748
t = transport.get_transport('.')
695
749
self.assertIsInstance(t, local.LocalTransport)
696
self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
750
self.assertEqual(t.base, urlutils.local_path_to_url('.') + '/')
698
752
def test_get_transport_from_local_url(self):
699
753
here = osutils.abspath('.')
700
754
here_url = urlutils.local_path_to_url(here) + '/'
701
755
t = transport.get_transport(here_url)
702
756
self.assertIsInstance(t, local.LocalTransport)
703
self.assertEquals(t.base, here_url)
757
self.assertEqual(t.base, here_url)
705
759
def test_local_abspath(self):
706
760
here = osutils.abspath('.')
707
761
t = transport.get_transport(here)
708
self.assertEquals(t.local_abspath(''), here)
762
self.assertEqual(t.local_abspath(''), here)
765
class TestLocalTransportMutation(tests.TestCaseInTempDir):
767
def test_local_transport_mkdir(self):
768
here = osutils.abspath('.')
769
t = transport.get_transport(here)
771
self.assertTrue(os.path.exists('test'))
773
def test_local_transport_mkdir_permission_denied(self):
774
# See https://bugs.launchpad.net/bzr/+bug/606537
775
here = osutils.abspath('.')
776
t = transport.get_transport(here)
777
def fake_chmod(path, mode):
778
e = OSError('permission denied')
779
e.errno = errno.EPERM
781
self.overrideAttr(os, 'chmod', fake_chmod)
783
t.mkdir('test2', mode=0707)
784
self.assertTrue(os.path.exists('test'))
785
self.assertTrue(os.path.exists('test2'))
788
class TestLocalTransportWriteStream(tests.TestCaseWithTransport):
790
def test_local_fdatasync_calls_fdatasync(self):
791
"""Check fdatasync on a stream tries to flush the data to the OS.
793
We can't easily observe the external effect but we can at least see
797
fdatasync = getattr(os, 'fdatasync', sentinel)
798
if fdatasync is sentinel:
799
raise tests.TestNotApplicable('fdatasync not supported')
800
t = self.get_transport('.')
801
calls = self.recordCalls(os, 'fdatasync')
802
w = t.open_write_stream('out')
805
with open('out', 'rb') as f:
806
# Should have been flushed.
807
self.assertEqual(f.read(), 'foo')
808
self.assertEqual(len(calls), 1, calls)
810
def test_missing_directory(self):
811
t = self.get_transport('.')
812
self.assertRaises(errors.NoSuchFile, t.open_write_stream, 'dir/foo')
711
815
class TestWin32LocalTransport(tests.TestCase):
713
817
def test_unc_clone_to_root(self):
818
self.requireFeature(features.win32_feature)
714
819
# Win32 UNC path like \\HOST\path
715
820
# clone to root should stop at least at \\HOST part
717
822
t = local.EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
718
823
for i in xrange(4):
719
824
t = t.clone('..')
720
self.assertEquals(t.base, 'file://HOST/')
825
self.assertEqual(t.base, 'file://HOST/')
721
826
# make sure we reach the root
722
827
t = t.clone('..')
723
self.assertEquals(t.base, 'file://HOST/')
828
self.assertEqual(t.base, 'file://HOST/')
726
831
class TestConnectedTransport(tests.TestCase):
729
834
def test_parse_url(self):
730
835
t = transport.ConnectedTransport(
731
836
'http://simple.example.com/home/source')
732
self.assertEquals(t._host, 'simple.example.com')
733
self.assertEquals(t._port, None)
734
self.assertEquals(t._path, '/home/source/')
735
self.assertTrue(t._user is None)
736
self.assertTrue(t._password is None)
837
self.assertEqual(t._parsed_url.host, 'simple.example.com')
838
self.assertEqual(t._parsed_url.port, None)
839
self.assertEqual(t._parsed_url.path, '/home/source/')
840
self.assertTrue(t._parsed_url.user is None)
841
self.assertTrue(t._parsed_url.password is None)
738
self.assertEquals(t.base, 'http://simple.example.com/home/source/')
843
self.assertEqual(t.base, 'http://simple.example.com/home/source/')
740
845
def test_parse_url_with_at_in_user(self):
742
847
t = transport.ConnectedTransport('ftp://user@host.com@www.host.com/')
743
self.assertEquals(t._user, 'user@host.com')
848
self.assertEqual(t._parsed_url.user, 'user@host.com')
745
850
def test_parse_quoted_url(self):
746
851
t = transport.ConnectedTransport(
747
852
'http://ro%62ey:h%40t@ex%41mple.com:2222/path')
748
self.assertEquals(t._host, 'exAmple.com')
749
self.assertEquals(t._port, 2222)
750
self.assertEquals(t._user, 'robey')
751
self.assertEquals(t._password, 'h@t')
752
self.assertEquals(t._path, '/path/')
853
self.assertEqual(t._parsed_url.host, 'exAmple.com')
854
self.assertEqual(t._parsed_url.port, 2222)
855
self.assertEqual(t._parsed_url.user, 'robey')
856
self.assertEqual(t._parsed_url.password, 'h@t')
857
self.assertEqual(t._parsed_url.path, '/path/')
754
859
# Base should not keep track of the password
755
self.assertEquals(t.base, 'http://robey@exAmple.com:2222/path/')
860
self.assertEqual(t.base, 'http://ro%62ey@ex%41mple.com:2222/path/')
757
862
def test_parse_invalid_url(self):
758
863
self.assertRaises(errors.InvalidURL,
774
879
'sftp://user@host.com:33/abs/path/sub')
775
880
# Make sure it works when we don't supply a username
776
881
t = transport.ConnectedTransport('sftp://host.com/abs/path')
777
self.assertEquals(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
882
self.assertEqual(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
779
884
# Make sure it works when parts of the path will be url encoded
780
885
t = transport.ConnectedTransport('sftp://host.com/dev/%path')
781
self.assertEquals(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
886
self.assertEqual(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
783
888
def test_connection_sharing_propagate_credentials(self):
784
889
t = transport.ConnectedTransport('ftp://user@host.com/abs/path')
785
self.assertEquals('user', t._user)
786
self.assertEquals('host.com', t._host)
890
self.assertEqual('user', t._parsed_url.user)
891
self.assertEqual('host.com', t._parsed_url.host)
787
892
self.assertIs(None, t._get_connection())
788
self.assertIs(None, t._password)
893
self.assertIs(None, t._parsed_url.password)
789
894
c = t.clone('subdir')
790
895
self.assertIs(None, c._get_connection())
791
self.assertIs(None, t._password)
896
self.assertIs(None, t._parsed_url.password)
793
898
# Simulate the user entering a password
794
899
password = 'secret'
814
919
def test_reuse_same_transport(self):
815
920
possible_transports = []
816
t1 = transport.get_transport('http://foo/',
921
t1 = transport.get_transport_from_url('http://foo/',
817
922
possible_transports=possible_transports)
818
923
self.assertEqual([t1], possible_transports)
819
t2 = transport.get_transport('http://foo/',
924
t2 = transport.get_transport_from_url('http://foo/',
820
925
possible_transports=[t1])
821
926
self.assertIs(t1, t2)
823
928
# Also check that final '/' are handled correctly
824
t3 = transport.get_transport('http://foo/path/')
825
t4 = transport.get_transport('http://foo/path',
929
t3 = transport.get_transport_from_url('http://foo/path/')
930
t4 = transport.get_transport_from_url('http://foo/path',
826
931
possible_transports=[t3])
827
932
self.assertIs(t3, t4)
829
t5 = transport.get_transport('http://foo/path')
830
t6 = transport.get_transport('http://foo/path/',
934
t5 = transport.get_transport_from_url('http://foo/path')
935
t6 = transport.get_transport_from_url('http://foo/path/',
831
936
possible_transports=[t5])
832
937
self.assertIs(t5, t6)
834
939
def test_don_t_reuse_different_transport(self):
835
t1 = transport.get_transport('http://foo/path')
836
t2 = transport.get_transport('http://bar/path',
940
t1 = transport.get_transport_from_url('http://foo/path')
941
t2 = transport.get_transport_from_url('http://bar/path',
837
942
possible_transports=[t1])
838
943
self.assertIsNot(t1, t2)
841
946
class TestTransportTrace(tests.TestCase):
844
t = transport.get_transport('trace+memory://')
948
def test_decorator(self):
949
t = transport.get_transport_from_url('trace+memory://')
845
950
self.assertIsInstance(
846
951
t, bzrlib.transport.trace.TransportTraceDecorator)
848
953
def test_clone_preserves_activity(self):
849
t = transport.get_transport('trace+memory://')
954
t = transport.get_transport_from_url('trace+memory://')
850
955
t2 = t.clone('.')
851
956
self.assertTrue(t is not t2)
852
957
self.assertTrue(t._activity is t2._activity)
991
1096
def test_truncation(self):
992
1097
fake_html = "<p>something!\n" * 1000
993
1098
result = http.unhtml_roughly(fake_html)
994
self.assertEquals(len(result), 1000)
1099
self.assertEqual(len(result), 1000)
995
1100
self.assertStartsWith(result, " something!")
1103
class SomeDirectory(object):
1105
def look_up(self, name, url):
1109
class TestLocationToUrl(tests.TestCase):
1111
def get_base_location(self):
1112
path = osutils.abspath('/foo/bar')
1113
if path.startswith('/'):
1114
url = 'file://%s' % (path,)
1116
# On Windows, abspaths start with the drive letter, so we have to
1117
# add in the extra '/'
1118
url = 'file:///%s' % (path,)
1121
def test_regular_url(self):
1122
self.assertEqual("file://foo", location_to_url("file://foo"))
1124
def test_directory(self):
1125
directories.register("bar:", SomeDirectory, "Dummy directory")
1126
self.addCleanup(directories.remove, "bar:")
1127
self.assertEqual("http://bar", location_to_url("bar:"))
1129
def test_unicode_url(self):
1130
self.assertRaises(errors.InvalidURL, location_to_url,
1131
"http://fo/\xc3\xaf".decode("utf-8"))
1133
def test_unicode_path(self):
1134
path, url = self.get_base_location()
1135
location = path + "\xc3\xaf".decode("utf-8")
1137
self.assertEqual(url, location_to_url(location))
1139
def test_path(self):
1140
path, url = self.get_base_location()
1141
self.assertEqual(url, location_to_url(path))
1143
def test_relative_file_url(self):
1144
self.assertEqual(urlutils.local_path_to_url(".") + "/bar",
1145
location_to_url("file:bar"))
1147
def test_absolute_file_url(self):
1148
self.assertEqual("file:///bar", location_to_url("file:/bar"))