~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/per_transport.py

• browse files at revision 6015.59.1
• compare with another revision
• download diff
• download tarball
• view history from revision 6015.59.1

• reverse the comparison (6015.59.1 to 6449.5.3)
• stop comparing with revision 6449.5.3

Show diffs side-by-side

added

removed

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,

57

56

    _get_transport_modules,

58

57

    )

59

58

from bzrlib.transport.memory import MemoryTransport

60

 

from bzrlib.transport.remote import RemoteTransport

61

59

 

62

60

 

63

61

def get_transport_test_permutations(module):

1052

1050

        except NotImplementedError:

1053

1051

            raise TestSkipped("Transport %s has no bogus URL support." %

1054

1052

                              self._server.__class__)

1055

 

        t = _mod_transport.get_transport_from_url(url)

 

1053

        t = _mod_transport.get_transport(url)

1056

1054

        self.assertRaises((ConnectionError, NoSuchFile), t.get, '.bzr/branch')

1057

1055

 

1058

1056

    def test_stat(self):

1138

1136

            raise TestSkipped("Transport %s does not support symlinks." %

1139

1137

                              self._server.__class__)

1140

1138

        except IOError:

1141

 

            self.knownFailure("Paramiko fails to create symlinks during tests")

 

1139

            raise tests.KnownFailure("Paramiko fails to create symlinks during tests")

1142

1140

 

1143

1141

    def test_list_dir(self):

1144

1142

        # TODO: Test list_dir, just try once, and if it throws, stop testing

1208

1206

            raise TestSkipped("not a connected transport")

1209

1207

 

1210

1208

        t2 = t1.clone('subdir')

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)

 

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)

1216

1214

 

1217

1215

    def test__reuse_for(self):

1218

1216

        t = self.get_transport()

1225

1223

 

1226

1224

            Only the parameters different from None will be changed.

1227

1225

            """

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))

 

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)

1236

1234

 

1237

 

        if t._parsed_url.scheme == 'ftp':

 

1235

        if t._scheme == 'ftp':

1238

1236

            scheme = 'sftp'

1239

1237

        else:

1240

1238

            scheme = 'ftp'

1241

1239

        self.assertIsNot(t, t._reuse_for(new_url(scheme=scheme)))

1242

 

        if t._parsed_url.user == 'me':

 

1240

        if t._user == 'me':

1243

1241

            user = 'you'

1244

1242

        else:

1245

1243

            user = 'me'

1256

1254

        #   (they may be typed by the user when prompted for example)

1257

1255

        self.assertIs(t, t._reuse_for(new_url(password='from space')))

1258

1256

        # We will not connect, we can use a invalid host

1259

 

        self.assertIsNot(t, t._reuse_for(new_url(host=t._parsed_url.host + 'bar')))

1260

 

        if t._parsed_url.port == 1234:

 

1257

        self.assertIsNot(t, t._reuse_for(new_url(host=t._host + 'bar')))

 

1258

        if t._port == 1234:

1261

1259

            port = 4321

1262

1260

        else:

1263

1261

            port = 1234

1420

1418

 

1421

1419

        # smoke test for abspath on win32.

1422

1420

        # a transport based on 'file:///' never fully qualifies the drive.

1423

 

        transport = _mod_transport.get_transport_from_url("file:///")

 

1421

        transport = _mod_transport.get_transport("file:///")

1424

1422

        self.assertEqual(transport.abspath("/"), "file:///")

1425

1423

 

1426

1424

        # but a transport that starts with a drive spec must keep it.

1427

 

        transport = _mod_transport.get_transport_from_url("file:///C:/")

 

1425

        transport = _mod_transport.get_transport("file:///C:/")

1428

1426

        self.assertEqual(transport.abspath("/"), "file:///C:/")

1429

1427

 

1430

1428

    def test_local_abspath(self):

1557

1555

 

1558

1556

        no_unicode_support = getattr(self._server, 'no_unicode_support', False)

1559

1557

        if no_unicode_support:

1560

 

            self.knownFailure("test server cannot handle unicode paths")

 

1558

            raise tests.KnownFailure("test server cannot handle unicode paths")

1561

1559

 

1562

1560

        try:

1563

1561

            self.build_tree(files, transport=t, line_endings='binary')

1778

1776

        self.assertListRaises((errors.ShortReadvError, errors.InvalidRange),

1779

1777

                              transport.readv, 'a', [(12,2)])

1780

1778

 

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

 

 

1808

1779

    def test_stat_symlink(self):

1809

1780

        # if a transport points directly to a symlink (and supports symlinks

1810

1781

        # at all) you can tell this.  helps with bug 32669.

1816

1787

        t2 = t.clone('link')

1817

1788

        st = t2.stat('')

1818

1789

        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)

• Older »

Loggerhead is a web-based interface for Breezy
Version: 2.0.1