← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/per_transport.py

(gz) Remove bzrlib/util/elementtree/ package (Martin Packman)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/per_transport.py
53
53
from bzrlib.tests.test_transport import TestTransportImplementation
54
54
from bzrlib.transport import (
55
55
    ConnectedTransport,
 
56
    Transport,
56
57
    _get_transport_modules,
57
58
    )
58
59
from bzrlib.transport.memory import MemoryTransport
 
60
from bzrlib.transport.remote import RemoteTransport
59
61
 
60
62
 
61
63
def get_transport_test_permutations(module):
1050
1052
        except NotImplementedError:
1051
1053
            raise TestSkipped("Transport %s has no bogus URL support." %
1052
1054
                              self._server.__class__)
1053
 
        t = _mod_transport.get_transport(url)
 
1055
        t = _mod_transport.get_transport_from_url(url)
1054
1056
        self.assertRaises((ConnectionError, NoSuchFile), t.get, '.bzr/branch')
1055
1057
 
1056
1058
    def test_stat(self):
1136
1138
            raise TestSkipped("Transport %s does not support symlinks." %
1137
1139
                              self._server.__class__)
1138
1140
        except IOError:
1139
 
            raise tests.KnownFailure("Paramiko fails to create symlinks during tests")
 
1141
            self.knownFailure("Paramiko fails to create symlinks during tests")
1140
1142
 
1141
1143
    def test_list_dir(self):
1142
1144
        # TODO: Test list_dir, just try once, and if it throws, stop testing
1206
1208
            raise TestSkipped("not a connected transport")
1207
1209
 
1208
1210
        t2 = t1.clone('subdir')
1209
 
        self.assertEquals(t1._scheme, t2._scheme)
1210
 
        self.assertEquals(t1._user, t2._user)
1211
 
        self.assertEquals(t1._password, t2._password)
1212
 
        self.assertEquals(t1._host, t2._host)
1213
 
        self.assertEquals(t1._port, t2._port)
 
1211
        self.assertEquals(t1._parsed_url.scheme, t2._parsed_url.scheme)
 
1212
        self.assertEquals(t1._parsed_url.user, t2._parsed_url.user)
 
1213
        self.assertEquals(t1._parsed_url.password, t2._parsed_url.password)
 
1214
        self.assertEquals(t1._parsed_url.host, t2._parsed_url.host)
 
1215
        self.assertEquals(t1._parsed_url.port, t2._parsed_url.port)
1214
1216
 
1215
1217
    def test__reuse_for(self):
1216
1218
        t = self.get_transport()
1223
1225
 
1224
1226
            Only the parameters different from None will be changed.
1225
1227
            """
1226
 
            if scheme   is None: scheme   = t._scheme
1227
 
            if user     is None: user     = t._user
1228
 
            if password is None: password = t._password
1229
 
            if user     is None: user     = t._user
1230
 
            if host     is None: host     = t._host
1231
 
            if port     is None: port     = t._port
1232
 
            if path     is None: path     = t._path
1233
 
            return t._unsplit_url(scheme, user, password, host, port, path)
 
1228
            if scheme   is None: scheme   = t._parsed_url.scheme
 
1229
            if user     is None: user     = t._parsed_url.user
 
1230
            if password is None: password = t._parsed_url.password
 
1231
            if user     is None: user     = t._parsed_url.user
 
1232
            if host     is None: host     = t._parsed_url.host
 
1233
            if port     is None: port     = t._parsed_url.port
 
1234
            if path     is None: path     = t._parsed_url.path
 
1235
            return str(urlutils.URL(scheme, user, password, host, port, path))
1234
1236
 
1235
 
        if t._scheme == 'ftp':
 
1237
        if t._parsed_url.scheme == 'ftp':
1236
1238
            scheme = 'sftp'
1237
1239
        else:
1238
1240
            scheme = 'ftp'
1239
1241
        self.assertIsNot(t, t._reuse_for(new_url(scheme=scheme)))
1240
 
        if t._user == 'me':
 
1242
        if t._parsed_url.user == 'me':
1241
1243
            user = 'you'
1242
1244
        else:
1243
1245
            user = 'me'
1254
1256
        #   (they may be typed by the user when prompted for example)
1255
1257
        self.assertIs(t, t._reuse_for(new_url(password='from space')))
1256
1258
        # We will not connect, we can use a invalid host
1257
 
        self.assertIsNot(t, t._reuse_for(new_url(host=t._host + 'bar')))
1258
 
        if t._port == 1234:
 
1259
        self.assertIsNot(t, t._reuse_for(new_url(host=t._parsed_url.host + 'bar')))
 
1260
        if t._parsed_url.port == 1234:
1259
1261
            port = 4321
1260
1262
        else:
1261
1263
            port = 1234
1418
1420
 
1419
1421
        # smoke test for abspath on win32.
1420
1422
        # a transport based on 'file:///' never fully qualifies the drive.
1421
 
        transport = _mod_transport.get_transport("file:///")
 
1423
        transport = _mod_transport.get_transport_from_url("file:///")
1422
1424
        self.assertEqual(transport.abspath("/"), "file:///")
1423
1425
 
1424
1426
        # but a transport that starts with a drive spec must keep it.
1425
 
        transport = _mod_transport.get_transport("file:///C:/")
 
1427
        transport = _mod_transport.get_transport_from_url("file:///C:/")
1426
1428
        self.assertEqual(transport.abspath("/"), "file:///C:/")
1427
1429
 
1428
1430
    def test_local_abspath(self):
1555
1557
 
1556
1558
        no_unicode_support = getattr(self._server, 'no_unicode_support', False)
1557
1559
        if no_unicode_support:
1558
 
            raise tests.KnownFailure("test server cannot handle unicode paths")
 
1560
            self.knownFailure("test server cannot handle unicode paths")
1559
1561
 
1560
1562
        try:
1561
1563
            self.build_tree(files, transport=t, line_endings='binary')
1776
1778
        self.assertListRaises((errors.ShortReadvError, errors.InvalidRange),
1777
1779
                              transport.readv, 'a', [(12,2)])
1778
1780
 
 
1781
    def test_no_segment_parameters(self):
 
1782
        """Segment parameters should be stripped and stored in
 
1783
        transport.segment_parameters."""
 
1784
        transport = self.get_transport("foo")
 
1785
        self.assertEquals({}, transport.get_segment_parameters())
 
1786
 
 
1787
    def test_segment_parameters(self):
 
1788
        """Segment parameters should be stripped and stored in
 
1789
        transport.get_segment_parameters()."""
 
1790
        base_url = self._server.get_url()
 
1791
        parameters = {"key1": "val1", "key2": "val2"}
 
1792
        url = urlutils.join_segment_parameters(base_url, parameters)
 
1793
        transport = _mod_transport.get_transport_from_url(url)
 
1794
        self.assertEquals(parameters, transport.get_segment_parameters())
 
1795
 
 
1796
    def test_set_segment_parameters(self):
 
1797
        """Segment parameters can be set and show up in base."""
 
1798
        transport = self.get_transport("foo")
 
1799
        orig_base = transport.base
 
1800
        transport.set_segment_parameter("arm", "board")
 
1801
        self.assertEquals("%s,arm=board" % orig_base, transport.base)
 
1802
        self.assertEquals({"arm": "board"}, transport.get_segment_parameters())
 
1803
        transport.set_segment_parameter("arm", None)
 
1804
        transport.set_segment_parameter("nonexistant", None)
 
1805
        self.assertEquals({}, transport.get_segment_parameters())
 
1806
        self.assertEquals(orig_base, transport.base)
 
1807
 
1779
1808
    def test_stat_symlink(self):
1780
1809
        # if a transport points directly to a symlink (and supports symlinks
1781
1810
        # at all) you can tell this.  helps with bug 32669.
1787
1816
        t2 = t.clone('link')
1788
1817
        st = t2.stat('')
1789
1818
        self.assertTrue(stat.S_ISLNK(st.st_mode))
 
1819
 
 
1820
    def test_abspath_url_unquote_unreserved(self):
 
1821
        """URLs from abspath should have unreserved characters unquoted
 
1822
        
 
1823
        Need consistent quoting notably for tildes, see lp:842223 for more.
 
1824
        """
 
1825
        t = self.get_transport()
 
1826
        needlessly_escaped_dir = "%2D%2E%30%39%41%5A%5F%61%7A%7E/"
 
1827
        self.assertEqual(t.base + "-.09AZ_az~",
 
1828
            t.abspath(needlessly_escaped_dir))
 
1829
 
 
1830
    def test_clone_url_unquote_unreserved(self):
 
1831
        """Base URL of a cloned branch needs unreserved characters unquoted
 
1832
        
 
1833
        Cloned transports should be prefix comparable for things like the
 
1834
        isolation checking of tests, see lp:842223 for more.
 
1835
        """
 
1836
        t1 = self.get_transport()
 
1837
        needlessly_escaped_dir = "%2D%2E%30%39%41%5A%5F%61%7A%7E/"
 
1838
        self.build_tree([needlessly_escaped_dir], transport=t1)
 
1839
        t2 = t1.clone(needlessly_escaped_dir)
 
1840
        self.assertEqual(t1.base + "-.09AZ_az~/", t2.base)
 
1841
 
 
1842
    def test_hook_post_connection_one(self):
 
1843
        """Fire post_connect hook after a ConnectedTransport is first used"""
 
1844
        log = []
 
1845
        Transport.hooks.install_named_hook("post_connect", log.append, None)
 
1846
        t = self.get_transport()
 
1847
        self.assertEqual([], log)
 
1848
        t.has("non-existant")
 
1849
        if isinstance(t, RemoteTransport):
 
1850
            self.assertEqual([t.get_smart_medium()], log)
 
1851
        elif isinstance(t, ConnectedTransport):
 
1852
            self.assertEqual([t], log)
 
1853
        else:
 
1854
            self.assertEqual([], log)
 
1855
 
 
1856
    def test_hook_post_connection_multi(self):
 
1857
        """Fire post_connect hook once per unshared underlying connection"""
 
1858
        log = []
 
1859
        Transport.hooks.install_named_hook("post_connect", log.append, None)
 
1860
        t1 = self.get_transport()
 
1861
        t2 = t1.clone(".")
 
1862
        t3 = self.get_transport()
 
1863
        self.assertEqual([], log)
 
1864
        t1.has("x")
 
1865
        t2.has("x")
 
1866
        t3.has("x")
 
1867
        if isinstance(t1, RemoteTransport):
 
1868
            self.assertEqual([t.get_smart_medium() for t in [t1, t3]], log)
 
1869
        elif isinstance(t1, ConnectedTransport):
 
1870
            self.assertEqual([t1, t3], log)
 
1871
        else:
 
1872
            self.assertEqual([], log)