← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/per_transport.py

  • Committer: Richard Wilbur
  • Date: 2016-02-04 19:07:28 UTC
  • mto: This revision was merged to the branch mainline in revision 6618.
  • Revision ID: richard.wilbur@gmail.com-20160204190728-p0zvfii6zase0fw7
Update COPYING.txt from the original http://www.gnu.org/licenses/gpl-2.0.txt  (Only differences were in whitespace.)  Thanks to Petr Stodulka for pointing out the discrepancy.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/per_transport.py
1
 
# Copyright (C) 2005-2011 Canonical Ltd
 
1
# Copyright (C) 2005-2011, 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
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):
207
209
                    ]
208
210
        self.build_tree(files, transport=t, line_endings='binary')
209
211
        self.assertRaises(NoSuchFile, t.get, 'c')
210
 
        self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])
211
 
        self.assertListRaises(NoSuchFile, t.get_multi, iter(['a', 'b', 'c']))
 
212
        def iterate_and_close(func, *args):
 
213
            for f in func(*args):
 
214
                # We call f.read() here because things like paramiko actually
 
215
                # spawn a thread to prefetch the content, which we want to
 
216
                # consume before we close the handle.
 
217
                content = f.read()
 
218
                f.close()
 
219
        self.assertRaises(NoSuchFile, iterate_and_close,
 
220
                          t.get_multi, ['a', 'b', 'c'])
 
221
        self.assertRaises(NoSuchFile, iterate_and_close,
 
222
                          t.get_multi, iter(['a', 'b', 'c']))
212
223
 
213
224
    def test_get_directory_read_gives_ReadError(self):
214
225
        """consistent errors for read() on a file returned by get()."""
509
520
        self.assertTransportMode(t, 'dir777', 0777)
510
521
 
511
522
    def test_put_bytes_unicode(self):
512
 
        # Expect put_bytes to raise AssertionError or UnicodeEncodeError if
513
 
        # given unicode "bytes".  UnicodeEncodeError doesn't really make sense
514
 
        # (we don't want to encode unicode here at all, callers should be
515
 
        # strictly passing bytes to put_bytes), but we allow it for backwards
516
 
        # compatibility.  At some point we should use a specific exception.
517
 
        # See https://bugs.launchpad.net/bzr/+bug/106898.
518
523
        t = self.get_transport()
519
524
        if t.is_readonly():
520
525
            return
521
526
        unicode_string = u'\u1234'
522
 
        self.assertRaises(
523
 
            (AssertionError, UnicodeEncodeError),
524
 
            t.put_bytes, 'foo', unicode_string)
525
 
 
526
 
    def test_put_file_unicode(self):
527
 
        # Like put_bytes, except with a StringIO.StringIO of a unicode string.
528
 
        # This situation can happen (and has) if code is careless about the type
529
 
        # of "string" they initialise/write to a StringIO with.  We cannot use
530
 
        # cStringIO, because it never returns unicode from read.
531
 
        # Like put_bytes, UnicodeEncodeError isn't quite the right exception to
532
 
        # raise, but we raise it for hysterical raisins.
533
 
        t = self.get_transport()
534
 
        if t.is_readonly():
535
 
            return
536
 
        unicode_file = pyStringIO(u'\u1234')
537
 
        self.assertRaises(UnicodeEncodeError, t.put_file, 'foo', unicode_file)
 
527
        self.assertRaises(TypeError, t.put_bytes, 'foo', unicode_string)
538
528
 
539
529
    def test_mkdir(self):
540
530
        t = self.get_transport()
619
609
    def test_opening_a_file_stream_can_set_mode(self):
620
610
        t = self.get_transport()
621
611
        if t.is_readonly():
 
612
            self.assertRaises((TransportNotPossible, NotImplementedError),
 
613
                              t.open_write_stream, 'foo')
622
614
            return
623
615
        if not t._can_roundtrip_unix_modebits():
624
616
            # Can't roundtrip, so no need to run this test
625
617
            return
 
618
 
626
619
        def check_mode(name, mode, expected):
627
620
            handle = t.open_write_stream(name, mode=mode)
628
621
            handle.close()
910
903
    def test_rename_dir_succeeds(self):
911
904
        t = self.get_transport()
912
905
        if t.is_readonly():
913
 
            raise TestSkipped("transport is readonly")
 
906
            self.assertRaises((TransportNotPossible, NotImplementedError),
 
907
                              t.rename, 'foo', 'bar')
 
908
            return
914
909
        t.mkdir('adir')
915
910
        t.mkdir('adir/asubdir')
916
911
        t.rename('adir', 'bdir')
921
916
        """Attempting to replace a nonemtpy directory should fail"""
922
917
        t = self.get_transport()
923
918
        if t.is_readonly():
924
 
            raise TestSkipped("transport is readonly")
 
919
            self.assertRaises((TransportNotPossible, NotImplementedError),
 
920
                              t.rename, 'foo', 'bar')
 
921
            return
925
922
        t.mkdir('adir')
926
923
        t.mkdir('adir/asubdir')
927
924
        t.mkdir('bdir')
1041
1038
        except NotImplementedError:
1042
1039
            raise TestSkipped("Transport %s has no bogus URL support." %
1043
1040
                              self._server.__class__)
1044
 
        t = _mod_transport.get_transport(url)
 
1041
        t = _mod_transport.get_transport_from_url(url)
1045
1042
        self.assertRaises((ConnectionError, NoSuchFile), t.get, '.bzr/branch')
1046
1043
 
1047
1044
    def test_stat(self):
1079
1076
        self.assertListRaises(NoSuchFile, t.stat_multi, iter(['a', 'c', 'd']))
1080
1077
        self.build_tree(['subdir/', 'subdir/file'], transport=t)
1081
1078
        subdir = t.clone('subdir')
1082
 
        subdir.stat('./file')
1083
 
        subdir.stat('.')
 
1079
        st = subdir.stat('./file')
 
1080
        st = subdir.stat('.')
1084
1081
 
1085
1082
    def test_hardlink(self):
1086
1083
        from stat import ST_NLINK
1127
1124
            raise TestSkipped("Transport %s does not support symlinks." %
1128
1125
                              self._server.__class__)
1129
1126
        except IOError:
1130
 
            raise tests.KnownFailure("Paramiko fails to create symlinks during tests")
 
1127
            self.knownFailure("Paramiko fails to create symlinks during tests")
1131
1128
 
1132
1129
    def test_list_dir(self):
1133
1130
        # TODO: Test list_dir, just try once, and if it throws, stop testing
1197
1194
            raise TestSkipped("not a connected transport")
1198
1195
 
1199
1196
        t2 = t1.clone('subdir')
1200
 
        self.assertEquals(t1._scheme, t2._scheme)
1201
 
        self.assertEquals(t1._user, t2._user)
1202
 
        self.assertEquals(t1._password, t2._password)
1203
 
        self.assertEquals(t1._host, t2._host)
1204
 
        self.assertEquals(t1._port, t2._port)
 
1197
        self.assertEquals(t1._parsed_url.scheme, t2._parsed_url.scheme)
 
1198
        self.assertEquals(t1._parsed_url.user, t2._parsed_url.user)
 
1199
        self.assertEquals(t1._parsed_url.password, t2._parsed_url.password)
 
1200
        self.assertEquals(t1._parsed_url.host, t2._parsed_url.host)
 
1201
        self.assertEquals(t1._parsed_url.port, t2._parsed_url.port)
1205
1202
 
1206
1203
    def test__reuse_for(self):
1207
1204
        t = self.get_transport()
1214
1211
 
1215
1212
            Only the parameters different from None will be changed.
1216
1213
            """
1217
 
            if scheme   is None: scheme   = t._scheme
1218
 
            if user     is None: user     = t._user
1219
 
            if password is None: password = t._password
1220
 
            if user     is None: user     = t._user
1221
 
            if host     is None: host     = t._host
1222
 
            if port     is None: port     = t._port
1223
 
            if path     is None: path     = t._path
1224
 
            return t._unsplit_url(scheme, user, password, host, port, path)
 
1214
            if scheme   is None: scheme   = t._parsed_url.scheme
 
1215
            if user     is None: user     = t._parsed_url.user
 
1216
            if password is None: password = t._parsed_url.password
 
1217
            if user     is None: user     = t._parsed_url.user
 
1218
            if host     is None: host     = t._parsed_url.host
 
1219
            if port     is None: port     = t._parsed_url.port
 
1220
            if path     is None: path     = t._parsed_url.path
 
1221
            return str(urlutils.URL(scheme, user, password, host, port, path))
1225
1222
 
1226
 
        if t._scheme == 'ftp':
 
1223
        if t._parsed_url.scheme == 'ftp':
1227
1224
            scheme = 'sftp'
1228
1225
        else:
1229
1226
            scheme = 'ftp'
1230
1227
        self.assertIsNot(t, t._reuse_for(new_url(scheme=scheme)))
1231
 
        if t._user == 'me':
 
1228
        if t._parsed_url.user == 'me':
1232
1229
            user = 'you'
1233
1230
        else:
1234
1231
            user = 'me'
1245
1242
        #   (they may be typed by the user when prompted for example)
1246
1243
        self.assertIs(t, t._reuse_for(new_url(password='from space')))
1247
1244
        # We will not connect, we can use a invalid host
1248
 
        self.assertIsNot(t, t._reuse_for(new_url(host=t._host + 'bar')))
1249
 
        if t._port == 1234:
 
1245
        self.assertIsNot(t, t._reuse_for(new_url(host=t._parsed_url.host + 'bar')))
 
1246
        if t._parsed_url.port == 1234:
1250
1247
            port = 4321
1251
1248
        else:
1252
1249
            port = 1234
1409
1406
 
1410
1407
        # smoke test for abspath on win32.
1411
1408
        # a transport based on 'file:///' never fully qualifies the drive.
1412
 
        transport = _mod_transport.get_transport("file:///")
 
1409
        transport = _mod_transport.get_transport_from_url("file:///")
1413
1410
        self.assertEqual(transport.abspath("/"), "file:///")
1414
1411
 
1415
1412
        # but a transport that starts with a drive spec must keep it.
1416
 
        transport = _mod_transport.get_transport("file:///C:/")
 
1413
        transport = _mod_transport.get_transport_from_url("file:///C:/")
1417
1414
        self.assertEqual(transport.abspath("/"), "file:///C:/")
1418
1415
 
1419
1416
    def test_local_abspath(self):
1546
1543
 
1547
1544
        no_unicode_support = getattr(self._server, 'no_unicode_support', False)
1548
1545
        if no_unicode_support:
1549
 
            raise tests.KnownFailure("test server cannot handle unicode paths")
 
1546
            self.knownFailure("test server cannot handle unicode paths")
1550
1547
 
1551
1548
        try:
1552
1549
            self.build_tree(files, transport=t, line_endings='binary')
1617
1614
    def test_readv(self):
1618
1615
        transport = self.get_transport()
1619
1616
        if transport.is_readonly():
1620
 
            file('a', 'w').write('0123456789')
 
1617
            with file('a', 'w') as f: f.write('0123456789')
1621
1618
        else:
1622
1619
            transport.put_bytes('a', '0123456789')
1623
1620
 
1633
1630
    def test_readv_out_of_order(self):
1634
1631
        transport = self.get_transport()
1635
1632
        if transport.is_readonly():
1636
 
            file('a', 'w').write('0123456789')
 
1633
            with file('a', 'w') as f: f.write('0123456789')
1637
1634
        else:
1638
1635
            transport.put_bytes('a', '01234567890')
1639
1636
 
1711
1708
        transport = self.get_transport()
1712
1709
        # test from observed failure case.
1713
1710
        if transport.is_readonly():
1714
 
            file('a', 'w').write('a'*1024*1024)
 
1711
            with file('a', 'w') as f: f.write('a'*1024*1024)
1715
1712
        else:
1716
1713
            transport.put_bytes('a', 'a'*1024*1024)
1717
1714
        broken_vector = [(465219, 800), (225221, 800), (445548, 800),
1751
1748
    def test_readv_short_read(self):
1752
1749
        transport = self.get_transport()
1753
1750
        if transport.is_readonly():
1754
 
            file('a', 'w').write('0123456789')
 
1751
            with file('a', 'w') as f: f.write('0123456789')
1755
1752
        else:
1756
1753
            transport.put_bytes('a', '01234567890')
1757
1754
 
1767
1764
        self.assertListRaises((errors.ShortReadvError, errors.InvalidRange),
1768
1765
                              transport.readv, 'a', [(12,2)])
1769
1766
 
 
1767
    def test_no_segment_parameters(self):
 
1768
        """Segment parameters should be stripped and stored in
 
1769
        transport.segment_parameters."""
 
1770
        transport = self.get_transport("foo")
 
1771
        self.assertEquals({}, transport.get_segment_parameters())
 
1772
 
 
1773
    def test_segment_parameters(self):
 
1774
        """Segment parameters should be stripped and stored in
 
1775
        transport.get_segment_parameters()."""
 
1776
        base_url = self._server.get_url()
 
1777
        parameters = {"key1": "val1", "key2": "val2"}
 
1778
        url = urlutils.join_segment_parameters(base_url, parameters)
 
1779
        transport = _mod_transport.get_transport_from_url(url)
 
1780
        self.assertEquals(parameters, transport.get_segment_parameters())
 
1781
 
 
1782
    def test_set_segment_parameters(self):
 
1783
        """Segment parameters can be set and show up in base."""
 
1784
        transport = self.get_transport("foo")
 
1785
        orig_base = transport.base
 
1786
        transport.set_segment_parameter("arm", "board")
 
1787
        self.assertEquals("%s,arm=board" % orig_base, transport.base)
 
1788
        self.assertEquals({"arm": "board"}, transport.get_segment_parameters())
 
1789
        transport.set_segment_parameter("arm", None)
 
1790
        transport.set_segment_parameter("nonexistant", None)
 
1791
        self.assertEquals({}, transport.get_segment_parameters())
 
1792
        self.assertEquals(orig_base, transport.base)
 
1793
 
1770
1794
    def test_stat_symlink(self):
1771
1795
        # if a transport points directly to a symlink (and supports symlinks
1772
1796
        # at all) you can tell this.  helps with bug 32669.
1778
1802
        t2 = t.clone('link')
1779
1803
        st = t2.stat('')
1780
1804
        self.assertTrue(stat.S_ISLNK(st.st_mode))
 
1805
 
 
1806
    def test_abspath_url_unquote_unreserved(self):
 
1807
        """URLs from abspath should have unreserved characters unquoted
 
1808
        
 
1809
        Need consistent quoting notably for tildes, see lp:842223 for more.
 
1810
        """
 
1811
        t = self.get_transport()
 
1812
        needlessly_escaped_dir = "%2D%2E%30%39%41%5A%5F%61%7A%7E/"
 
1813
        self.assertEqual(t.base + "-.09AZ_az~",
 
1814
            t.abspath(needlessly_escaped_dir))
 
1815
 
 
1816
    def test_clone_url_unquote_unreserved(self):
 
1817
        """Base URL of a cloned branch needs unreserved characters unquoted
 
1818
        
 
1819
        Cloned transports should be prefix comparable for things like the
 
1820
        isolation checking of tests, see lp:842223 for more.
 
1821
        """
 
1822
        t1 = self.get_transport()
 
1823
        needlessly_escaped_dir = "%2D%2E%30%39%41%5A%5F%61%7A%7E/"
 
1824
        self.build_tree([needlessly_escaped_dir], transport=t1)
 
1825
        t2 = t1.clone(needlessly_escaped_dir)
 
1826
        self.assertEqual(t1.base + "-.09AZ_az~/", t2.base)
 
1827
 
 
1828
    def test_hook_post_connection_one(self):
 
1829
        """Fire post_connect hook after a ConnectedTransport is first used"""
 
1830
        log = []
 
1831
        Transport.hooks.install_named_hook("post_connect", log.append, None)
 
1832
        t = self.get_transport()
 
1833
        self.assertEqual([], log)
 
1834
        t.has("non-existant")
 
1835
        if isinstance(t, RemoteTransport):
 
1836
            self.assertEqual([t.get_smart_medium()], log)
 
1837
        elif isinstance(t, ConnectedTransport):
 
1838
            self.assertEqual([t], log)
 
1839
        else:
 
1840
            self.assertEqual([], log)
 
1841
 
 
1842
    def test_hook_post_connection_multi(self):
 
1843
        """Fire post_connect hook once per unshared underlying connection"""
 
1844
        log = []
 
1845
        Transport.hooks.install_named_hook("post_connect", log.append, None)
 
1846
        t1 = self.get_transport()
 
1847
        t2 = t1.clone(".")
 
1848
        t3 = self.get_transport()
 
1849
        self.assertEqual([], log)
 
1850
        t1.has("x")
 
1851
        t2.has("x")
 
1852
        t3.has("x")
 
1853
        if isinstance(t1, RemoteTransport):
 
1854
            self.assertEqual([t.get_smart_medium() for t in [t1, t3]], log)
 
1855
        elif isinstance(t1, ConnectedTransport):
 
1856
            self.assertEqual([t1, t3], log)
 
1857
        else:
 
1858
            self.assertEqual([], log)