← 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: Canonical.com Patch Queue Manager
  • Date: 2011-08-17 18:13:57 UTC
  • mfrom: (5268.7.29 transport-segments)
  • Revision ID: pqm@pqm.ubuntu.com-20110817181357-y5q5eth1hk8bl3om
(jelmer) Allow specifying the colocated branch to use in the branch URL,
 and retrieving the branch name using ControlDir._get_selected_branch.
 (Jelmer Vernooij)

Show diffs side-by-side

added added

removed removed

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