/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport_implementations.py

  • Committer: Aaron Bentley
  • Date: 2007-02-06 14:52:16 UTC
  • mfrom: (2266 +trunk)
  • mto: This revision was merged to the branch mainline in revision 2268.
  • Revision ID: abentley@panoramicfeedback.com-20070206145216-fcpi8o3ufvuzwbp9
Merge bzr.dev

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport_implementations.py
1
 
# Copyright (C) 2004, 2005, 2006 by Canonical Ltd
 
1
# Copyright (C) 2004, 2005, 2006 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
38
38
from bzrlib.symbol_versioning import zero_eleven
39
39
from bzrlib.tests import TestCaseInTempDir, TestSkipped
40
40
from bzrlib.tests.test_transport import TestTransportImplementation
41
 
from bzrlib.transport import memory, smart
 
41
from bzrlib.transport import memory, smart, chroot
42
42
import bzrlib.transport
43
43
 
44
44
 
57
57
        """Check that transport.get(relpath).read() == content."""
58
58
        self.assertEqualDiff(content, transport.get(relpath).read())
59
59
 
60
 
    def assertListRaises(self, excClass, func, *args, **kwargs):
61
 
        """Fail unless excClass is raised when the iterator from func is used.
62
 
        
63
 
        Many transport functions can return generators this makes sure
64
 
        to wrap them in a list() call to make sure the whole generator
65
 
        is run, and that the proper exception is raised.
66
 
        """
67
 
        try:
68
 
            list(func(*args, **kwargs))
69
 
        except excClass:
70
 
            return
71
 
        else:
72
 
            if getattr(excClass,'__name__', None) is not None:
73
 
                excName = excClass.__name__
74
 
            else:
75
 
                excName = str(excClass)
76
 
            raise self.failureException, "%s not raised" % excName
77
 
 
78
60
    def test_has(self):
79
61
        t = self.get_transport()
80
62
 
94
76
 
95
77
    def test_has_root_works(self):
96
78
        current_transport = self.get_transport()
97
 
        # import pdb;pdb.set_trace()
 
79
        if isinstance(current_transport, chroot.ChrootTransportDecorator):
 
80
            raise TestSkipped("ChrootTransportDecorator disallows clone('..')")
98
81
        self.assertTrue(current_transport.has('/'))
99
82
        root = current_transport.clone('/')
100
83
        self.assertTrue(root.has(''))
897
880
        except NotImplementedError:
898
881
            raise TestSkipped("Transport %s has no bogus URL support." %
899
882
                              self._server.__class__)
 
883
        # This should be:  but SSH still connects on construction. No COOKIE!
 
884
        # self.assertRaises((ConnectionError, NoSuchFile), t.get, '.bzr/branch')
900
885
        try:
901
886
            t = bzrlib.transport.get_transport(url)
902
887
            t.get('.bzr/branch')
959
944
            l.sort()
960
945
            return l
961
946
 
962
 
        # SftpServer creates control files in the working directory
963
 
        # so lets move down a directory to avoid those.
964
 
        if not t.is_readonly():
965
 
            t.mkdir('wd')
966
 
        else:
967
 
            os.mkdir('wd')
968
 
        t = t.clone('wd')
969
 
 
970
947
        self.assertEqual([], sorted_list('.'))
971
948
        # c2 is precisely one letter longer than c here to test that
972
949
        # suffixing is not confused.
973
950
        # a%25b checks that quoting is done consistently across transports
974
951
        tree_names = ['a', 'a%25b', 'b', 'c/', 'c/d', 'c/e', 'c2/']
 
952
 
975
953
        if not t.is_readonly():
976
954
            self.build_tree(tree_names, transport=t)
977
955
        else:
978
 
            self.build_tree(['wd/' + name for name in tree_names])
 
956
            self.build_tree(tree_names)
979
957
 
980
958
        self.assertEqual(
981
959
            ['a', 'a%2525b', 'b', 'c', 'c2'], sorted_list('.'))
985
963
            t.delete('c/d')
986
964
            t.delete('b')
987
965
        else:
988
 
            os.unlink('wd/c/d')
989
 
            os.unlink('wd/b')
 
966
            os.unlink('c/d')
 
967
            os.unlink('b')
990
968
            
991
969
        self.assertEqual(['a', 'a%2525b', 'c', 'c2'], sorted_list('.'))
992
970
        self.assertEqual(['e'], sorted_list('c'))
1012
990
    def test_clone(self):
1013
991
        # TODO: Test that clone moves up and down the filesystem
1014
992
        t1 = self.get_transport()
 
993
        if isinstance(t1, chroot.ChrootTransportDecorator):
 
994
            raise TestSkipped("ChrootTransportDecorator disallows clone('..')")
1015
995
 
1016
996
        self.build_tree(['a', 'b/', 'b/c'], transport=t1)
1017
997
 
1044
1024
 
1045
1025
    def test_clone_to_root(self):
1046
1026
        orig_transport = self.get_transport()
 
1027
        if isinstance(orig_transport, chroot.ChrootTransportDecorator):
 
1028
            raise TestSkipped("ChrootTransportDecorator disallows clone('..')")
1047
1029
        # Repeatedly go up to a parent directory until we're at the root
1048
1030
        # directory of this transport
1049
1031
        root_transport = orig_transport
1050
1032
        new_transport = root_transport.clone("..")
1051
 
        # as we are walking up directories, the path must be must be 
 
1033
        # as we are walking up directories, the path must be
1052
1034
        # growing less, except at the top
1053
1035
        self.assertTrue(len(new_transport.base) < len(root_transport.base)
1054
1036
            or new_transport.base == root_transport.base)
1055
1037
        while new_transport.base != root_transport.base:
1056
1038
            root_transport = new_transport
1057
1039
            new_transport = root_transport.clone("..")
1058
 
            # as we are walking up directories, the path must be must be 
 
1040
            # as we are walking up directories, the path must be
1059
1041
            # growing less, except at the top
1060
1042
            self.assertTrue(len(new_transport.base) < len(root_transport.base)
1061
1043
                or new_transport.base == root_transport.base)
1072
1054
    def test_clone_from_root(self):
1073
1055
        """At the root, cloning to a simple dir should just do string append."""
1074
1056
        orig_transport = self.get_transport()
 
1057
        if isinstance(orig_transport, chroot.ChrootTransportDecorator):
 
1058
            raise TestSkipped("ChrootTransportDecorator disallows clone('/')")
1075
1059
        root_transport = orig_transport.clone('/')
1076
1060
        self.assertEqual(root_transport.base + '.bzr/',
1077
1061
            root_transport.clone('.bzr').base)
1092
1076
 
1093
1077
    def test_relpath_at_root(self):
1094
1078
        t = self.get_transport()
 
1079
        if isinstance(t, chroot.ChrootTransportDecorator):
 
1080
            raise TestSkipped("ChrootTransportDecorator disallows clone('..')")
1095
1081
        # clone all the way to the top
1096
1082
        new_transport = t.clone('..')
1097
1083
        while new_transport.base != t.base:
1107
1093
        # that have aliasing problems like symlinks should go in backend
1108
1094
        # specific test cases.
1109
1095
        transport = self.get_transport()
 
1096
        if isinstance(transport, chroot.ChrootTransportDecorator):
 
1097
            raise TestSkipped("ChrootTransportDecorator disallows clone('..')")
1110
1098
        
1111
1099
        self.assertEqual(transport.base + 'relpath',
1112
1100
                         transport.abspath('relpath'))
1117
1105
        # the abspath of "/" and "/foo/.." should result in the same location
1118
1106
        self.assertEqual(transport.abspath("/"), transport.abspath("/foo/.."))
1119
1107
 
 
1108
        self.assertEqual(transport.clone("/").abspath('foo'),
 
1109
                         transport.abspath("/foo"))
 
1110
 
1120
1111
    def test_local_abspath(self):
1121
1112
        transport = self.get_transport()
1122
1113
        try:
1128
1119
 
1129
1120
    def test_abspath_at_root(self):
1130
1121
        t = self.get_transport()
 
1122
        if isinstance(t, chroot.ChrootTransportDecorator):
 
1123
            raise TestSkipped("ChrootTransportDecorator disallows clone('..')")
1131
1124
        # clone all the way to the top
1132
1125
        new_transport = t.clone('..')
1133
1126
        while new_transport.base != t.base:
1232
1225
        # check that our server (whatever it is) is accessable reliably
1233
1226
        # via get_transport and multiple connections share content.
1234
1227
        transport = self.get_transport()
 
1228
        if isinstance(transport, chroot.ChrootTransportDecorator):
 
1229
            raise TestSkipped("ChrootTransportDecorator disallows clone('..')")
1235
1230
        if transport.is_readonly():
1236
1231
            return
1237
1232
        transport.put_bytes('foo', 'bar')
1290
1285
        else:
1291
1286
            transport.put_bytes('a', '0123456789')
1292
1287
 
 
1288
        d = list(transport.readv('a', ((0, 1),)))
 
1289
        self.assertEqual(d[0], (0, '0'))
 
1290
 
1293
1291
        d = list(transport.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
1294
1292
        self.assertEqual(d[0], (0, '0'))
1295
1293
        self.assertEqual(d[1], (1, '1'))
1309
1307
        self.assertEqual(d[2], (0, '0'))
1310
1308
        self.assertEqual(d[3], (3, '34'))
1311
1309
 
1312
 
    def test_get_smart_client(self):
1313
 
        """All transports must either give a smart client, or know they can't.
1314
 
 
1315
 
        For some transports such as http this might depend on probing to see 
1316
 
        what's actually present on the other end.  (But we can adjust for that 
1317
 
        in the future.)
 
1310
    def test_get_smart_medium(self):
 
1311
        """All transports must either give a smart medium, or know they can't.
1318
1312
        """
1319
1313
        transport = self.get_transport()
1320
1314
        try:
1321
 
            client = transport.get_smart_client()
1322
 
            # XXX: should be a more general class
1323
 
            self.assertIsInstance(client, smart.SmartStreamClient)
1324
 
        except NoSmartServer:
 
1315
            medium = transport.get_smart_medium()
 
1316
            self.assertIsInstance(medium, smart.SmartClientMedium)
 
1317
        except errors.NoSmartMedium:
1325
1318
            # as long as we got it we're fine
1326
1319
            pass
1327
1320
 
1334
1327
 
1335
1328
        # This is intentionally reading off the end of the file
1336
1329
        # since we are sure that it cannot get there
1337
 
        self.assertListRaises((errors.ShortReadvError, AssertionError),
 
1330
        self.assertListRaises((errors.ShortReadvError, errors.InvalidRange,
 
1331
                               # Can be raised by paramiko
 
1332
                               AssertionError),
1338
1333
                              transport.readv, 'a', [(1,1), (8,10)])
1339
1334
 
1340
1335
        # This is trying to seek past the end of the file, it should
1341
1336
        # also raise a special error
1342
 
        self.assertListRaises(errors.ShortReadvError,
 
1337
        self.assertListRaises((errors.ShortReadvError, errors.InvalidRange),
1343
1338
                              transport.readv, 'a', [(12,2)])