← 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): Florent Gallaire
  • Date: 2017-03-17 12:23:47 UTC
  • mfrom: (6621.1.1 fix-gmtime-lite)
  • Revision ID: tarmac-20170317122347-2nnf3cicpgrhux3h
Fix for Windows and 32-bit platforms buggy gmtime(). [r=vila,richard-wilbur]

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)
 
809
 
 
810
    def test_missing_directory(self):
 
811
        t = self.get_transport('.')
 
812
        self.assertRaises(errors.NoSuchFile, t.open_write_stream, 'dir/foo')
762
813
 
763
814
 
764
815
class TestWin32LocalTransport(tests.TestCase):
765
816
 
766
817
    def test_unc_clone_to_root(self):
 
818
        self.requireFeature(features.win32_feature)
767
819
        # Win32 UNC path like \\HOST\path
768
820
        # clone to root should stop at least at \\HOST part
769
821
        # not on \\
770
822
        t = local.EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
771
823
        for i in xrange(4):
772
824
            t = t.clone('..')
773
 
        self.assertEquals(t.base, 'file://HOST/')
 
825
        self.assertEqual(t.base, 'file://HOST/')
774
826
        # make sure we reach the root
775
827
        t = t.clone('..')
776
 
        self.assertEquals(t.base, 'file://HOST/')
 
828
        self.assertEqual(t.base, 'file://HOST/')
777
829
 
778
830
 
779
831
class TestConnectedTransport(tests.TestCase):
782
834
    def test_parse_url(self):
783
835
        t = transport.ConnectedTransport(
784
836
            'http://simple.example.com/home/source')
785
 
        self.assertEquals(t._parsed_url.host, 'simple.example.com')
786
 
        self.assertEquals(t._parsed_url.port, None)
787
 
        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/')
788
840
        self.assertTrue(t._parsed_url.user is None)
789
841
        self.assertTrue(t._parsed_url.password is None)
790
842
 
791
 
        self.assertEquals(t.base, 'http://simple.example.com/home/source/')
 
843
        self.assertEqual(t.base, 'http://simple.example.com/home/source/')
792
844
 
793
845
    def test_parse_url_with_at_in_user(self):
794
846
        # Bug 228058
795
847
        t = transport.ConnectedTransport('ftp://user@host.com@www.host.com/')
796
 
        self.assertEquals(t._parsed_url.user, 'user@host.com')
 
848
        self.assertEqual(t._parsed_url.user, 'user@host.com')
797
849
 
798
850
    def test_parse_quoted_url(self):
799
851
        t = transport.ConnectedTransport(
800
852
            'http://ro%62ey:h%40t@ex%41mple.com:2222/path')
801
 
        self.assertEquals(t._parsed_url.host, 'exAmple.com')
802
 
        self.assertEquals(t._parsed_url.port, 2222)
803
 
        self.assertEquals(t._parsed_url.user, 'robey')
804
 
        self.assertEquals(t._parsed_url.password, 'h@t')
805
 
        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/')
806
858
 
807
859
        # Base should not keep track of the password
808
 
        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/')
809
861
 
810
862
    def test_parse_invalid_url(self):
811
863
        self.assertRaises(errors.InvalidURL,
815
867
    def test_relpath(self):
816
868
        t = transport.ConnectedTransport('sftp://user@host.com/abs/path')
817
869
 
818
 
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'),
 
870
        self.assertEqual(t.relpath('sftp://user@host.com/abs/path/sub'),
819
871
            'sub')
820
872
        self.assertRaises(errors.PathNotChild, t.relpath,
821
873
                          'http://user@host.com/abs/path/sub')
827
879
                          'sftp://user@host.com:33/abs/path/sub')
828
880
        # Make sure it works when we don't supply a username
829
881
        t = transport.ConnectedTransport('sftp://host.com/abs/path')
830
 
        self.assertEquals(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
 
882
        self.assertEqual(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
831
883
 
832
884
        # Make sure it works when parts of the path will be url encoded
833
885
        t = transport.ConnectedTransport('sftp://host.com/dev/%path')
834
 
        self.assertEquals(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
 
886
        self.assertEqual(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
835
887
 
836
888
    def test_connection_sharing_propagate_credentials(self):
837
889
        t = transport.ConnectedTransport('ftp://user@host.com/abs/path')
838
 
        self.assertEquals('user', t._parsed_url.user)
839
 
        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)
840
892
        self.assertIs(None, t._get_connection())
841
893
        self.assertIs(None, t._parsed_url.password)
842
894
        c = t.clone('subdir')
1044
1096
    def test_truncation(self):
1045
1097
        fake_html = "<p>something!\n" * 1000
1046
1098
        result = http.unhtml_roughly(fake_html)
1047
 
        self.assertEquals(len(result), 1000)
 
1099
        self.assertEqual(len(result), 1000)
1048
1100
        self.assertStartsWith(result, " something!")
1049
1101
 
1050
1102
 
1067
1119
        return path, url
1068
1120
 
1069
1121
    def test_regular_url(self):
1070
 
        self.assertEquals("file://foo", location_to_url("file://foo"))
 
1122
        self.assertEqual("file://foo", location_to_url("file://foo"))
1071
1123
 
1072
1124
    def test_directory(self):
1073
1125
        directories.register("bar:", SomeDirectory, "Dummy directory")
1074
1126
        self.addCleanup(directories.remove, "bar:")
1075
 
        self.assertEquals("http://bar", location_to_url("bar:"))
 
1127
        self.assertEqual("http://bar", location_to_url("bar:"))
1076
1128
 
1077
1129
    def test_unicode_url(self):
1078
1130
        self.assertRaises(errors.InvalidURL, location_to_url,
1082
1134
        path, url = self.get_base_location()
1083
1135
        location = path + "\xc3\xaf".decode("utf-8")
1084
1136
        url += '%C3%AF'
1085
 
        self.assertEquals(url, location_to_url(location))
 
1137
        self.assertEqual(url, location_to_url(location))
1086
1138
 
1087
1139
    def test_path(self):
1088
1140
        path, url = self.get_base_location()
1089
 
        self.assertEquals(url, location_to_url(path))
 
1141
        self.assertEqual(url, location_to_url(path))
1090
1142
 
1091
1143
    def test_relative_file_url(self):
1092
 
        self.assertEquals(urlutils.local_path_to_url(".") + "/bar",
 
1144
        self.assertEqual(urlutils.local_path_to_url(".") + "/bar",
1093
1145
            location_to_url("file:bar"))
1094
1146
 
1095
1147
    def test_absolute_file_url(self):
1096
 
        self.assertEquals("file:///bar", location_to_url("file:/bar"))
 
1148
        self.assertEqual("file:///bar", location_to_url("file:/bar"))