← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

  • Committer: Tarmac
  • Author(s): Vincent Ladeuil
  • Date: 2017-01-30 14:42:05 UTC
  • mfrom: (6620.1.1 trunk)
  • Revision ID: tarmac-20170130144205-r8fh2xpmiuxyozpv
Merge  2.7 into trunk including fix for bug #1657238 [r=vila]

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
1
 
# Copyright (C) 2005-2011 Canonical Ltd
 
1
# Copyright (C) 2005-2011, 2015, 2016 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
16
16
 
17
17
 
18
18
from cStringIO import StringIO
 
19
import errno
19
20
import os
20
21
import subprocess
21
22
import sys
94
95
            transport.get_transport_from_url('foo://fooserver/foo')
95
96
        except errors.UnsupportedProtocol, e:
96
97
            e_str = str(e)
97
 
            self.assertEquals('Unsupported protocol'
 
98
            self.assertEqual('Unsupported protocol'
98
99
                                ' for url "foo://fooserver/foo":'
99
100
                                ' Unable to import library "some_lib":'
100
101
                                ' testing missing dependency', str(e))
120
121
            transport.get_transport_from_url('ssh://fooserver/foo')
121
122
        except errors.UnsupportedProtocol, e:
122
123
            e_str = str(e)
123
 
            self.assertEquals('Unsupported protocol'
 
124
            self.assertEqual('Unsupported protocol'
124
125
                              ' for url "ssh://fooserver/foo":'
125
126
                              ' bzr supports bzr+ssh to operate over ssh,'
126
127
                              ' use "bzr+ssh://fooserver/foo".',
297
298
 
298
299
    def test_has_missing(self):
299
300
        t = memory.MemoryTransport()
300
 
        self.assertEquals(False, t.has('foo'))
 
301
        self.assertEqual(False, t.has('foo'))
301
302
 
302
303
    def test_has_present(self):
303
304
        t = memory.MemoryTransport()
304
305
        t.append_bytes('foo', 'content')
305
 
        self.assertEquals(True, t.has('foo'))
 
306
        self.assertEqual(True, t.has('foo'))
306
307
 
307
308
    def test_list_dir(self):
308
309
        t = memory.MemoryTransport()
311
312
        t.put_bytes('dir/subfoo', 'content')
312
313
        t.put_bytes('dirlike', 'content')
313
314
 
314
 
        self.assertEquals(['dir', 'dirlike', 'foo'], sorted(t.list_dir('.')))
315
 
        self.assertEquals(['subfoo'], sorted(t.list_dir('dir')))
 
315
        self.assertEqual(['dir', 'dirlike', 'foo'], sorted(t.list_dir('.')))
 
316
        self.assertEqual(['subfoo'], sorted(t.list_dir('dir')))
316
317
 
317
318
    def test_mkdir(self):
318
319
        t = memory.MemoryTransport()
442
443
        self.assertEqual('chroot-%d:///' % id(server), server.get_url())
443
444
 
444
445
 
 
446
class TestHooks(tests.TestCase):
 
447
    """Basic tests for transport hooks"""
 
448
 
 
449
    def _get_connected_transport(self):
 
450
        return transport.ConnectedTransport("bogus:nowhere")
 
451
 
 
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,))
 
457
 
 
458
    def test_post_connect(self):
 
459
        """Ensure the post_connect hook is called when _set_transport is"""
 
460
        calls = []
 
461
        transport.Transport.hooks.install_named_hook("post_connect",
 
462
            calls.append, None)
 
463
        t = self._get_connected_transport()
 
464
        self.assertLength(0, calls)
 
465
        t._set_connection("connection", "auth")
 
466
        self.assertEqual(calls, [t])
 
467
 
 
468
 
445
469
class PathFilteringDecoratorTransportTest(tests.TestCase):
446
470
    """Pathfilter decoration specific tests."""
447
471
 
679
703
    def test_with_path(self):
680
704
        t = transport.get_transport_from_path(self.test_dir)
681
705
        self.assertIsInstance(t, local.LocalTransport)
682
 
        self.assertEquals(t.base.rstrip("/"),
 
706
        self.assertEqual(t.base.rstrip("/"),
683
707
            urlutils.local_path_to_url(self.test_dir))
684
708
 
685
709
    def test_with_url(self):
686
710
        t = transport.get_transport_from_path("file:")
687
711
        self.assertIsInstance(t, local.LocalTransport)
688
 
        self.assertEquals(t.base.rstrip("/"),
 
712
        self.assertEqual(t.base.rstrip("/"),
689
713
            urlutils.local_path_to_url(os.path.join(self.test_dir, "file:")))
690
714
 
691
715
 
699
723
        url = urlutils.local_path_to_url(self.test_dir)
700
724
        t = transport.get_transport_from_url(url)
701
725
        self.assertIsInstance(t, local.LocalTransport)
702
 
        self.assertEquals(t.base.rstrip("/"), url)
 
726
        self.assertEqual(t.base.rstrip("/"), url)
703
727
 
704
728
    def test_with_url_and_segment_parameters(self):
705
729
        url = urlutils.local_path_to_url(self.test_dir)+",branch=foo"
706
730
        t = transport.get_transport_from_url(url)
707
731
        self.assertIsInstance(t, local.LocalTransport)
708
 
        self.assertEquals(t.base.rstrip("/"), url)
 
732
        self.assertEqual(t.base.rstrip("/"), url)
709
733
        with open(os.path.join(self.test_dir, "afile"), 'w') as f:
710
734
            f.write("data")
711
735
        self.assertTrue(t.has("afile"))
717
741
        here = osutils.abspath('.')
718
742
        t = transport.get_transport(here)
719
743
        self.assertIsInstance(t, local.LocalTransport)
720
 
        self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
 
744
        self.assertEqual(t.base, urlutils.local_path_to_url(here) + '/')
721
745
 
722
746
    def test_get_transport_from_relpath(self):
723
747
        here = osutils.abspath('.')
724
748
        t = transport.get_transport('.')
725
749
        self.assertIsInstance(t, local.LocalTransport)
726
 
        self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
 
750
        self.assertEqual(t.base, urlutils.local_path_to_url('.') + '/')
727
751
 
728
752
    def test_get_transport_from_local_url(self):
729
753
        here = osutils.abspath('.')
730
754
        here_url = urlutils.local_path_to_url(here) + '/'
731
755
        t = transport.get_transport(here_url)
732
756
        self.assertIsInstance(t, local.LocalTransport)
733
 
        self.assertEquals(t.base, here_url)
 
757
        self.assertEqual(t.base, here_url)
734
758
 
735
759
    def test_local_abspath(self):
736
760
        here = osutils.abspath('.')
737
761
        t = transport.get_transport(here)
738
 
        self.assertEquals(t.local_abspath(''), here)
 
762
        self.assertEqual(t.local_abspath(''), here)
 
763
 
 
764
 
 
765
class TestLocalTransportMutation(tests.TestCaseInTempDir):
 
766
 
 
767
    def test_local_transport_mkdir(self):
 
768
        here = osutils.abspath('.')
 
769
        t = transport.get_transport(here)
 
770
        t.mkdir('test')
 
771
        self.assertTrue(os.path.exists('test'))
 
772
 
 
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
 
780
            raise e
 
781
        self.overrideAttr(os, 'chmod', fake_chmod)
 
782
        t.mkdir('test')
 
783
        t.mkdir('test2', mode=0707)
 
784
        self.assertTrue(os.path.exists('test'))
 
785
        self.assertTrue(os.path.exists('test2'))
739
786
 
740
787
 
741
788
class TestLocalTransportWriteStream(tests.TestCaseWithTransport):
757
804
        w.fdatasync()
758
805
        with open('out', 'rb') as f:
759
806
            # Should have been flushed.
760
 
            self.assertEquals(f.read(), 'foo')
761
 
        self.assertEquals(len(calls), 1, calls)
 
807
            self.assertEqual(f.read(), 'foo')
 
808
        self.assertEqual(len(calls), 1, calls)
762
809
 
763
810
    def test_missing_directory(self):
764
811
        t = self.get_transport('.')
768
815
class TestWin32LocalTransport(tests.TestCase):
769
816
 
770
817
    def test_unc_clone_to_root(self):
 
818
        self.requireFeature(features.win32_feature)
771
819
        # Win32 UNC path like \\HOST\path
772
820
        # clone to root should stop at least at \\HOST part
773
821
        # not on \\
774
822
        t = local.EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
775
823
        for i in xrange(4):
776
824
            t = t.clone('..')
777
 
        self.assertEquals(t.base, 'file://HOST/')
 
825
        self.assertEqual(t.base, 'file://HOST/')
778
826
        # make sure we reach the root
779
827
        t = t.clone('..')
780
 
        self.assertEquals(t.base, 'file://HOST/')
 
828
        self.assertEqual(t.base, 'file://HOST/')
781
829
 
782
830
 
783
831
class TestConnectedTransport(tests.TestCase):
786
834
    def test_parse_url(self):
787
835
        t = transport.ConnectedTransport(
788
836
            'http://simple.example.com/home/source')
789
 
        self.assertEquals(t._parsed_url.host, 'simple.example.com')
790
 
        self.assertEquals(t._parsed_url.port, None)
791
 
        self.assertEquals(t._parsed_url.path, '/home/source/')
 
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/')
792
840
        self.assertTrue(t._parsed_url.user is None)
793
841
        self.assertTrue(t._parsed_url.password is None)
794
842
 
795
 
        self.assertEquals(t.base, 'http://simple.example.com/home/source/')
 
843
        self.assertEqual(t.base, 'http://simple.example.com/home/source/')
796
844
 
797
845
    def test_parse_url_with_at_in_user(self):
798
846
        # Bug 228058
799
847
        t = transport.ConnectedTransport('ftp://user@host.com@www.host.com/')
800
 
        self.assertEquals(t._parsed_url.user, 'user@host.com')
 
848
        self.assertEqual(t._parsed_url.user, 'user@host.com')
801
849
 
802
850
    def test_parse_quoted_url(self):
803
851
        t = transport.ConnectedTransport(
804
852
            'http://ro%62ey:h%40t@ex%41mple.com:2222/path')
805
 
        self.assertEquals(t._parsed_url.host, 'exAmple.com')
806
 
        self.assertEquals(t._parsed_url.port, 2222)
807
 
        self.assertEquals(t._parsed_url.user, 'robey')
808
 
        self.assertEquals(t._parsed_url.password, 'h@t')
809
 
        self.assertEquals(t._parsed_url.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/')
810
858
 
811
859
        # Base should not keep track of the password
812
 
        self.assertEquals(t.base, 'http://ro%62ey@ex%41mple.com:2222/path/')
 
860
        self.assertEqual(t.base, 'http://ro%62ey@ex%41mple.com:2222/path/')
813
861
 
814
862
    def test_parse_invalid_url(self):
815
863
        self.assertRaises(errors.InvalidURL,
819
867
    def test_relpath(self):
820
868
        t = transport.ConnectedTransport('sftp://user@host.com/abs/path')
821
869
 
822
 
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'),
 
870
        self.assertEqual(t.relpath('sftp://user@host.com/abs/path/sub'),
823
871
            'sub')
824
872
        self.assertRaises(errors.PathNotChild, t.relpath,
825
873
                          'http://user@host.com/abs/path/sub')
831
879
                          'sftp://user@host.com:33/abs/path/sub')
832
880
        # Make sure it works when we don't supply a username
833
881
        t = transport.ConnectedTransport('sftp://host.com/abs/path')
834
 
        self.assertEquals(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
 
882
        self.assertEqual(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
835
883
 
836
884
        # Make sure it works when parts of the path will be url encoded
837
885
        t = transport.ConnectedTransport('sftp://host.com/dev/%path')
838
 
        self.assertEquals(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
 
886
        self.assertEqual(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
839
887
 
840
888
    def test_connection_sharing_propagate_credentials(self):
841
889
        t = transport.ConnectedTransport('ftp://user@host.com/abs/path')
842
 
        self.assertEquals('user', t._parsed_url.user)
843
 
        self.assertEquals('host.com', t._parsed_url.host)
 
890
        self.assertEqual('user', t._parsed_url.user)
 
891
        self.assertEqual('host.com', t._parsed_url.host)
844
892
        self.assertIs(None, t._get_connection())
845
893
        self.assertIs(None, t._parsed_url.password)
846
894
        c = t.clone('subdir')
1048
1096
    def test_truncation(self):
1049
1097
        fake_html = "<p>something!\n" * 1000
1050
1098
        result = http.unhtml_roughly(fake_html)
1051
 
        self.assertEquals(len(result), 1000)
 
1099
        self.assertEqual(len(result), 1000)
1052
1100
        self.assertStartsWith(result, " something!")
1053
1101
 
1054
1102
 
1071
1119
        return path, url
1072
1120
 
1073
1121
    def test_regular_url(self):
1074
 
        self.assertEquals("file://foo", location_to_url("file://foo"))
 
1122
        self.assertEqual("file://foo", location_to_url("file://foo"))
1075
1123
 
1076
1124
    def test_directory(self):
1077
1125
        directories.register("bar:", SomeDirectory, "Dummy directory")
1078
1126
        self.addCleanup(directories.remove, "bar:")
1079
 
        self.assertEquals("http://bar", location_to_url("bar:"))
 
1127
        self.assertEqual("http://bar", location_to_url("bar:"))
1080
1128
 
1081
1129
    def test_unicode_url(self):
1082
1130
        self.assertRaises(errors.InvalidURL, location_to_url,
1086
1134
        path, url = self.get_base_location()
1087
1135
        location = path + "\xc3\xaf".decode("utf-8")
1088
1136
        url += '%C3%AF'
1089
 
        self.assertEquals(url, location_to_url(location))
 
1137
        self.assertEqual(url, location_to_url(location))
1090
1138
 
1091
1139
    def test_path(self):
1092
1140
        path, url = self.get_base_location()
1093
 
        self.assertEquals(url, location_to_url(path))
 
1141
        self.assertEqual(url, location_to_url(path))
1094
1142
 
1095
1143
    def test_relative_file_url(self):
1096
 
        self.assertEquals(urlutils.local_path_to_url(".") + "/bar",
 
1144
        self.assertEqual(urlutils.local_path_to_url(".") + "/bar",
1097
1145
            location_to_url("file:bar"))
1098
1146
 
1099
1147
    def test_absolute_file_url(self):
1100
 
        self.assertEquals("file:///bar", location_to_url("file:/bar"))
 
1148
        self.assertEqual("file:///bar", location_to_url("file:/bar"))