← 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: 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/per_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
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):
520
518
        self.assertTransportMode(t, 'dir777', 0777)
521
519
 
522
520
    def test_put_bytes_unicode(self):
 
521
        # Expect put_bytes to raise AssertionError or UnicodeEncodeError if
 
522
        # given unicode "bytes".  UnicodeEncodeError doesn't really make sense
 
523
        # (we don't want to encode unicode here at all, callers should be
 
524
        # strictly passing bytes to put_bytes), but we allow it for backwards
 
525
        # compatibility.  At some point we should use a specific exception.
 
526
        # See https://bugs.launchpad.net/bzr/+bug/106898.
523
527
        t = self.get_transport()
524
528
        if t.is_readonly():
525
529
            return
526
530
        unicode_string = u'\u1234'
527
 
        self.assertRaises(TypeError, t.put_bytes, 'foo', unicode_string)
 
531
        self.assertRaises(
 
532
            (AssertionError, UnicodeEncodeError),
 
533
            t.put_bytes, 'foo', unicode_string)
 
534
 
 
535
    def test_put_file_unicode(self):
 
536
        # Like put_bytes, except with a StringIO.StringIO of a unicode string.
 
537
        # This situation can happen (and has) if code is careless about the type
 
538
        # of "string" they initialise/write to a StringIO with.  We cannot use
 
539
        # cStringIO, because it never returns unicode from read.
 
540
        # Like put_bytes, UnicodeEncodeError isn't quite the right exception to
 
541
        # raise, but we raise it for hysterical raisins.
 
542
        t = self.get_transport()
 
543
        if t.is_readonly():
 
544
            return
 
545
        unicode_file = pyStringIO(u'\u1234')
 
546
        self.assertRaises(UnicodeEncodeError, t.put_file, 'foo', unicode_file)
528
547
 
529
548
    def test_mkdir(self):
530
549
        t = self.get_transport()
609
628
    def test_opening_a_file_stream_can_set_mode(self):
610
629
        t = self.get_transport()
611
630
        if t.is_readonly():
612
 
            self.assertRaises((TransportNotPossible, NotImplementedError),
613
 
                              t.open_write_stream, 'foo')
614
631
            return
615
632
        if not t._can_roundtrip_unix_modebits():
616
633
            # Can't roundtrip, so no need to run this test
617
634
            return
618
 
 
619
635
        def check_mode(name, mode, expected):
620
636
            handle = t.open_write_stream(name, mode=mode)
621
637
            handle.close()
903
919
    def test_rename_dir_succeeds(self):
904
920
        t = self.get_transport()
905
921
        if t.is_readonly():
906
 
            self.assertRaises((TransportNotPossible, NotImplementedError),
907
 
                              t.rename, 'foo', 'bar')
908
 
            return
 
922
            raise TestSkipped("transport is readonly")
909
923
        t.mkdir('adir')
910
924
        t.mkdir('adir/asubdir')
911
925
        t.rename('adir', 'bdir')
916
930
        """Attempting to replace a nonemtpy directory should fail"""
917
931
        t = self.get_transport()
918
932
        if t.is_readonly():
919
 
            self.assertRaises((TransportNotPossible, NotImplementedError),
920
 
                              t.rename, 'foo', 'bar')
921
 
            return
 
933
            raise TestSkipped("transport is readonly")
922
934
        t.mkdir('adir')
923
935
        t.mkdir('adir/asubdir')
924
936
        t.mkdir('bdir')
988
1000
        # perhaps all of this could be done in a subdirectory
989
1001
 
990
1002
        t.put_bytes('a', 'a first file\n')
991
 
        self.assertEqual([True, False], list(t.has_multi(['a', 'b'])))
 
1003
        self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
992
1004
 
993
1005
        t.move('a', 'b')
994
1006
        self.assertTrue(t.has('b'))
995
1007
        self.assertFalse(t.has('a'))
996
1008
 
997
1009
        self.check_transport_contents('a first file\n', t, 'b')
998
 
        self.assertEqual([False, True], list(t.has_multi(['a', 'b'])))
 
1010
        self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
999
1011
 
1000
1012
        # Overwrite a file
1001
1013
        t.put_bytes('c', 'c this file\n')
1038
1050
        except NotImplementedError:
1039
1051
            raise TestSkipped("Transport %s has no bogus URL support." %
1040
1052
                              self._server.__class__)
1041
 
        t = _mod_transport.get_transport_from_url(url)
 
1053
        t = _mod_transport.get_transport(url)
1042
1054
        self.assertRaises((ConnectionError, NoSuchFile), t.get, '.bzr/branch')
1043
1055
 
1044
1056
    def test_stat(self):
1194
1206
            raise TestSkipped("not a connected transport")
1195
1207
 
1196
1208
        t2 = t1.clone('subdir')
1197
 
        self.assertEqual(t1._parsed_url.scheme, t2._parsed_url.scheme)
1198
 
        self.assertEqual(t1._parsed_url.user, t2._parsed_url.user)
1199
 
        self.assertEqual(t1._parsed_url.password, t2._parsed_url.password)
1200
 
        self.assertEqual(t1._parsed_url.host, t2._parsed_url.host)
1201
 
        self.assertEqual(t1._parsed_url.port, t2._parsed_url.port)
 
1209
        self.assertEquals(t1._parsed_url.scheme, t2._parsed_url.scheme)
 
1210
        self.assertEquals(t1._parsed_url.user, t2._parsed_url.user)
 
1211
        self.assertEquals(t1._parsed_url.password, t2._parsed_url.password)
 
1212
        self.assertEquals(t1._parsed_url.host, t2._parsed_url.host)
 
1213
        self.assertEquals(t1._parsed_url.port, t2._parsed_url.port)
1202
1214
 
1203
1215
    def test__reuse_for(self):
1204
1216
        t = self.get_transport()
1406
1418
 
1407
1419
        # smoke test for abspath on win32.
1408
1420
        # a transport based on 'file:///' never fully qualifies the drive.
1409
 
        transport = _mod_transport.get_transport_from_url("file:///")
 
1421
        transport = _mod_transport.get_transport("file:///")
1410
1422
        self.assertEqual(transport.abspath("/"), "file:///")
1411
1423
 
1412
1424
        # but a transport that starts with a drive spec must keep it.
1413
 
        transport = _mod_transport.get_transport_from_url("file:///C:/")
 
1425
        transport = _mod_transport.get_transport("file:///C:/")
1414
1426
        self.assertEqual(transport.abspath("/"), "file:///C:/")
1415
1427
 
1416
1428
    def test_local_abspath(self):
1768
1780
        """Segment parameters should be stripped and stored in
1769
1781
        transport.segment_parameters."""
1770
1782
        transport = self.get_transport("foo")
1771
 
        self.assertEqual({}, transport.get_segment_parameters())
 
1783
        self.assertEquals({}, transport.get_segment_parameters())
1772
1784
 
1773
1785
    def test_segment_parameters(self):
1774
1786
        """Segment parameters should be stripped and stored in
1776
1788
        base_url = self._server.get_url()
1777
1789
        parameters = {"key1": "val1", "key2": "val2"}
1778
1790
        url = urlutils.join_segment_parameters(base_url, parameters)
1779
 
        transport = _mod_transport.get_transport_from_url(url)
1780
 
        self.assertEqual(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.assertEqual("%s,arm=board" % orig_base, transport.base)
1788
 
        self.assertEqual({"arm": "board"}, transport.get_segment_parameters())
1789
 
        transport.set_segment_parameter("arm", None)
1790
 
        transport.set_segment_parameter("nonexistant", None)
1791
 
        self.assertEqual({}, transport.get_segment_parameters())
1792
 
        self.assertEqual(orig_base, transport.base)
 
1791
        transport = _mod_transport.get_transport(url)
 
1792
        self.assertEquals(parameters, transport.get_segment_parameters())
1793
1793
 
1794
1794
    def test_stat_symlink(self):
1795
1795
        # if a transport points directly to a symlink (and supports symlinks
1802
1802
        t2 = t.clone('link')
1803
1803
        st = t2.stat('')
1804
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)