← 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: 2010-09-01 08:02:42 UTC
  • mfrom: (5390.3.3 faster-revert-593560)
  • Revision ID: pqm@pqm.ubuntu.com-20100901080242-esg62ody4frwmy66
(spiv) Avoid repeatedly calling self.target.all_file_ids() in
 InterTree.iter_changes. (Andrew Bennetts)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/per_transport.py
1
 
# Copyright (C) 2004, 2005, 2006, 2007 Canonical Ltd
 
1
# Copyright (C) 2005-2010 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
52
52
    TestNotApplicable,
53
53
    multiply_tests,
54
54
    )
 
55
from bzrlib.tests import test_server
55
56
from bzrlib.tests.test_transport import TestTransportImplementation
56
57
from bzrlib.transport import (
57
58
    ConnectedTransport,
79
80
            permutations = get_transport_test_permutations(
80
81
                reduce(getattr, (module).split('.')[1:], __import__(module)))
81
82
            for (klass, server_factory) in permutations:
82
 
                scenario = (server_factory.__name__,
 
83
                scenario = ('%s,%s' % (klass.__name__, server_factory.__name__),
83
84
                    {"transport_class":klass,
84
85
                     "transport_server":server_factory})
85
86
                result.append(scenario)
170
171
        self.assertEqual(True, t.has_any(['b', 'b', 'b']))
171
172
 
172
173
    def test_has_root_works(self):
173
 
        from bzrlib.smart import server
174
 
        if self.transport_server is server.SmartTCPServer_for_testing:
 
174
        if self.transport_server is test_server.SmartTCPServer_for_testing:
175
175
            raise TestNotApplicable(
176
176
                "SmartTCPServer_for_testing intentionally does not allow "
177
177
                "access to /.")
203
203
        for content, f in itertools.izip(contents, content_f):
204
204
            self.assertEqual(content, f.read())
205
205
 
 
206
    def test_get_unknown_file(self):
 
207
        t = self.get_transport()
 
208
        files = ['a', 'b']
 
209
        contents = ['contents of a\n',
 
210
                    'contents of b\n',
 
211
                    ]
 
212
        self.build_tree(files, transport=t, line_endings='binary')
206
213
        self.assertRaises(NoSuchFile, t.get, 'c')
207
214
        self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])
208
215
        self.assertListRaises(NoSuchFile, t.get_multi, iter(['a', 'b', 'c']))
242
249
        for content, fname in zip(contents, files):
243
250
            self.assertEqual(content, t.get_bytes(fname))
244
251
 
 
252
    def test_get_bytes_unknown_file(self):
 
253
        t = self.get_transport()
245
254
        self.assertRaises(NoSuchFile, t.get_bytes, 'c')
246
255
 
247
256
    def test_get_with_open_write_stream_sees_all_content(self):
674
683
            for f in files:
675
684
                self.assertTransportMode(temp_transport, f, mode)
676
685
 
 
686
    def test_create_prefix(self):
 
687
        t = self.get_transport()
 
688
        sub = t.clone('foo').clone('bar')
 
689
        try:
 
690
            sub.create_prefix()
 
691
        except TransportNotPossible:
 
692
            self.assertTrue(t.is_readonly())
 
693
        else:
 
694
            self.assertTrue(t.has('foo/bar'))
 
695
 
677
696
    def test_append_file(self):
678
697
        t = self.get_transport()
679
698
 
1063
1082
        subdir.stat('./file')
1064
1083
        subdir.stat('.')
1065
1084
 
 
1085
    def test_hardlink(self):
 
1086
        from stat import ST_NLINK
 
1087
 
 
1088
        t = self.get_transport()
 
1089
 
 
1090
        source_name = "original_target"
 
1091
        link_name = "target_link"
 
1092
 
 
1093
        self.build_tree([source_name], transport=t)
 
1094
 
 
1095
        try:
 
1096
            t.hardlink(source_name, link_name)
 
1097
 
 
1098
            self.failUnless(t.has(source_name))
 
1099
            self.failUnless(t.has(link_name))
 
1100
 
 
1101
            st = t.stat(link_name)
 
1102
            self.failUnlessEqual(st[ST_NLINK], 2)
 
1103
        except TransportNotPossible:
 
1104
            raise TestSkipped("Transport %s does not support hardlinks." %
 
1105
                              self._server.__class__)
 
1106
 
 
1107
    def test_symlink(self):
 
1108
        from stat import S_ISLNK
 
1109
 
 
1110
        t = self.get_transport()
 
1111
 
 
1112
        source_name = "original_target"
 
1113
        link_name = "target_link"
 
1114
 
 
1115
        self.build_tree([source_name], transport=t)
 
1116
 
 
1117
        try:
 
1118
            t.symlink(source_name, link_name)
 
1119
 
 
1120
            self.failUnless(t.has(source_name))
 
1121
            self.failUnless(t.has(link_name))
 
1122
 
 
1123
            st = t.stat(link_name)
 
1124
            self.failUnless(S_ISLNK(st.st_mode),
 
1125
                "expected symlink, got mode %o" % st.st_mode)
 
1126
        except TransportNotPossible:
 
1127
            raise TestSkipped("Transport %s does not support symlinks." %
 
1128
                              self._server.__class__)
 
1129
        except IOError:
 
1130
            raise tests.KnownFailure("Paramiko fails to create symlinks during tests")
 
1131
 
1066
1132
    def test_list_dir(self):
1067
1133
        # TODO: Test list_dir, just try once, and if it throws, stop testing
1068
1134
        t = self.get_transport()
1199
1265
        self.assertIs(t._get_connection(), c._get_connection())
1200
1266
 
1201
1267
        # Temporary failure, we need to create a new dummy connection
1202
 
        new_connection = object()
 
1268
        new_connection = None
1203
1269
        t._set_connection(new_connection)
1204
1270
        # Check that both transports use the same connection
1205
1271
        self.assertIs(new_connection, t._get_connection())
1477
1543
                 u'\u65e5', # Kanji person
1478
1544
                ]
1479
1545
 
 
1546
        no_unicode_support = getattr(self._server, 'no_unicode_support', False)
 
1547
        if no_unicode_support:
 
1548
            raise tests.KnownFailure("test server cannot handle unicode paths")
 
1549
 
1480
1550
        try:
1481
1551
            self.build_tree(files, transport=t, line_endings='binary')
1482
1552
        except UnicodeError:
1695
1765
        # also raise a special error
1696
1766
        self.assertListRaises((errors.ShortReadvError, errors.InvalidRange),
1697
1767
                              transport.readv, 'a', [(12,2)])
 
1768
 
 
1769
    def test_stat_symlink(self):
 
1770
        # if a transport points directly to a symlink (and supports symlinks
 
1771
        # at all) you can tell this.  helps with bug 32669.
 
1772
        t = self.get_transport()
 
1773
        try:
 
1774
            t.symlink('target', 'link')
 
1775
        except TransportNotPossible:
 
1776
            raise TestSkipped("symlinks not supported")
 
1777
        t2 = t.clone('link')
 
1778
        st = t2.stat('')
 
1779
        self.assertTrue(stat.S_ISLNK(st.st_mode))