/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar
2018.18.22 by Martin Pool
merge bzr.dev
1
# Copyright (C) 2004, 2005, 2006, 2007 Canonical Ltd
1887.1.1 by Adeodato Simó
Do not separate paragraphs in the copyright statement with blank lines,
2
#
1530.1.3 by Robert Collins
transport implementations now tested consistently.
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
1887.1.1 by Adeodato Simó
Do not separate paragraphs in the copyright statement with blank lines,
7
#
1530.1.3 by Robert Collins
transport implementations now tested consistently.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
# GNU General Public License for more details.
1887.1.1 by Adeodato Simó
Do not separate paragraphs in the copyright statement with blank lines,
12
#
1530.1.3 by Robert Collins
transport implementations now tested consistently.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
17
"""Tests for Transport implementations.
18
19
Transport implementations tested here are supplied by
20
TransportTestProviderAdapter.
21
"""
22
23
import os
24
from cStringIO import StringIO
2018.5.131 by Andrew Bennetts
Be strict about unicode passed to transport.put_{bytes,file} and SmartClient.call_with_body_bytes, fixing part of TestLockableFiles_RemoteLockDir.test_read_write.
25
from StringIO import StringIO as pyStringIO
1530.1.15 by Robert Collins
Move put mode tests into test_transport_implementation.
26
import stat
27
import sys
2553.2.5 by Robert Collins
And overhaul TransportTestProviderAdapter too.
28
import unittest
1530.1.3 by Robert Collins
transport implementations now tested consistently.
29
1755.3.7 by John Arbash Meinel
Clean up and write tests for permissions. Now we use fstat which should be cheap, and lets us check the permissions and the file size
30
from bzrlib import (
2001.3.2 by John Arbash Meinel
Force all transports to raise ShortReadvError if they can
31
    errors,
1755.3.7 by John Arbash Meinel
Clean up and write tests for permissions. Now we use fstat which should be cheap, and lets us check the permissions and the file size
32
    osutils,
33
    urlutils,
34
    )
2400.2.1 by Robert Collins
Split out the improvement to Transport.local_abspath to raise NotLocalURL from the hpss-faster-copy branch. (Martin Pool, Ian Clatworthy)
35
from bzrlib.errors import (ConnectionError,
36
                           DirectoryNotEmpty,
37
                           FileExists,
38
                           InvalidURL,
39
                           LockError,
40
                           NoSmartServer,
41
                           NoSuchFile,
42
                           NotLocalUrl,
43
                           PathError,
44
                           TransportNotPossible,
45
                           )
1685.1.9 by John Arbash Meinel
Updated LocalTransport so that it's base is now a URL rather than a local path. This helps consistency with all other functions. To do so, I added local_abspath() which returns the local path, and local_path_to/from_url
46
from bzrlib.osutils import getcwd
2018.5.139 by Andrew Bennetts
Merge from bzr.dev, resolving conflicts.
47
from bzrlib.smart import medium
1955.3.29 by John Arbash Meinel
Use applyDeprecated instead of callDeprecated
48
from bzrlib.symbol_versioning import zero_eleven
2553.2.5 by Robert Collins
And overhaul TransportTestProviderAdapter too.
49
from bzrlib.tests import TestCaseInTempDir, TestScenarioApplier, TestSkipped
1871.1.2 by Robert Collins
Reduce code duplication in transport-parameterised tests.
50
from bzrlib.tests.test_transport import TestTransportImplementation
2485.8.25 by Vincent Ladeuil
Separate abspath from _remote_path, the intents are different.
51
from bzrlib.transport import (
52
    ConnectedTransport,
53
    get_transport,
2485.8.50 by Vincent Ladeuil
merge bzr.dev @ 2584 resolving conflicts
54
    _get_transport_modules,
2485.8.25 by Vincent Ladeuil
Separate abspath from _remote_path, the intents are different.
55
    )
56
from bzrlib.transport.memory import MemoryTransport
1530.1.3 by Robert Collins
transport implementations now tested consistently.
57
58
2553.2.5 by Robert Collins
And overhaul TransportTestProviderAdapter too.
59
class TransportTestProviderAdapter(TestScenarioApplier):
60
    """A tool to generate a suite testing all transports for a single test.
61
62
    This is done by copying the test once for each transport and injecting
63
    the transport_class and transport_server classes into each copy. Each copy
64
    is also given a new id() to make it easy to identify.
65
    """
66
67
    def __init__(self):
68
        self.scenarios = self._test_permutations()
69
70
    def get_transport_test_permutations(self, module):
71
        """Get the permutations module wants to have tested."""
72
        if getattr(module, 'get_test_permutations', None) is None:
73
            raise AssertionError("transport module %s doesn't provide get_test_permutations()"
74
                    % module.__name__)
75
            ##warning("transport module %s doesn't provide get_test_permutations()"
76
            ##       % module.__name__)
77
            return []
78
        return module.get_test_permutations()
79
80
    def _test_permutations(self):
81
        """Return a list of the klass, server_factory pairs to test."""
82
        result = []
83
        for module in _get_transport_modules():
84
            try:
85
                permutations = self.get_transport_test_permutations(
86
                    reduce(getattr, (module).split('.')[1:], __import__(module)))
87
                for (klass, server_factory) in permutations:
88
                    scenario = (server_factory.__name__,
89
                        {"transport_class":klass,
90
                         "transport_server":server_factory})
91
                    result.append(scenario)
92
            except errors.DependencyNotPresent, e:
93
                # Continue even if a dependency prevents us 
94
                # from running this test
95
                pass
96
        return result
97
98
99
1871.1.2 by Robert Collins
Reduce code duplication in transport-parameterised tests.
100
class TransportTests(TestTransportImplementation):
101
2018.5.24 by Andrew Bennetts
Setting NO_SMART_VFS in environment will disable VFS methods in the smart server. (Robert Collins, John Arbash Meinel, Andrew Bennetts)
102
    def setUp(self):
103
        super(TransportTests, self).setUp()
2402.1.2 by Andrew Bennetts
Deal with review comments.
104
        self._captureVar('BZR_NO_SMART_VFS', None)
2018.5.24 by Andrew Bennetts
Setting NO_SMART_VFS in environment will disable VFS methods in the smart server. (Robert Collins, John Arbash Meinel, Andrew Bennetts)
105
1530.1.3 by Robert Collins
transport implementations now tested consistently.
106
    def check_transport_contents(self, content, transport, relpath):
107
        """Check that transport.get(relpath).read() == content."""
1530.1.21 by Robert Collins
Review feedback fixes.
108
        self.assertEqualDiff(content, transport.get(relpath).read())
1530.1.15 by Robert Collins
Move put mode tests into test_transport_implementation.
109
2475.3.2 by John Arbash Meinel
Add Transport.ensure_base()
110
    def test_ensure_base_missing(self):
111
        """.ensure_base() should create the directory if it doesn't exist"""
112
        t = self.get_transport()
113
        t_a = t.clone('a')
114
        if t_a.is_readonly():
115
            self.assertRaises(TransportNotPossible,
116
                              t_a.ensure_base)
117
            return
118
        self.assertTrue(t_a.ensure_base())
119
        self.assertTrue(t.has('a'))
120
121
    def test_ensure_base_exists(self):
122
        """.ensure_base() should just be happy if it already exists"""
123
        t = self.get_transport()
124
        if t.is_readonly():
125
            return
126
127
        t.mkdir('a')
128
        t_a = t.clone('a')
129
        # ensure_base returns False if it didn't create the base
130
        self.assertFalse(t_a.ensure_base())
131
132
    def test_ensure_base_missing_parent(self):
133
        """.ensure_base() will fail if the parent dir doesn't exist"""
134
        t = self.get_transport()
135
        if t.is_readonly():
136
            return
137
138
        t_a = t.clone('a')
139
        t_b = t_a.clone('b')
140
        self.assertRaises(NoSuchFile, t_b.ensure_base)
141
2634.1.1 by Robert Collins
(robertc) Reinstate the accidentally backed out external_url patch.
142
    def test_external_url(self):
143
        """.external_url either works or raises InProcessTransport."""
144
        t = self.get_transport()
145
        try:
146
            t.external_url()
147
        except errors.InProcessTransport:
148
            pass
149
1530.1.3 by Robert Collins
transport implementations now tested consistently.
150
    def test_has(self):
151
        t = self.get_transport()
152
153
        files = ['a', 'b', 'e', 'g', '%']
154
        self.build_tree(files, transport=t)
155
        self.assertEqual(True, t.has('a'))
156
        self.assertEqual(False, t.has('c'))
1685.1.45 by John Arbash Meinel
Moved url functions into bzrlib.urlutils
157
        self.assertEqual(True, t.has(urlutils.escape('%')))
1530.1.3 by Robert Collins
transport implementations now tested consistently.
158
        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
159
                [True, True, False, False, True, False, True, False])
160
        self.assertEqual(True, t.has_any(['a', 'b', 'c']))
1685.1.45 by John Arbash Meinel
Moved url functions into bzrlib.urlutils
161
        self.assertEqual(False, t.has_any(['c', 'd', 'f', urlutils.escape('%%')]))
1530.1.3 by Robert Collins
transport implementations now tested consistently.
162
        self.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']))),
163
                [True, True, False, False, True, False, True, False])
164
        self.assertEqual(False, t.has_any(['c', 'c', 'c']))
165
        self.assertEqual(True, t.has_any(['b', 'b', 'b']))
166
1986.1.10 by Robert Collins
Merge from bzr.dev, fixing found bugs handling 'has('/')' in MemoryTransport and SFTP transports.
167
    def test_has_root_works(self):
168
        current_transport = self.get_transport()
169
        self.assertTrue(current_transport.has('/'))
170
        root = current_transport.clone('/')
171
        self.assertTrue(root.has(''))
172
1530.1.3 by Robert Collins
transport implementations now tested consistently.
173
    def test_get(self):
174
        t = self.get_transport()
175
176
        files = ['a', 'b', 'e', 'g']
177
        contents = ['contents of a\n',
178
                    'contents of b\n',
179
                    'contents of e\n',
180
                    'contents of g\n',
181
                    ]
1551.2.39 by abentley
Fix line endings in tests
182
        self.build_tree(files, transport=t, line_endings='binary')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
183
        self.check_transport_contents('contents of a\n', t, 'a')
184
        content_f = t.get_multi(files)
185
        for content, f in zip(contents, content_f):
186
            self.assertEqual(content, f.read())
187
188
        content_f = t.get_multi(iter(files))
189
        for content, f in zip(contents, content_f):
190
            self.assertEqual(content, f.read())
191
192
        self.assertRaises(NoSuchFile, t.get, 'c')
193
        self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])
194
        self.assertListRaises(NoSuchFile, t.get_multi, iter(['a', 'b', 'c']))
195
2052.6.2 by Robert Collins
Merge bzr.dev.
196
    def test_get_directory_read_gives_ReadError(self):
197
        """consistent errors for read() on a file returned by get()."""
2052.6.1 by Robert Collins
``Transport.get`` has had its interface made more clear for ease of use.
198
        t = self.get_transport()
199
        if t.is_readonly():
200
            self.build_tree(['a directory/'])
201
        else:
202
            t.mkdir('a%20directory')
203
        # getting the file must either work or fail with a PathError
204
        try:
205
            a_file = t.get('a%20directory')
2052.6.2 by Robert Collins
Merge bzr.dev.
206
        except (errors.PathError, errors.RedirectRequested):
2052.6.1 by Robert Collins
``Transport.get`` has had its interface made more clear for ease of use.
207
            # early failure return immediately.
208
            return
209
        # having got a file, read() must either work (i.e. http reading a dir listing) or
210
        # fail with ReadError
211
        try:
212
            a_file.read()
213
        except errors.ReadError:
214
            pass
215
1955.3.3 by John Arbash Meinel
Implement and test 'get_bytes'
216
    def test_get_bytes(self):
217
        t = self.get_transport()
218
219
        files = ['a', 'b', 'e', 'g']
220
        contents = ['contents of a\n',
221
                    'contents of b\n',
222
                    'contents of e\n',
223
                    'contents of g\n',
224
                    ]
225
        self.build_tree(files, transport=t, line_endings='binary')
226
        self.check_transport_contents('contents of a\n', t, 'a')
227
228
        for content, fname in zip(contents, files):
229
            self.assertEqual(content, t.get_bytes(fname))
230
231
        self.assertRaises(NoSuchFile, t.get_bytes, 'c')
232
1530.1.3 by Robert Collins
transport implementations now tested consistently.
233
    def test_put(self):
234
        t = self.get_transport()
235
236
        if t.is_readonly():
1955.3.6 by John Arbash Meinel
Lots of deprecation warnings, but no errors
237
            return
238
1955.3.29 by John Arbash Meinel
Use applyDeprecated instead of callDeprecated
239
        self.applyDeprecated(zero_eleven, t.put, 'a', 'string\ncontents\n')
1955.3.6 by John Arbash Meinel
Lots of deprecation warnings, but no errors
240
        self.check_transport_contents('string\ncontents\n', t, 'a')
241
1955.3.29 by John Arbash Meinel
Use applyDeprecated instead of callDeprecated
242
        self.applyDeprecated(zero_eleven,
243
                             t.put, 'b', StringIO('file-like\ncontents\n'))
1955.3.6 by John Arbash Meinel
Lots of deprecation warnings, but no errors
244
        self.check_transport_contents('file-like\ncontents\n', t, 'b')
245
1910.15.6 by Andrew Bennetts
Merge from bzr.dev
246
        self.assertRaises(NoSuchFile,
1910.19.14 by Robert Collins
Fix up all tests to pass, remove a couple more deprecated function calls, and break the dependency on sftp for the smart transport.
247
            self.applyDeprecated,
248
            zero_eleven,
249
            t.put, 'path/doesnt/exist/c', StringIO('contents'))
1910.15.6 by Andrew Bennetts
Merge from bzr.dev
250
1955.3.1 by John Arbash Meinel
Add put_bytes() and a base-level implementation for it
251
    def test_put_bytes(self):
252
        t = self.get_transport()
253
254
        if t.is_readonly():
255
            self.assertRaises(TransportNotPossible,
256
                    t.put_bytes, 'a', 'some text for a\n')
257
            return
258
259
        t.put_bytes('a', 'some text for a\n')
260
        self.failUnless(t.has('a'))
261
        self.check_transport_contents('some text for a\n', t, 'a')
262
263
        # The contents should be overwritten
264
        t.put_bytes('a', 'new text for a\n')
265
        self.check_transport_contents('new text for a\n', t, 'a')
266
267
        self.assertRaises(NoSuchFile,
268
                          t.put_bytes, 'path/doesnt/exist/c', 'contents')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
269
1955.3.27 by John Arbash Meinel
rename non_atomic_put_* to put_*non_atomic, and re-order the functions
270
    def test_put_bytes_non_atomic(self):
271
        t = self.get_transport()
272
273
        if t.is_readonly():
274
            self.assertRaises(TransportNotPossible,
275
                    t.put_bytes_non_atomic, 'a', 'some text for a\n')
276
            return
277
278
        self.failIf(t.has('a'))
279
        t.put_bytes_non_atomic('a', 'some text for a\n')
280
        self.failUnless(t.has('a'))
281
        self.check_transport_contents('some text for a\n', t, 'a')
282
        # Put also replaces contents
283
        t.put_bytes_non_atomic('a', 'new\ncontents for\na\n')
284
        self.check_transport_contents('new\ncontents for\na\n', t, 'a')
285
286
        # Make sure we can create another file
287
        t.put_bytes_non_atomic('d', 'contents for\nd\n')
288
        # And overwrite 'a' with empty contents
289
        t.put_bytes_non_atomic('a', '')
290
        self.check_transport_contents('contents for\nd\n', t, 'd')
291
        self.check_transport_contents('', t, 'a')
292
293
        self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'no/such/path',
294
                                       'contents\n')
295
        # Now test the create_parent flag
296
        self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'dir/a',
297
                                       'contents\n')
298
        self.failIf(t.has('dir/a'))
299
        t.put_bytes_non_atomic('dir/a', 'contents for dir/a\n',
300
                               create_parent_dir=True)
301
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
302
        
303
        # But we still get NoSuchFile if we can't make the parent dir
304
        self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'not/there/a',
305
                                       'contents\n',
306
                                       create_parent_dir=True)
307
308
    def test_put_bytes_permissions(self):
309
        t = self.get_transport()
310
311
        if t.is_readonly():
312
            return
313
        if not t._can_roundtrip_unix_modebits():
314
            # Can't roundtrip, so no need to run this test
315
            return
316
        t.put_bytes('mode644', 'test text\n', mode=0644)
317
        self.assertTransportMode(t, 'mode644', 0644)
318
        t.put_bytes('mode666', 'test text\n', mode=0666)
319
        self.assertTransportMode(t, 'mode666', 0666)
320
        t.put_bytes('mode600', 'test text\n', mode=0600)
321
        self.assertTransportMode(t, 'mode600', 0600)
322
        # Yes, you can put_bytes a file such that it becomes readonly
323
        t.put_bytes('mode400', 'test text\n', mode=0400)
324
        self.assertTransportMode(t, 'mode400', 0400)
325
326
        # The default permissions should be based on the current umask
327
        umask = osutils.get_umask()
328
        t.put_bytes('nomode', 'test text\n', mode=None)
329
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
330
        
331
    def test_put_bytes_non_atomic_permissions(self):
332
        t = self.get_transport()
333
334
        if t.is_readonly():
335
            return
336
        if not t._can_roundtrip_unix_modebits():
337
            # Can't roundtrip, so no need to run this test
338
            return
339
        t.put_bytes_non_atomic('mode644', 'test text\n', mode=0644)
340
        self.assertTransportMode(t, 'mode644', 0644)
341
        t.put_bytes_non_atomic('mode666', 'test text\n', mode=0666)
342
        self.assertTransportMode(t, 'mode666', 0666)
343
        t.put_bytes_non_atomic('mode600', 'test text\n', mode=0600)
344
        self.assertTransportMode(t, 'mode600', 0600)
345
        t.put_bytes_non_atomic('mode400', 'test text\n', mode=0400)
346
        self.assertTransportMode(t, 'mode400', 0400)
347
348
        # The default permissions should be based on the current umask
349
        umask = osutils.get_umask()
350
        t.put_bytes_non_atomic('nomode', 'test text\n', mode=None)
351
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
1946.2.12 by John Arbash Meinel
Add ability to pass a directory mode to non_atomic_put
352
353
        # We should also be able to set the mode for a parent directory
354
        # when it is created
355
        t.put_bytes_non_atomic('dir700/mode664', 'test text\n', mode=0664,
356
                               dir_mode=0700, create_parent_dir=True)
357
        self.assertTransportMode(t, 'dir700', 0700)
358
        t.put_bytes_non_atomic('dir770/mode664', 'test text\n', mode=0664,
359
                               dir_mode=0770, create_parent_dir=True)
360
        self.assertTransportMode(t, 'dir770', 0770)
361
        t.put_bytes_non_atomic('dir777/mode664', 'test text\n', mode=0664,
362
                               dir_mode=0777, create_parent_dir=True)
363
        self.assertTransportMode(t, 'dir777', 0777)
1955.3.27 by John Arbash Meinel
rename non_atomic_put_* to put_*non_atomic, and re-order the functions
364
        
365
    def test_put_file(self):
366
        t = self.get_transport()
367
368
        if t.is_readonly():
369
            self.assertRaises(TransportNotPossible,
370
                    t.put_file, 'a', StringIO('some text for a\n'))
371
            return
372
373
        t.put_file('a', StringIO('some text for a\n'))
374
        self.failUnless(t.has('a'))
375
        self.check_transport_contents('some text for a\n', t, 'a')
376
        # Put also replaces contents
377
        t.put_file('a', StringIO('new\ncontents for\na\n'))
378
        self.check_transport_contents('new\ncontents for\na\n', t, 'a')
379
        self.assertRaises(NoSuchFile,
380
                          t.put_file, 'path/doesnt/exist/c',
381
                              StringIO('contents'))
382
383
    def test_put_file_non_atomic(self):
384
        t = self.get_transport()
385
386
        if t.is_readonly():
387
            self.assertRaises(TransportNotPossible,
388
                    t.put_file_non_atomic, 'a', StringIO('some text for a\n'))
389
            return
390
391
        self.failIf(t.has('a'))
392
        t.put_file_non_atomic('a', StringIO('some text for a\n'))
393
        self.failUnless(t.has('a'))
394
        self.check_transport_contents('some text for a\n', t, 'a')
395
        # Put also replaces contents
396
        t.put_file_non_atomic('a', StringIO('new\ncontents for\na\n'))
397
        self.check_transport_contents('new\ncontents for\na\n', t, 'a')
398
399
        # Make sure we can create another file
400
        t.put_file_non_atomic('d', StringIO('contents for\nd\n'))
401
        # And overwrite 'a' with empty contents
402
        t.put_file_non_atomic('a', StringIO(''))
403
        self.check_transport_contents('contents for\nd\n', t, 'd')
404
        self.check_transport_contents('', t, 'a')
405
406
        self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'no/such/path',
407
                                       StringIO('contents\n'))
408
        # Now test the create_parent flag
409
        self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'dir/a',
410
                                       StringIO('contents\n'))
411
        self.failIf(t.has('dir/a'))
412
        t.put_file_non_atomic('dir/a', StringIO('contents for dir/a\n'),
1955.3.20 by John Arbash Meinel
Add non_atomic_put_bytes() and tests for it
413
                              create_parent_dir=True)
1946.1.8 by John Arbash Meinel
Update non_atomic_put to have a create_parent_dir flag
414
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
415
        
416
        # But we still get NoSuchFile if we can't make the parent dir
1955.3.27 by John Arbash Meinel
rename non_atomic_put_* to put_*non_atomic, and re-order the functions
417
        self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'not/there/a',
1955.3.20 by John Arbash Meinel
Add non_atomic_put_bytes() and tests for it
418
                                       StringIO('contents\n'),
419
                                       create_parent_dir=True)
420
1955.3.7 by John Arbash Meinel
Fix the deprecation warnings in the transport tests themselves
421
    def test_put_file_permissions(self):
1955.3.18 by John Arbash Meinel
[merge] Transport.non_atomic_put()
422
1530.1.15 by Robert Collins
Move put mode tests into test_transport_implementation.
423
        t = self.get_transport()
424
425
        if t.is_readonly():
426
            return
1711.4.32 by John Arbash Meinel
Skip permission tests on win32 no modebits
427
        if not t._can_roundtrip_unix_modebits():
428
            # Can't roundtrip, so no need to run this test
429
            return
1955.3.7 by John Arbash Meinel
Fix the deprecation warnings in the transport tests themselves
430
        t.put_file('mode644', StringIO('test text\n'), mode=0644)
1530.1.21 by Robert Collins
Review feedback fixes.
431
        self.assertTransportMode(t, 'mode644', 0644)
1955.3.7 by John Arbash Meinel
Fix the deprecation warnings in the transport tests themselves
432
        t.put_file('mode666', StringIO('test text\n'), mode=0666)
1530.1.21 by Robert Collins
Review feedback fixes.
433
        self.assertTransportMode(t, 'mode666', 0666)
1955.3.7 by John Arbash Meinel
Fix the deprecation warnings in the transport tests themselves
434
        t.put_file('mode600', StringIO('test text\n'), mode=0600)
1530.1.21 by Robert Collins
Review feedback fixes.
435
        self.assertTransportMode(t, 'mode600', 0600)
1530.1.15 by Robert Collins
Move put mode tests into test_transport_implementation.
436
        # Yes, you can put a file such that it becomes readonly
1955.3.7 by John Arbash Meinel
Fix the deprecation warnings in the transport tests themselves
437
        t.put_file('mode400', StringIO('test text\n'), mode=0400)
1530.1.21 by Robert Collins
Review feedback fixes.
438
        self.assertTransportMode(t, 'mode400', 0400)
1955.3.7 by John Arbash Meinel
Fix the deprecation warnings in the transport tests themselves
439
440
        # XXX: put_multi is deprecated, so do we really care anymore?
1955.3.29 by John Arbash Meinel
Use applyDeprecated instead of callDeprecated
441
        self.applyDeprecated(zero_eleven, t.put_multi,
442
                             [('mmode644', StringIO('text\n'))], mode=0644)
1530.1.21 by Robert Collins
Review feedback fixes.
443
        self.assertTransportMode(t, 'mmode644', 0644)
1755.3.7 by John Arbash Meinel
Clean up and write tests for permissions. Now we use fstat which should be cheap, and lets us check the permissions and the file size
444
445
        # The default permissions should be based on the current umask
446
        umask = osutils.get_umask()
1955.3.7 by John Arbash Meinel
Fix the deprecation warnings in the transport tests themselves
447
        t.put_file('nomode', StringIO('test text\n'), mode=None)
1755.3.7 by John Arbash Meinel
Clean up and write tests for permissions. Now we use fstat which should be cheap, and lets us check the permissions and the file size
448
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
449
        
1955.3.27 by John Arbash Meinel
rename non_atomic_put_* to put_*non_atomic, and re-order the functions
450
    def test_put_file_non_atomic_permissions(self):
451
        t = self.get_transport()
452
453
        if t.is_readonly():
454
            return
455
        if not t._can_roundtrip_unix_modebits():
456
            # Can't roundtrip, so no need to run this test
457
            return
458
        t.put_file_non_atomic('mode644', StringIO('test text\n'), mode=0644)
459
        self.assertTransportMode(t, 'mode644', 0644)
460
        t.put_file_non_atomic('mode666', StringIO('test text\n'), mode=0666)
461
        self.assertTransportMode(t, 'mode666', 0666)
462
        t.put_file_non_atomic('mode600', StringIO('test text\n'), mode=0600)
463
        self.assertTransportMode(t, 'mode600', 0600)
464
        # Yes, you can put_file_non_atomic a file such that it becomes readonly
465
        t.put_file_non_atomic('mode400', StringIO('test text\n'), mode=0400)
466
        self.assertTransportMode(t, 'mode400', 0400)
467
468
        # The default permissions should be based on the current umask
469
        umask = osutils.get_umask()
470
        t.put_file_non_atomic('nomode', StringIO('test text\n'), mode=None)
471
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
472
        
1946.2.12 by John Arbash Meinel
Add ability to pass a directory mode to non_atomic_put
473
        # We should also be able to set the mode for a parent directory
474
        # when it is created
475
        sio = StringIO()
476
        t.put_file_non_atomic('dir700/mode664', sio, mode=0664,
477
                              dir_mode=0700, create_parent_dir=True)
478
        self.assertTransportMode(t, 'dir700', 0700)
479
        t.put_file_non_atomic('dir770/mode664', sio, mode=0664,
480
                              dir_mode=0770, create_parent_dir=True)
481
        self.assertTransportMode(t, 'dir770', 0770)
482
        t.put_file_non_atomic('dir777/mode664', sio, mode=0664,
483
                              dir_mode=0777, create_parent_dir=True)
484
        self.assertTransportMode(t, 'dir777', 0777)
485
2018.5.131 by Andrew Bennetts
Be strict about unicode passed to transport.put_{bytes,file} and SmartClient.call_with_body_bytes, fixing part of TestLockableFiles_RemoteLockDir.test_read_write.
486
    def test_put_bytes_unicode(self):
487
        # Expect put_bytes to raise AssertionError or UnicodeEncodeError if
488
        # given unicode "bytes".  UnicodeEncodeError doesn't really make sense
489
        # (we don't want to encode unicode here at all, callers should be
490
        # strictly passing bytes to put_bytes), but we allow it for backwards
491
        # compatibility.  At some point we should use a specific exception.
2414.1.2 by Andrew Bennetts
Deal with review comments.
492
        # See https://bugs.launchpad.net/bzr/+bug/106898.
2018.5.131 by Andrew Bennetts
Be strict about unicode passed to transport.put_{bytes,file} and SmartClient.call_with_body_bytes, fixing part of TestLockableFiles_RemoteLockDir.test_read_write.
493
        t = self.get_transport()
494
        if t.is_readonly():
495
            return
496
        unicode_string = u'\u1234'
497
        self.assertRaises(
498
            (AssertionError, UnicodeEncodeError),
499
            t.put_bytes, 'foo', unicode_string)
500
501
    def test_put_file_unicode(self):
502
        # Like put_bytes, except with a StringIO.StringIO of a unicode string.
503
        # This situation can happen (and has) if code is careless about the type
504
        # of "string" they initialise/write to a StringIO with.  We cannot use
505
        # cStringIO, because it never returns unicode from read.
506
        # Like put_bytes, UnicodeEncodeError isn't quite the right exception to
507
        # raise, but we raise it for hysterical raisins.
508
        t = self.get_transport()
509
        if t.is_readonly():
510
            return
511
        unicode_file = pyStringIO(u'\u1234')
512
        self.assertRaises(UnicodeEncodeError, t.put_file, 'foo', unicode_file)
513
1955.3.27 by John Arbash Meinel
rename non_atomic_put_* to put_*non_atomic, and re-order the functions
514
    def test_put_multi(self):
515
        t = self.get_transport()
516
517
        if t.is_readonly():
518
            return
1955.3.29 by John Arbash Meinel
Use applyDeprecated instead of callDeprecated
519
        self.assertEqual(2, self.applyDeprecated(zero_eleven,
1955.3.27 by John Arbash Meinel
rename non_atomic_put_* to put_*non_atomic, and re-order the functions
520
            t.put_multi, [('a', StringIO('new\ncontents for\na\n')),
521
                          ('d', StringIO('contents\nfor d\n'))]
522
            ))
523
        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd'])),
524
                [True, False, False, True])
525
        self.check_transport_contents('new\ncontents for\na\n', t, 'a')
526
        self.check_transport_contents('contents\nfor d\n', t, 'd')
527
1955.3.29 by John Arbash Meinel
Use applyDeprecated instead of callDeprecated
528
        self.assertEqual(2, self.applyDeprecated(zero_eleven,
1955.3.27 by John Arbash Meinel
rename non_atomic_put_* to put_*non_atomic, and re-order the functions
529
            t.put_multi, iter([('a', StringIO('diff\ncontents for\na\n')),
530
                              ('d', StringIO('another contents\nfor d\n'))])
531
            ))
532
        self.check_transport_contents('diff\ncontents for\na\n', t, 'a')
533
        self.check_transport_contents('another contents\nfor d\n', t, 'd')
534
1530.1.15 by Robert Collins
Move put mode tests into test_transport_implementation.
535
    def test_put_permissions(self):
536
        t = self.get_transport()
537
538
        if t.is_readonly():
539
            return
1711.4.32 by John Arbash Meinel
Skip permission tests on win32 no modebits
540
        if not t._can_roundtrip_unix_modebits():
541
            # Can't roundtrip, so no need to run this test
542
            return
1910.15.6 by Andrew Bennetts
Merge from bzr.dev
543
        self.applyDeprecated(zero_eleven, t.put, 'mode644',
544
                             StringIO('test text\n'), mode=0644)
1530.1.21 by Robert Collins
Review feedback fixes.
545
        self.assertTransportMode(t, 'mode644', 0644)
1910.15.6 by Andrew Bennetts
Merge from bzr.dev
546
        self.applyDeprecated(zero_eleven, t.put, 'mode666',
547
                             StringIO('test text\n'), mode=0666)
1530.1.21 by Robert Collins
Review feedback fixes.
548
        self.assertTransportMode(t, 'mode666', 0666)
1910.15.6 by Andrew Bennetts
Merge from bzr.dev
549
        self.applyDeprecated(zero_eleven, t.put, 'mode600',
550
                             StringIO('test text\n'), mode=0600)
1530.1.21 by Robert Collins
Review feedback fixes.
551
        self.assertTransportMode(t, 'mode600', 0600)
1530.1.15 by Robert Collins
Move put mode tests into test_transport_implementation.
552
        # Yes, you can put a file such that it becomes readonly
1910.15.6 by Andrew Bennetts
Merge from bzr.dev
553
        self.applyDeprecated(zero_eleven, t.put, 'mode400',
554
                             StringIO('test text\n'), mode=0400)
1530.1.21 by Robert Collins
Review feedback fixes.
555
        self.assertTransportMode(t, 'mode400', 0400)
1910.15.6 by Andrew Bennetts
Merge from bzr.dev
556
        self.applyDeprecated(zero_eleven, t.put_multi,
557
                             [('mmode644', StringIO('text\n'))], mode=0644)
1530.1.21 by Robert Collins
Review feedback fixes.
558
        self.assertTransportMode(t, 'mmode644', 0644)
1755.3.7 by John Arbash Meinel
Clean up and write tests for permissions. Now we use fstat which should be cheap, and lets us check the permissions and the file size
559
560
        # The default permissions should be based on the current umask
561
        umask = osutils.get_umask()
1910.15.6 by Andrew Bennetts
Merge from bzr.dev
562
        self.applyDeprecated(zero_eleven, t.put, 'nomode',
563
                             StringIO('test text\n'), mode=None)
1755.3.7 by John Arbash Meinel
Clean up and write tests for permissions. Now we use fstat which should be cheap, and lets us check the permissions and the file size
564
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
565
        
1530.1.3 by Robert Collins
transport implementations now tested consistently.
566
    def test_mkdir(self):
567
        t = self.get_transport()
568
569
        if t.is_readonly():
570
            # cannot mkdir on readonly transports. We're not testing for 
571
            # cache coherency because cache behaviour is not currently
572
            # defined for the transport interface.
573
            self.assertRaises(TransportNotPossible, t.mkdir, '.')
574
            self.assertRaises(TransportNotPossible, t.mkdir, 'new_dir')
575
            self.assertRaises(TransportNotPossible, t.mkdir_multi, ['new_dir'])
576
            self.assertRaises(TransportNotPossible, t.mkdir, 'path/doesnt/exist')
577
            return
578
        # Test mkdir
579
        t.mkdir('dir_a')
580
        self.assertEqual(t.has('dir_a'), True)
581
        self.assertEqual(t.has('dir_b'), False)
582
583
        t.mkdir('dir_b')
584
        self.assertEqual(t.has('dir_b'), True)
585
586
        t.mkdir_multi(['dir_c', 'dir_d'])
587
588
        t.mkdir_multi(iter(['dir_e', 'dir_f']))
589
        self.assertEqual(list(t.has_multi(
590
            ['dir_a', 'dir_b', 'dir_c', 'dir_q',
591
             'dir_d', 'dir_e', 'dir_f', 'dir_b'])),
592
            [True, True, True, False,
593
             True, True, True, True])
594
595
        # we were testing that a local mkdir followed by a transport
596
        # mkdir failed thusly, but given that we * in one process * do not
597
        # concurrently fiddle with disk dirs and then use transport to do 
598
        # things, the win here seems marginal compared to the constraint on
599
        # the interface. RBC 20051227
600
        t.mkdir('dir_g')
601
        self.assertRaises(FileExists, t.mkdir, 'dir_g')
602
603
        # Test get/put in sub-directories
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
604
        t.put_bytes('dir_a/a', 'contents of dir_a/a')
605
        t.put_file('dir_b/b', StringIO('contents of dir_b/b'))
1530.1.3 by Robert Collins
transport implementations now tested consistently.
606
        self.check_transport_contents('contents of dir_a/a', t, 'dir_a/a')
607
        self.check_transport_contents('contents of dir_b/b', t, 'dir_b/b')
608
1530.1.4 by Robert Collins
integrate Memory tests into transport interface tests.
609
        # mkdir of a dir with an absent parent
610
        self.assertRaises(NoSuchFile, t.mkdir, 'missing/dir')
611
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
612
    def test_mkdir_permissions(self):
613
        t = self.get_transport()
614
        if t.is_readonly():
615
            return
1608.2.7 by Martin Pool
Rename supports_unix_modebits to _can_roundtrip_unix_modebits for clarity
616
        if not t._can_roundtrip_unix_modebits():
1608.2.5 by Martin Pool
Add Transport.supports_unix_modebits, so tests can
617
            # no sense testing on this transport
618
            return
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
619
        # Test mkdir with a mode
620
        t.mkdir('dmode755', mode=0755)
1530.1.21 by Robert Collins
Review feedback fixes.
621
        self.assertTransportMode(t, 'dmode755', 0755)
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
622
        t.mkdir('dmode555', mode=0555)
1530.1.21 by Robert Collins
Review feedback fixes.
623
        self.assertTransportMode(t, 'dmode555', 0555)
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
624
        t.mkdir('dmode777', mode=0777)
1530.1.21 by Robert Collins
Review feedback fixes.
625
        self.assertTransportMode(t, 'dmode777', 0777)
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
626
        t.mkdir('dmode700', mode=0700)
1530.1.21 by Robert Collins
Review feedback fixes.
627
        self.assertTransportMode(t, 'dmode700', 0700)
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
628
        t.mkdir_multi(['mdmode755'], mode=0755)
1530.1.21 by Robert Collins
Review feedback fixes.
629
        self.assertTransportMode(t, 'mdmode755', 0755)
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
630
1755.3.7 by John Arbash Meinel
Clean up and write tests for permissions. Now we use fstat which should be cheap, and lets us check the permissions and the file size
631
        # Default mode should be based on umask
632
        umask = osutils.get_umask()
633
        t.mkdir('dnomode', mode=None)
634
        self.assertTransportMode(t, 'dnomode', 0777 & ~umask)
635
2671.3.2 by Robert Collins
Start open_file_stream logic.
636
    def test_opening_a_file_stream_creates_file(self):
637
        t = self.get_transport()
638
        if t.is_readonly():
639
            return
640
        handle = t.open_file_stream('foo')
641
        try:
642
            self.assertEqual('', t.get_bytes('foo'))
643
        finally:
644
            t.close_file_stream('foo')
645
1530.1.3 by Robert Collins
transport implementations now tested consistently.
646
    def test_copy_to(self):
1534.4.21 by Robert Collins
Extend the copy_to tests to smoke test server-to-same-server copies to catch optimised code paths, and fix sftps optimised code path by removing dead code.
647
        # FIXME: test:   same server to same server (partly done)
648
        # same protocol two servers
649
        # and    different protocols (done for now except for MemoryTransport.
650
        # - RBC 20060122
651
652
        def simple_copy_files(transport_from, transport_to):
653
            files = ['a', 'b', 'c', 'd']
654
            self.build_tree(files, transport=transport_from)
1563.2.3 by Robert Collins
Change the return signature of transport.append and append_multi to return the length of the pre-append content.
655
            self.assertEqual(4, transport_from.copy_to(files, transport_to))
1534.4.21 by Robert Collins
Extend the copy_to tests to smoke test server-to-same-server copies to catch optimised code paths, and fix sftps optimised code path by removing dead code.
656
            for f in files:
657
                self.check_transport_contents(transport_to.get(f).read(),
658
                                              transport_from, f)
659
1530.1.3 by Robert Collins
transport implementations now tested consistently.
660
        t = self.get_transport()
1685.1.42 by John Arbash Meinel
A couple more fixes to make sure memory:/// works correctly.
661
        temp_transport = MemoryTransport('memory:///')
1534.4.21 by Robert Collins
Extend the copy_to tests to smoke test server-to-same-server copies to catch optimised code paths, and fix sftps optimised code path by removing dead code.
662
        simple_copy_files(t, temp_transport)
663
        if not t.is_readonly():
664
            t.mkdir('copy_to_simple')
665
            t2 = t.clone('copy_to_simple')
666
            simple_copy_files(t, t2)
1530.1.3 by Robert Collins
transport implementations now tested consistently.
667
668
669
        # Test that copying into a missing directory raises
670
        # NoSuchFile
671
        if t.is_readonly():
1530.1.21 by Robert Collins
Review feedback fixes.
672
            self.build_tree(['e/', 'e/f'])
1530.1.3 by Robert Collins
transport implementations now tested consistently.
673
        else:
674
            t.mkdir('e')
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
675
            t.put_bytes('e/f', 'contents of e')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
676
        self.assertRaises(NoSuchFile, t.copy_to, ['e/f'], temp_transport)
677
        temp_transport.mkdir('e')
678
        t.copy_to(['e/f'], temp_transport)
679
680
        del temp_transport
1685.1.42 by John Arbash Meinel
A couple more fixes to make sure memory:/// works correctly.
681
        temp_transport = MemoryTransport('memory:///')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
682
683
        files = ['a', 'b', 'c', 'd']
684
        t.copy_to(iter(files), temp_transport)
685
        for f in files:
686
            self.check_transport_contents(temp_transport.get(f).read(),
687
                                          t, f)
688
        del temp_transport
689
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
690
        for mode in (0666, 0644, 0600, 0400):
1685.1.42 by John Arbash Meinel
A couple more fixes to make sure memory:/// works correctly.
691
            temp_transport = MemoryTransport("memory:///")
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
692
            t.copy_to(files, temp_transport, mode=mode)
693
            for f in files:
1530.1.21 by Robert Collins
Review feedback fixes.
694
                self.assertTransportMode(temp_transport, f, mode)
1530.1.16 by Robert Collins
Move mkdir and copy_to permissions tests to test_transport_impleentation.
695
1530.1.3 by Robert Collins
transport implementations now tested consistently.
696
    def test_append(self):
697
        t = self.get_transport()
698
699
        if t.is_readonly():
1955.3.15 by John Arbash Meinel
Deprecate 'Transport.append' in favor of Transport.append_file or Transport.append_bytes
700
            return
701
        t.put_bytes('a', 'diff\ncontents for\na\n')
702
        t.put_bytes('b', 'contents\nfor b\n')
703
1955.3.29 by John Arbash Meinel
Use applyDeprecated instead of callDeprecated
704
        self.assertEqual(20, self.applyDeprecated(zero_eleven,
1955.3.15 by John Arbash Meinel
Deprecate 'Transport.append' in favor of Transport.append_file or Transport.append_bytes
705
            t.append, 'a', StringIO('add\nsome\nmore\ncontents\n')))
706
707
        self.check_transport_contents(
708
            'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
709
            t, 'a')
710
711
        # And we can create new files, too
1955.3.29 by John Arbash Meinel
Use applyDeprecated instead of callDeprecated
712
        self.assertEqual(0, self.applyDeprecated(zero_eleven,
1955.3.15 by John Arbash Meinel
Deprecate 'Transport.append' in favor of Transport.append_file or Transport.append_bytes
713
            t.append, 'c', StringIO('some text\nfor a missing file\n')))
714
        self.check_transport_contents('some text\nfor a missing file\n',
715
                                      t, 'c')
716
    def test_append_file(self):
717
        t = self.get_transport()
718
719
        if t.is_readonly():
1530.1.3 by Robert Collins
transport implementations now tested consistently.
720
            self.assertRaises(TransportNotPossible,
1955.3.15 by John Arbash Meinel
Deprecate 'Transport.append' in favor of Transport.append_file or Transport.append_bytes
721
                    t.append_file, 'a', 'add\nsome\nmore\ncontents\n')
1955.3.2 by John Arbash Meinel
Implement and test 'Transport.append_bytes', cleanup the tests of plain append
722
            return
1955.3.10 by John Arbash Meinel
clean up append, append_bytes, and append_multi tests
723
        t.put_bytes('a', 'diff\ncontents for\na\n')
724
        t.put_bytes('b', 'contents\nfor b\n')
1955.3.2 by John Arbash Meinel
Implement and test 'Transport.append_bytes', cleanup the tests of plain append
725
726
        self.assertEqual(20,
1955.3.15 by John Arbash Meinel
Deprecate 'Transport.append' in favor of Transport.append_file or Transport.append_bytes
727
            t.append_file('a', StringIO('add\nsome\nmore\ncontents\n')))
1530.1.3 by Robert Collins
transport implementations now tested consistently.
728
729
        self.check_transport_contents(
730
            'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
731
            t, 'a')
732
1955.3.10 by John Arbash Meinel
clean up append, append_bytes, and append_multi tests
733
        # a file with no parent should fail..
734
        self.assertRaises(NoSuchFile,
1955.3.15 by John Arbash Meinel
Deprecate 'Transport.append' in favor of Transport.append_file or Transport.append_bytes
735
                          t.append_file, 'missing/path', StringIO('content'))
1955.3.10 by John Arbash Meinel
clean up append, append_bytes, and append_multi tests
736
737
        # And we can create new files, too
738
        self.assertEqual(0,
1955.3.15 by John Arbash Meinel
Deprecate 'Transport.append' in favor of Transport.append_file or Transport.append_bytes
739
            t.append_file('c', StringIO('some text\nfor a missing file\n')))
1955.3.10 by John Arbash Meinel
clean up append, append_bytes, and append_multi tests
740
        self.check_transport_contents('some text\nfor a missing file\n',
741
                                      t, 'c')
742
743
    def test_append_bytes(self):
744
        t = self.get_transport()
745
746
        if t.is_readonly():
747
            self.assertRaises(TransportNotPossible,
748
                    t.append_bytes, 'a', 'add\nsome\nmore\ncontents\n')
749
            return
750
751
        self.assertEqual(0, t.append_bytes('a', 'diff\ncontents for\na\n'))
752
        self.assertEqual(0, t.append_bytes('b', 'contents\nfor b\n'))
753
754
        self.assertEqual(20,
755
            t.append_bytes('a', 'add\nsome\nmore\ncontents\n'))
756
757
        self.check_transport_contents(
758
            'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
759
            t, 'a')
760
761
        # a file with no parent should fail..
762
        self.assertRaises(NoSuchFile,
763
                          t.append_bytes, 'missing/path', 'content')
764
765
    def test_append_multi(self):
766
        t = self.get_transport()
767
768
        if t.is_readonly():
769
            return
770
        t.put_bytes('a', 'diff\ncontents for\na\n'
771
                         'add\nsome\nmore\ncontents\n')
772
        t.put_bytes('b', 'contents\nfor b\n')
773
1955.3.2 by John Arbash Meinel
Implement and test 'Transport.append_bytes', cleanup the tests of plain append
774
        self.assertEqual((43, 15),
775
            t.append_multi([('a', StringIO('and\nthen\nsome\nmore\n')),
776
                            ('b', StringIO('some\nmore\nfor\nb\n'))]))
777
1530.1.3 by Robert Collins
transport implementations now tested consistently.
778
        self.check_transport_contents(
779
            'diff\ncontents for\na\n'
780
            'add\nsome\nmore\ncontents\n'
781
            'and\nthen\nsome\nmore\n',
782
            t, 'a')
783
        self.check_transport_contents(
784
                'contents\nfor b\n'
785
                'some\nmore\nfor\nb\n',
786
                t, 'b')
787
1955.3.2 by John Arbash Meinel
Implement and test 'Transport.append_bytes', cleanup the tests of plain append
788
        self.assertEqual((62, 31),
789
            t.append_multi(iter([('a', StringIO('a little bit more\n')),
790
                                 ('b', StringIO('from an iterator\n'))])))
1530.1.3 by Robert Collins
transport implementations now tested consistently.
791
        self.check_transport_contents(
792
            'diff\ncontents for\na\n'
793
            'add\nsome\nmore\ncontents\n'
794
            'and\nthen\nsome\nmore\n'
795
            'a little bit more\n',
796
            t, 'a')
797
        self.check_transport_contents(
798
                'contents\nfor b\n'
799
                'some\nmore\nfor\nb\n'
800
                'from an iterator\n',
801
                t, 'b')
802
1955.3.2 by John Arbash Meinel
Implement and test 'Transport.append_bytes', cleanup the tests of plain append
803
        self.assertEqual((80, 0),
804
            t.append_multi([('a', StringIO('some text in a\n')),
805
                            ('d', StringIO('missing file r\n'))]))
806
1530.1.3 by Robert Collins
transport implementations now tested consistently.
807
        self.check_transport_contents(
808
            'diff\ncontents for\na\n'
809
            'add\nsome\nmore\ncontents\n'
810
            'and\nthen\nsome\nmore\n'
811
            'a little bit more\n'
812
            'some text in a\n',
813
            t, 'a')
814
        self.check_transport_contents('missing file r\n', t, 'd')
815
1955.3.15 by John Arbash Meinel
Deprecate 'Transport.append' in favor of Transport.append_file or Transport.append_bytes
816
    def test_append_file_mode(self):
1955.3.10 by John Arbash Meinel
clean up append, append_bytes, and append_multi tests
817
        """Check that append accepts a mode parameter"""
1666.1.6 by Robert Collins
Make knit the default format.
818
        # check append accepts a mode
819
        t = self.get_transport()
820
        if t.is_readonly():
1955.3.10 by John Arbash Meinel
clean up append, append_bytes, and append_multi tests
821
            self.assertRaises(TransportNotPossible,
1955.3.15 by John Arbash Meinel
Deprecate 'Transport.append' in favor of Transport.append_file or Transport.append_bytes
822
                t.append_file, 'f', StringIO('f'), mode=None)
1666.1.6 by Robert Collins
Make knit the default format.
823
            return
1955.3.15 by John Arbash Meinel
Deprecate 'Transport.append' in favor of Transport.append_file or Transport.append_bytes
824
        t.append_file('f', StringIO('f'), mode=None)
1666.1.6 by Robert Collins
Make knit the default format.
825
        
1955.3.2 by John Arbash Meinel
Implement and test 'Transport.append_bytes', cleanup the tests of plain append
826
    def test_append_bytes_mode(self):
827
        # check append_bytes accepts a mode
828
        t = self.get_transport()
829
        if t.is_readonly():
1955.3.10 by John Arbash Meinel
clean up append, append_bytes, and append_multi tests
830
            self.assertRaises(TransportNotPossible,
831
                t.append_bytes, 'f', 'f', mode=None)
1955.3.2 by John Arbash Meinel
Implement and test 'Transport.append_bytes', cleanup the tests of plain append
832
            return
1955.3.10 by John Arbash Meinel
clean up append, append_bytes, and append_multi tests
833
        t.append_bytes('f', 'f', mode=None)
1955.3.2 by John Arbash Meinel
Implement and test 'Transport.append_bytes', cleanup the tests of plain append
834
        
1530.1.3 by Robert Collins
transport implementations now tested consistently.
835
    def test_delete(self):
836
        # TODO: Test Transport.delete
837
        t = self.get_transport()
838
839
        # Not much to do with a readonly transport
840
        if t.is_readonly():
1534.4.9 by Robert Collins
Add a readonly decorator for transports.
841
            self.assertRaises(TransportNotPossible, t.delete, 'missing')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
842
            return
843
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
844
        t.put_bytes('a', 'a little bit of text\n')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
845
        self.failUnless(t.has('a'))
846
        t.delete('a')
847
        self.failIf(t.has('a'))
848
849
        self.assertRaises(NoSuchFile, t.delete, 'a')
850
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
851
        t.put_bytes('a', 'a text\n')
852
        t.put_bytes('b', 'b text\n')
853
        t.put_bytes('c', 'c text\n')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
854
        self.assertEqual([True, True, True],
855
                list(t.has_multi(['a', 'b', 'c'])))
856
        t.delete_multi(['a', 'c'])
857
        self.assertEqual([False, True, False],
858
                list(t.has_multi(['a', 'b', 'c'])))
859
        self.failIf(t.has('a'))
860
        self.failUnless(t.has('b'))
861
        self.failIf(t.has('c'))
862
863
        self.assertRaises(NoSuchFile,
864
                t.delete_multi, ['a', 'b', 'c'])
865
866
        self.assertRaises(NoSuchFile,
867
                t.delete_multi, iter(['a', 'b', 'c']))
868
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
869
        t.put_bytes('a', 'another a text\n')
870
        t.put_bytes('c', 'another c text\n')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
871
        t.delete_multi(iter(['a', 'b', 'c']))
872
873
        # We should have deleted everything
874
        # SftpServer creates control files in the
875
        # working directory, so we can just do a
876
        # plain "listdir".
877
        # self.assertEqual([], os.listdir('.'))
878
2671.3.1 by Robert Collins
* New method ``bzrlib.transport.Transport.get_recommended_page_size``.
879
    def test_recommended_page_size(self):
880
        """Transports recommend a page size for partial access to files."""
881
        t = self.get_transport()
882
        self.assertIsInstance(t.recommended_page_size(), int)
883
1534.4.15 by Robert Collins
Remove shutil dependency in upgrade - create a delete_tree method for transports.
884
    def test_rmdir(self):
885
        t = self.get_transport()
886
        # Not much to do with a readonly transport
887
        if t.is_readonly():
888
            self.assertRaises(TransportNotPossible, t.rmdir, 'missing')
889
            return
890
        t.mkdir('adir')
891
        t.mkdir('adir/bdir')
892
        t.rmdir('adir/bdir')
1948.3.12 by Vincent LADEUIL
Fix Aaron's third review remarks.
893
        # ftp may not be able to raise NoSuchFile for lack of
894
        # details when failing
895
        self.assertRaises((NoSuchFile, PathError), t.rmdir, 'adir/bdir')
1534.4.15 by Robert Collins
Remove shutil dependency in upgrade - create a delete_tree method for transports.
896
        t.rmdir('adir')
1948.3.12 by Vincent LADEUIL
Fix Aaron's third review remarks.
897
        self.assertRaises((NoSuchFile, PathError), t.rmdir, 'adir')
1534.4.15 by Robert Collins
Remove shutil dependency in upgrade - create a delete_tree method for transports.
898
1553.5.10 by Martin Pool
New DirectoryNotEmpty exception, and raise this from local and memory
899
    def test_rmdir_not_empty(self):
900
        """Deleting a non-empty directory raises an exception
901
        
902
        sftp (and possibly others) don't give us a specific "directory not
903
        empty" exception -- we can just see that the operation failed.
904
        """
905
        t = self.get_transport()
906
        if t.is_readonly():
907
            return
908
        t.mkdir('adir')
909
        t.mkdir('adir/bdir')
910
        self.assertRaises(PathError, t.rmdir, 'adir')
911
2338.5.1 by Andrew Bennetts
Fix bug in MemoryTransport.rmdir.
912
    def test_rmdir_empty_but_similar_prefix(self):
913
        """rmdir does not get confused by sibling paths.
914
        
915
        A naive implementation of MemoryTransport would refuse to rmdir
916
        ".bzr/branch" if there is a ".bzr/branch-format" directory, because it
917
        uses "path.startswith(dir)" on all file paths to determine if directory
918
        is empty.
919
        """
920
        t = self.get_transport()
921
        if t.is_readonly():
922
            return
923
        t.mkdir('foo')
924
        t.put_bytes('foo-bar', '')
925
        t.mkdir('foo-baz')
926
        t.rmdir('foo')
927
        self.assertRaises((NoSuchFile, PathError), t.rmdir, 'foo')
928
        self.failUnless(t.has('foo-bar'))
929
1553.5.13 by Martin Pool
New Transport.rename that mustn't overwrite
930
    def test_rename_dir_succeeds(self):
931
        t = self.get_transport()
932
        if t.is_readonly():
933
            raise TestSkipped("transport is readonly")
934
        t.mkdir('adir')
935
        t.mkdir('adir/asubdir')
936
        t.rename('adir', 'bdir')
937
        self.assertTrue(t.has('bdir/asubdir'))
938
        self.assertFalse(t.has('adir'))
939
940
    def test_rename_dir_nonempty(self):
941
        """Attempting to replace a nonemtpy directory should fail"""
942
        t = self.get_transport()
943
        if t.is_readonly():
944
            raise TestSkipped("transport is readonly")
945
        t.mkdir('adir')
946
        t.mkdir('adir/asubdir')
947
        t.mkdir('bdir')
948
        t.mkdir('bdir/bsubdir')
1910.7.17 by Andrew Bennetts
Various cosmetic changes.
949
        # any kind of PathError would be OK, though we normally expect
950
        # DirectoryNotEmpty
1553.5.13 by Martin Pool
New Transport.rename that mustn't overwrite
951
        self.assertRaises(PathError, t.rename, 'bdir', 'adir')
952
        # nothing was changed so it should still be as before
953
        self.assertTrue(t.has('bdir/bsubdir'))
954
        self.assertFalse(t.has('adir/bdir'))
955
        self.assertFalse(t.has('adir/bsubdir'))
956
1534.4.15 by Robert Collins
Remove shutil dependency in upgrade - create a delete_tree method for transports.
957
    def test_delete_tree(self):
958
        t = self.get_transport()
959
960
        # Not much to do with a readonly transport
961
        if t.is_readonly():
962
            self.assertRaises(TransportNotPossible, t.delete_tree, 'missing')
963
            return
964
965
        # and does it like listing ?
966
        t.mkdir('adir')
967
        try:
968
            t.delete_tree('adir')
969
        except TransportNotPossible:
970
            # ok, this transport does not support delete_tree
971
            return
972
        
973
        # did it delete that trivial case?
974
        self.assertRaises(NoSuchFile, t.stat, 'adir')
975
976
        self.build_tree(['adir/',
977
                         'adir/file', 
978
                         'adir/subdir/', 
979
                         'adir/subdir/file', 
980
                         'adir/subdir2/',
981
                         'adir/subdir2/file',
982
                         ], transport=t)
983
984
        t.delete_tree('adir')
985
        # adir should be gone now.
986
        self.assertRaises(NoSuchFile, t.stat, 'adir')
987
1530.1.3 by Robert Collins
transport implementations now tested consistently.
988
    def test_move(self):
989
        t = self.get_transport()
990
991
        if t.is_readonly():
992
            return
993
994
        # TODO: I would like to use os.listdir() to
995
        # make sure there are no extra files, but SftpServer
996
        # creates control files in the working directory
997
        # perhaps all of this could be done in a subdirectory
998
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
999
        t.put_bytes('a', 'a first file\n')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
1000
        self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
1001
1002
        t.move('a', 'b')
1003
        self.failUnless(t.has('b'))
1004
        self.failIf(t.has('a'))
1005
1006
        self.check_transport_contents('a first file\n', t, 'b')
1007
        self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
1008
1009
        # Overwrite a file
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
1010
        t.put_bytes('c', 'c this file\n')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
1011
        t.move('c', 'b')
1012
        self.failIf(t.has('c'))
1013
        self.check_transport_contents('c this file\n', t, 'b')
1014
1015
        # TODO: Try to write a test for atomicity
1016
        # TODO: Test moving into a non-existant subdirectory
1017
        # TODO: Test Transport.move_multi
1018
1019
    def test_copy(self):
1020
        t = self.get_transport()
1021
1022
        if t.is_readonly():
1023
            return
1024
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
1025
        t.put_bytes('a', 'a file\n')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
1026
        t.copy('a', 'b')
1027
        self.check_transport_contents('a file\n', t, 'b')
1028
1029
        self.assertRaises(NoSuchFile, t.copy, 'c', 'd')
1030
        os.mkdir('c')
1031
        # What should the assert be if you try to copy a
1032
        # file over a directory?
1033
        #self.assertRaises(Something, t.copy, 'a', 'c')
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
1034
        t.put_bytes('d', 'text in d\n')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
1035
        t.copy('d', 'b')
1036
        self.check_transport_contents('text in d\n', t, 'b')
1037
1038
        # TODO: test copy_multi
1039
1040
    def test_connection_error(self):
1910.7.17 by Andrew Bennetts
Various cosmetic changes.
1041
        """ConnectionError is raised when connection is impossible.
1042
        
1043
        The error may be raised from either the constructor or the first
1044
        operation on the transport.
1045
        """
1530.1.9 by Robert Collins
Test bogus urls with http in the new infrastructure.
1046
        try:
1047
            url = self._server.get_bogus_url()
1048
        except NotImplementedError:
1049
            raise TestSkipped("Transport %s has no bogus URL support." %
1050
                              self._server.__class__)
2485.8.38 by Vincent Ladeuil
Finish sftp refactoring. Test suite passing.
1051
        t = get_transport(url)
1052
        self.assertRaises((ConnectionError, NoSuchFile), t.get, '.bzr/branch')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
1053
1054
    def test_stat(self):
1055
        # TODO: Test stat, just try once, and if it throws, stop testing
1056
        from stat import S_ISDIR, S_ISREG
1057
1058
        t = self.get_transport()
1059
1060
        try:
1061
            st = t.stat('.')
1062
        except TransportNotPossible, e:
1063
            # This transport cannot stat
1064
            return
1065
1066
        paths = ['a', 'b/', 'b/c', 'b/d/', 'b/d/e']
1067
        sizes = [14, 0, 16, 0, 18] 
1551.2.39 by abentley
Fix line endings in tests
1068
        self.build_tree(paths, transport=t, line_endings='binary')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
1069
1070
        for path, size in zip(paths, sizes):
1071
            st = t.stat(path)
1072
            if path.endswith('/'):
1073
                self.failUnless(S_ISDIR(st.st_mode))
1074
                # directory sizes are meaningless
1075
            else:
1076
                self.failUnless(S_ISREG(st.st_mode))
1077
                self.assertEqual(size, st.st_size)
1078
1079
        remote_stats = list(t.stat_multi(paths))
1080
        remote_iter_stats = list(t.stat_multi(iter(paths)))
1081
1082
        self.assertRaises(NoSuchFile, t.stat, 'q')
1083
        self.assertRaises(NoSuchFile, t.stat, 'b/a')
1084
1085
        self.assertListRaises(NoSuchFile, t.stat_multi, ['a', 'c', 'd'])
1086
        self.assertListRaises(NoSuchFile, t.stat_multi, iter(['a', 'c', 'd']))
1534.4.15 by Robert Collins
Remove shutil dependency in upgrade - create a delete_tree method for transports.
1087
        self.build_tree(['subdir/', 'subdir/file'], transport=t)
1088
        subdir = t.clone('subdir')
1089
        subdir.stat('./file')
1534.4.26 by Robert Collins
Move working tree initialisation out from Branch.initialize, deprecated Branch.initialize to Branch.create.
1090
        subdir.stat('.')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
1091
1092
    def test_list_dir(self):
1093
        # TODO: Test list_dir, just try once, and if it throws, stop testing
1094
        t = self.get_transport()
1095
        
1096
        if not t.listable():
1097
            self.assertRaises(TransportNotPossible, t.list_dir, '.')
1098
            return
1099
1100
        def sorted_list(d):
1101
            l = list(t.list_dir(d))
1102
            l.sort()
1103
            return l
1104
1910.7.1 by Andrew Bennetts
Make sure list_dir always returns url-escaped names.
1105
        self.assertEqual([], sorted_list('.'))
1534.4.15 by Robert Collins
Remove shutil dependency in upgrade - create a delete_tree method for transports.
1106
        # c2 is precisely one letter longer than c here to test that
1107
        # suffixing is not confused.
1959.2.1 by John Arbash Meinel
David Allouche: Make transports return escaped paths
1108
        # a%25b checks that quoting is done consistently across transports
1109
        tree_names = ['a', 'a%25b', 'b', 'c/', 'c/d', 'c/e', 'c2/']
2120.3.1 by John Arbash Meinel
Fix MemoryTransport.list_dir() implementation, and update tests
1110
1534.4.9 by Robert Collins
Add a readonly decorator for transports.
1111
        if not t.is_readonly():
1959.2.1 by John Arbash Meinel
David Allouche: Make transports return escaped paths
1112
            self.build_tree(tree_names, transport=t)
1534.4.9 by Robert Collins
Add a readonly decorator for transports.
1113
        else:
2120.3.1 by John Arbash Meinel
Fix MemoryTransport.list_dir() implementation, and update tests
1114
            self.build_tree(tree_names)
1530.1.3 by Robert Collins
transport implementations now tested consistently.
1115
1959.2.1 by John Arbash Meinel
David Allouche: Make transports return escaped paths
1116
        self.assertEqual(
1959.2.3 by John Arbash Meinel
Remove some unicode string notations
1117
            ['a', 'a%2525b', 'b', 'c', 'c2'], sorted_list('.'))
1910.7.1 by Andrew Bennetts
Make sure list_dir always returns url-escaped names.
1118
        self.assertEqual(['d', 'e'], sorted_list('c'))
1530.1.3 by Robert Collins
transport implementations now tested consistently.
1119
1534.4.9 by Robert Collins
Add a readonly decorator for transports.
1120
        if not t.is_readonly():
1121
            t.delete('c/d')
1122
            t.delete('b')
1123
        else:
2120.3.1 by John Arbash Meinel
Fix MemoryTransport.list_dir() implementation, and update tests
1124
            os.unlink('c/d')
1125
            os.unlink('b')
1534.4.9 by Robert Collins
Add a readonly decorator for transports.
1126
            
1959.2.3 by John Arbash Meinel
Remove some unicode string notations
1127
        self.assertEqual(['a', 'a%2525b', 'c', 'c2'], sorted_list('.'))
1910.7.1 by Andrew Bennetts
Make sure list_dir always returns url-escaped names.
1128
        self.assertEqual(['e'], sorted_list('c'))
1530.1.3 by Robert Collins
transport implementations now tested consistently.
1129
1662.1.12 by Martin Pool
Translate unknown sftp errors to PathError, no NoSuchFile
1130
        self.assertListRaises(PathError, t.list_dir, 'q')
1131
        self.assertListRaises(PathError, t.list_dir, 'c/f')
1132
        self.assertListRaises(PathError, t.list_dir, 'a')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
1133
1910.7.1 by Andrew Bennetts
Make sure list_dir always returns url-escaped names.
1134
    def test_list_dir_result_is_url_escaped(self):
1135
        t = self.get_transport()
1136
        if not t.listable():
1137
            raise TestSkipped("transport not listable")
1138
1139
        if not t.is_readonly():
1140
            self.build_tree(['a/', 'a/%'], transport=t)
1141
        else:
1142
            self.build_tree(['a/', 'a/%'])
1143
        
1910.7.2 by Andrew Bennetts
Also assert that list_dir returns plain str objects.
1144
        names = list(t.list_dir('a'))
1145
        self.assertEqual(['%25'], names)
1146
        self.assertIsInstance(names[0], str)
1910.7.1 by Andrew Bennetts
Make sure list_dir always returns url-escaped names.
1147
2485.8.25 by Vincent Ladeuil
Separate abspath from _remote_path, the intents are different.
1148
    def test_clone_preserve_info(self):
1149
        t1 = self.get_transport()
1150
        if not isinstance(t1, ConnectedTransport):
1151
            raise TestSkipped("not a connected transport")
1152
1153
        t2 = t1.clone('subdir')
1154
        self.assertEquals(t1._scheme, t2._scheme)
1155
        self.assertEquals(t1._user, t2._user)
1156
        self.assertEquals(t1._password, t2._password)
1157
        self.assertEquals(t1._host, t2._host)
1158
        self.assertEquals(t1._port, t2._port)
1159
2485.8.39 by Vincent Ladeuil
Add tests around connection reuse.
1160
    def test__reuse_for(self):
1161
        t = self.get_transport()
1162
        if not isinstance(t, ConnectedTransport):
1163
            raise TestSkipped("not a connected transport")
1164
1165
        def new_url(scheme=None, user=None, password=None,
1166
                    host=None, port=None, path=None):
1167
            """Build a new url from t.base chaging only parts of it.
1168
1169
            Only the parameters different from None will be changed.
1170
            """
2485.8.53 by Vincent Ladeuil
Apply test_connection_sharing to all transports to make smart
1171
            if scheme   is None: scheme   = t._scheme
1172
            if user     is None: user     = t._user
2485.8.39 by Vincent Ladeuil
Add tests around connection reuse.
1173
            if password is None: password = t._password
2485.8.53 by Vincent Ladeuil
Apply test_connection_sharing to all transports to make smart
1174
            if user     is None: user     = t._user
1175
            if host     is None: host     = t._host
1176
            if port     is None: port     = t._port
1177
            if path     is None: path     = t._path
2485.8.39 by Vincent Ladeuil
Add tests around connection reuse.
1178
            return t._unsplit_url(scheme, user, password, host, port, path)
1179
1180
        self.assertIsNot(t, t._reuse_for(new_url(scheme='foo')))
1181
        if t._user == 'me':
1182
            user = 'you'
1183
        else:
1184
            user = 'me'
1185
        self.assertIsNot(t, t._reuse_for(new_url(user=user)))
1186
        # passwords are not taken into account because:
1187
        # - it makes no sense to have two different valid passwords for the
1188
        #   same user
1189
        # - _password in ConnectedTransport is intended to collect what the
1190
        #   user specified from the command-line and there are cases where the
1191
        #   new url can contain no password (if the url was built from an
1192
        #   existing transport.base for example)
1193
        # - password are considered part of the credentials provided at
1194
        #   connection creation time and as such may not be present in the url
1195
        #   (they may be typed by the user when prompted for example)
1196
        self.assertIs(t, t._reuse_for(new_url(password='from space')))
1197
        # We will not connect, we can use a invalid host
1198
        self.assertIsNot(t, t._reuse_for(new_url(host=t._host + 'bar')))
1199
        if t._port == 1234:
1200
            port = 4321
1201
        else:
1202
            port = 1234
1203
        self.assertIsNot(t, t._reuse_for(new_url(port=port)))
1204
2485.8.53 by Vincent Ladeuil
Apply test_connection_sharing to all transports to make smart
1205
    def test_connection_sharing(self):
1206
        t = self.get_transport()
1207
        if not isinstance(t, ConnectedTransport):
1208
            raise TestSkipped("not a connected transport")
1209
1210
        c = t.clone('subdir')
2485.8.54 by Vincent Ladeuil
Refactor medium uses by making a distinction betweem shared and real medium.
1211
        # Some transports will create the connection  only when needed
2485.8.53 by Vincent Ladeuil
Apply test_connection_sharing to all transports to make smart
1212
        t.has('surely_not') # Force connection
1213
        self.assertIs(t._get_connection(), c._get_connection())
1214
2485.8.54 by Vincent Ladeuil
Refactor medium uses by making a distinction betweem shared and real medium.
1215
        # Temporary failure, we need to create a new dummy connection
2485.8.53 by Vincent Ladeuil
Apply test_connection_sharing to all transports to make smart
1216
        new_connection = object()
1217
        t._set_connection(new_connection)
2485.8.54 by Vincent Ladeuil
Refactor medium uses by making a distinction betweem shared and real medium.
1218
        # Check that both transports use the same connection
2485.8.53 by Vincent Ladeuil
Apply test_connection_sharing to all transports to make smart
1219
        self.assertIs(new_connection, t._get_connection())
1220
        self.assertIs(new_connection, c._get_connection())
1221
2485.8.39 by Vincent Ladeuil
Add tests around connection reuse.
1222
    def test_reuse_connection_for_various_paths(self):
1223
        t = self.get_transport()
1224
        if not isinstance(t, ConnectedTransport):
1225
            raise TestSkipped("not a connected transport")
1226
1227
        t.has('surely_not') # Force connection
1228
        self.assertIsNot(None, t._get_connection())
1229
1230
        subdir = t._reuse_for(t.base + 'whatever/but/deep/down/the/path')
1231
        self.assertIsNot(t, subdir)
1232
        self.assertIs(t._get_connection(), subdir._get_connection())
1233
1234
        home = subdir._reuse_for(t.base + 'home')
1235
        self.assertIs(t._get_connection(), home._get_connection())
1236
        self.assertIs(subdir._get_connection(), home._get_connection())
1237
1530.1.3 by Robert Collins
transport implementations now tested consistently.
1238
    def test_clone(self):
1239
        # TODO: Test that clone moves up and down the filesystem
1240
        t1 = self.get_transport()
1241
1242
        self.build_tree(['a', 'b/', 'b/c'], transport=t1)
1243
1244
        self.failUnless(t1.has('a'))
1245
        self.failUnless(t1.has('b/c'))
1246
        self.failIf(t1.has('c'))
1247
1248
        t2 = t1.clone('b')
1249
        self.assertEqual(t1.base + 'b/', t2.base)
1250
1251
        self.failUnless(t2.has('c'))
1252
        self.failIf(t2.has('a'))
1253
1254
        t3 = t2.clone('..')
1255
        self.failUnless(t3.has('a'))
1256
        self.failIf(t3.has('c'))
1257
1258
        self.failIf(t1.has('b/d'))
1259
        self.failIf(t2.has('d'))
1260
        self.failIf(t3.has('b/d'))
1261
1262
        if t1.is_readonly():
1263
            open('b/d', 'wb').write('newfile\n')
1264
        else:
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
1265
            t2.put_bytes('d', 'newfile\n')
1530.1.3 by Robert Collins
transport implementations now tested consistently.
1266
1267
        self.failUnless(t1.has('b/d'))
1268
        self.failUnless(t2.has('d'))
1269
        self.failUnless(t3.has('b/d'))
1270
1910.15.1 by Andrew Bennetts
More tests for abspath and clone behaviour
1271
    def test_clone_to_root(self):
1272
        orig_transport = self.get_transport()
1273
        # Repeatedly go up to a parent directory until we're at the root
1274
        # directory of this transport
1275
        root_transport = orig_transport
1986.1.10 by Robert Collins
Merge from bzr.dev, fixing found bugs handling 'has('/')' in MemoryTransport and SFTP transports.
1276
        new_transport = root_transport.clone("..")
2245.6.1 by Alexander Belchenko
win32 UNC path: recursive cloning UNC path to root stops on //HOST, not on //
1277
        # as we are walking up directories, the path must be
1986.1.10 by Robert Collins
Merge from bzr.dev, fixing found bugs handling 'has('/')' in MemoryTransport and SFTP transports.
1278
        # growing less, except at the top
1279
        self.assertTrue(len(new_transport.base) < len(root_transport.base)
1280
            or new_transport.base == root_transport.base)
1281
        while new_transport.base != root_transport.base:
1282
            root_transport = new_transport
1283
            new_transport = root_transport.clone("..")
2245.6.1 by Alexander Belchenko
win32 UNC path: recursive cloning UNC path to root stops on //HOST, not on //
1284
            # as we are walking up directories, the path must be
1986.1.10 by Robert Collins
Merge from bzr.dev, fixing found bugs handling 'has('/')' in MemoryTransport and SFTP transports.
1285
            # growing less, except at the top
1286
            self.assertTrue(len(new_transport.base) < len(root_transport.base)
1287
                or new_transport.base == root_transport.base)
1910.15.1 by Andrew Bennetts
More tests for abspath and clone behaviour
1288
1289
        # Cloning to "/" should take us to exactly the same location.
1290
        self.assertEqual(root_transport.base, orig_transport.clone("/").base)
1986.1.10 by Robert Collins
Merge from bzr.dev, fixing found bugs handling 'has('/')' in MemoryTransport and SFTP transports.
1291
        # the abspath of "/" from the original transport should be the same
1292
        # as the base at the root:
1293
        self.assertEqual(orig_transport.abspath("/"), root_transport.base)
1910.15.1 by Andrew Bennetts
More tests for abspath and clone behaviour
1294
1910.15.5 by Andrew Bennetts
Transport behaviour at the root of the URL is now defined and tested.
1295
        # At the root, the URL must still end with / as its a directory
1296
        self.assertEqual(root_transport.base[-1], '/')
1297
1298
    def test_clone_from_root(self):
1299
        """At the root, cloning to a simple dir should just do string append."""
1300
        orig_transport = self.get_transport()
1301
        root_transport = orig_transport.clone('/')
1302
        self.assertEqual(root_transport.base + '.bzr/',
1303
            root_transport.clone('.bzr').base)
1304
1910.15.1 by Andrew Bennetts
More tests for abspath and clone behaviour
1305
    def test_base_url(self):
1306
        t = self.get_transport()
1307
        self.assertEqual('/', t.base[-1])
1308
1530.1.3 by Robert Collins
transport implementations now tested consistently.
1309
    def test_relpath(self):
1310
        t = self.get_transport()
1311
        self.assertEqual('', t.relpath(t.base))
1312
        # base ends with /
1313
        self.assertEqual('', t.relpath(t.base[:-1]))
1314
        # subdirs which dont exist should still give relpaths.
1315
        self.assertEqual('foo', t.relpath(t.base + 'foo'))
1316
        # trailing slash should be the same.
1317
        self.assertEqual('foo', t.relpath(t.base + 'foo/'))
1318
1636.1.1 by Robert Collins
Fix calling relpath() and abspath() on transports at their root.
1319
    def test_relpath_at_root(self):
1320
        t = self.get_transport()
1321
        # clone all the way to the top
1322
        new_transport = t.clone('..')
1323
        while new_transport.base != t.base:
1324
            t = new_transport
1325
            new_transport = t.clone('..')
1326
        # we must be able to get a relpath below the root
1327
        self.assertEqual('', t.relpath(t.base))
1328
        # and a deeper one should work too
1329
        self.assertEqual('foo/bar', t.relpath(t.base + 'foo/bar'))
1330
1530.1.3 by Robert Collins
transport implementations now tested consistently.
1331
    def test_abspath(self):
1332
        # smoke test for abspath. Corner cases for backends like unix fs's
1333
        # that have aliasing problems like symlinks should go in backend
1334
        # specific test cases.
1335
        transport = self.get_transport()
2485.8.21 by Vincent Ladeuil
Simplify debug.
1336
1530.1.3 by Robert Collins
transport implementations now tested consistently.
1337
        self.assertEqual(transport.base + 'relpath',
1338
                         transport.abspath('relpath'))
1339
1910.15.1 by Andrew Bennetts
More tests for abspath and clone behaviour
1340
        # This should work without raising an error.
1341
        transport.abspath("/")
1342
1343
        # the abspath of "/" and "/foo/.." should result in the same location
1344
        self.assertEqual(transport.abspath("/"), transport.abspath("/foo/.."))
1345
2070.3.1 by Andrew Bennetts
Fix memory_transport.abspath('/foo')
1346
        self.assertEqual(transport.clone("/").abspath('foo'),
1347
                         transport.abspath("/foo"))
1348
1685.1.9 by John Arbash Meinel
Updated LocalTransport so that it's base is now a URL rather than a local path. This helps consistency with all other functions. To do so, I added local_abspath() which returns the local path, and local_path_to/from_url
1349
    def test_local_abspath(self):
1350
        transport = self.get_transport()
1351
        try:
1352
            p = transport.local_abspath('.')
2018.18.4 by Martin Pool
Change Transport.local_abspath to raise NotLocalUrl, and test.
1353
        except (errors.NotLocalUrl, TransportNotPossible), e:
2018.18.22 by Martin Pool
merge bzr.dev
1354
            # should be formattable
2018.18.4 by Martin Pool
Change Transport.local_abspath to raise NotLocalUrl, and test.
1355
            s = str(e)
1685.1.9 by John Arbash Meinel
Updated LocalTransport so that it's base is now a URL rather than a local path. This helps consistency with all other functions. To do so, I added local_abspath() which returns the local path, and local_path_to/from_url
1356
        else:
1357
            self.assertEqual(getcwd(), p)
1358
1636.1.1 by Robert Collins
Fix calling relpath() and abspath() on transports at their root.
1359
    def test_abspath_at_root(self):
1360
        t = self.get_transport()
1361
        # clone all the way to the top
1362
        new_transport = t.clone('..')
1363
        while new_transport.base != t.base:
1364
            t = new_transport
1365
            new_transport = t.clone('..')
1366
        # we must be able to get a abspath of the root when we ask for
1367
        # t.abspath('..') - this due to our choice that clone('..')
1368
        # should return the root from the root, combined with the desire that
1369
        # the url from clone('..') and from abspath('..') should be the same.
1370
        self.assertEqual(t.base, t.abspath('..'))
1371
        # '' should give us the root
1372
        self.assertEqual(t.base, t.abspath(''))
1373
        # and a path should append to the url
1374
        self.assertEqual(t.base + 'foo', t.abspath('foo'))
1375
1530.1.3 by Robert Collins
transport implementations now tested consistently.
1376
    def test_iter_files_recursive(self):
1530.1.4 by Robert Collins
integrate Memory tests into transport interface tests.
1377
        transport = self.get_transport()
1378
        if not transport.listable():
1553.5.13 by Martin Pool
New Transport.rename that mustn't overwrite
1379
            self.assertRaises(TransportNotPossible,
1530.1.4 by Robert Collins
integrate Memory tests into transport interface tests.
1380
                              transport.iter_files_recursive)
1381
            return
1553.5.13 by Martin Pool
New Transport.rename that mustn't overwrite
1382
        self.build_tree(['isolated/',
1530.1.4 by Robert Collins
integrate Memory tests into transport interface tests.
1383
                         'isolated/dir/',
1384
                         'isolated/dir/foo',
1385
                         'isolated/dir/bar',
1959.2.1 by John Arbash Meinel
David Allouche: Make transports return escaped paths
1386
                         'isolated/dir/b%25z', # make sure quoting is correct
1530.1.4 by Robert Collins
integrate Memory tests into transport interface tests.
1387
                         'isolated/bar'],
1388
                        transport=transport)
1530.1.3 by Robert Collins
transport implementations now tested consistently.
1389
        paths = set(transport.iter_files_recursive())
1553.5.13 by Martin Pool
New Transport.rename that mustn't overwrite
1390
        # nb the directories are not converted
1391
        self.assertEqual(paths,
1392
                    set(['isolated/dir/foo',
1393
                         'isolated/dir/bar',
1959.2.1 by John Arbash Meinel
David Allouche: Make transports return escaped paths
1394
                         'isolated/dir/b%2525z',
1553.5.13 by Martin Pool
New Transport.rename that mustn't overwrite
1395
                         'isolated/bar']))
1396
        sub_transport = transport.clone('isolated')
1397
        paths = set(sub_transport.iter_files_recursive())
1959.2.1 by John Arbash Meinel
David Allouche: Make transports return escaped paths
1398
        self.assertEqual(paths,
1399
            set(['dir/foo', 'dir/bar', 'dir/b%2525z', 'bar']))
1400
1401
    def test_copy_tree(self):
1402
        # TODO: test file contents and permissions are preserved. This test was
1403
        # added just to ensure that quoting was handled correctly.
1404
        # -- David Allouche 2006-08-11
1405
        transport = self.get_transport()
1406
        if not transport.listable():
1407
            self.assertRaises(TransportNotPossible,
1408
                              transport.iter_files_recursive)
1409
            return
1410
        if transport.is_readonly():
1411
            return
1412
        self.build_tree(['from/',
1413
                         'from/dir/',
1414
                         'from/dir/foo',
1415
                         'from/dir/bar',
1416
                         'from/dir/b%25z', # make sure quoting is correct
1417
                         'from/bar'],
1418
                        transport=transport)
1419
        transport.copy_tree('from', 'to')
1420
        paths = set(transport.iter_files_recursive())
1421
        self.assertEqual(paths,
1422
                    set(['from/dir/foo',
1423
                         'from/dir/bar',
1424
                         'from/dir/b%2525z',
1425
                         'from/bar',
1426
                         'to/dir/foo',
1427
                         'to/dir/bar',
1428
                         'to/dir/b%2525z',
1429
                         'to/bar',]))
1185.85.76 by John Arbash Meinel
Adding an InvalidURL so transports can report they expect utf-8 quoted paths. Updated tests
1430
1431
    def test_unicode_paths(self):
1685.1.57 by Martin Pool
[broken] Skip unicode blackbox tests if not supported by filesystem
1432
        """Test that we can read/write files with Unicode names."""
1185.85.76 by John Arbash Meinel
Adding an InvalidURL so transports can report they expect utf-8 quoted paths. Updated tests
1433
        t = self.get_transport()
1434
1711.7.36 by John Arbash Meinel
Use different filenames to avoid path collisions on win32 w/ FAT32
1435
        # With FAT32 and certain encodings on win32
1436
        # '\xe5' and '\xe4' actually map to the same file
1437
        # adding a suffix kicks in the 'preserving but insensitive'
1438
        # route, and maintains the right files
1439
        files = [u'\xe5.1', # a w/ circle iso-8859-1
1440
                 u'\xe4.2', # a w/ dots iso-8859-1
1185.85.76 by John Arbash Meinel
Adding an InvalidURL so transports can report they expect utf-8 quoted paths. Updated tests
1441
                 u'\u017d', # Z with umlat iso-8859-2
1442
                 u'\u062c', # Arabic j
1443
                 u'\u0410', # Russian A
1444
                 u'\u65e5', # Kanji person
1445
                ]
1446
1685.1.72 by Wouter van Heyst
StubSFTPServer should use bytestreams rather than unicode
1447
        try:
1711.4.13 by John Arbash Meinel
Use line_endings='binary' for win32
1448
            self.build_tree(files, transport=t, line_endings='binary')
1685.1.72 by Wouter van Heyst
StubSFTPServer should use bytestreams rather than unicode
1449
        except UnicodeError:
1450
            raise TestSkipped("cannot handle unicode paths in current encoding")
1185.85.76 by John Arbash Meinel
Adding an InvalidURL so transports can report they expect utf-8 quoted paths. Updated tests
1451
1452
        # A plain unicode string is not a valid url
1453
        for fname in files:
1454
            self.assertRaises(InvalidURL, t.get, fname)
1455
1456
        for fname in files:
1457
            fname_utf8 = fname.encode('utf-8')
1458
            contents = 'contents of %s\n' % (fname_utf8,)
1685.1.45 by John Arbash Meinel
Moved url functions into bzrlib.urlutils
1459
            self.check_transport_contents(contents, t, urlutils.escape(fname))
1185.85.76 by John Arbash Meinel
Adding an InvalidURL so transports can report they expect utf-8 quoted paths. Updated tests
1460
1534.4.26 by Robert Collins
Move working tree initialisation out from Branch.initialize, deprecated Branch.initialize to Branch.create.
1461
    def test_connect_twice_is_same_content(self):
2485.8.21 by Vincent Ladeuil
Simplify debug.
1462
        # check that our server (whatever it is) is accessible reliably
1534.4.26 by Robert Collins
Move working tree initialisation out from Branch.initialize, deprecated Branch.initialize to Branch.create.
1463
        # via get_transport and multiple connections share content.
1464
        transport = self.get_transport()
1465
        if transport.is_readonly():
1466
            return
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
1467
        transport.put_bytes('foo', 'bar')
2485.8.21 by Vincent Ladeuil
Simplify debug.
1468
        transport3 = self.get_transport()
1469
        self.check_transport_contents('bar', transport3, 'foo')
1534.4.26 by Robert Collins
Move working tree initialisation out from Branch.initialize, deprecated Branch.initialize to Branch.create.
1470
        # its base should be usable.
2485.8.25 by Vincent Ladeuil
Separate abspath from _remote_path, the intents are different.
1471
        transport4 = get_transport(transport.base)
2485.8.21 by Vincent Ladeuil
Simplify debug.
1472
        self.check_transport_contents('bar', transport4, 'foo')
1534.4.26 by Robert Collins
Move working tree initialisation out from Branch.initialize, deprecated Branch.initialize to Branch.create.
1473
1474
        # now opening at a relative url should give use a sane result:
1475
        transport.mkdir('newdir')
2485.8.25 by Vincent Ladeuil
Separate abspath from _remote_path, the intents are different.
1476
        transport5 = get_transport(transport.base + "newdir")
2485.8.21 by Vincent Ladeuil
Simplify debug.
1477
        transport6 = transport5.clone('..')
1478
        self.check_transport_contents('bar', transport6, 'foo')
1534.4.26 by Robert Collins
Move working tree initialisation out from Branch.initialize, deprecated Branch.initialize to Branch.create.
1479
1480
    def test_lock_write(self):
1910.16.1 by Andrew Bennetts
lock_read and lock_write may raise TransportNotPossible.
1481
        """Test transport-level write locks.
1482
1483
        These are deprecated and transports may decline to support them.
1484
        """
1534.4.26 by Robert Collins
Move working tree initialisation out from Branch.initialize, deprecated Branch.initialize to Branch.create.
1485
        transport = self.get_transport()
1486
        if transport.is_readonly():
1487
            self.assertRaises(TransportNotPossible, transport.lock_write, 'foo')
1488
            return
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
1489
        transport.put_bytes('lock', '')
1910.16.1 by Andrew Bennetts
lock_read and lock_write may raise TransportNotPossible.
1490
        try:
1491
            lock = transport.lock_write('lock')
1752.3.6 by Andrew Bennetts
Merge from test-tweaks
1492
        except TransportNotPossible:
1910.16.1 by Andrew Bennetts
lock_read and lock_write may raise TransportNotPossible.
1493
            return
1534.4.26 by Robert Collins
Move working tree initialisation out from Branch.initialize, deprecated Branch.initialize to Branch.create.
1494
        # TODO make this consistent on all platforms:
1495
        # self.assertRaises(LockError, transport.lock_write, 'lock')
1496
        lock.unlock()
1497
1498
    def test_lock_read(self):
1910.16.1 by Andrew Bennetts
lock_read and lock_write may raise TransportNotPossible.
1499
        """Test transport-level read locks.
1500
1501
        These are deprecated and transports may decline to support them.
1502
        """
1534.4.26 by Robert Collins
Move working tree initialisation out from Branch.initialize, deprecated Branch.initialize to Branch.create.
1503
        transport = self.get_transport()
1504
        if transport.is_readonly():
1505
            file('lock', 'w').close()
1506
        else:
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
1507
            transport.put_bytes('lock', '')
1910.16.1 by Andrew Bennetts
lock_read and lock_write may raise TransportNotPossible.
1508
        try:
1509
            lock = transport.lock_read('lock')
1752.3.6 by Andrew Bennetts
Merge from test-tweaks
1510
        except TransportNotPossible:
1910.16.1 by Andrew Bennetts
lock_read and lock_write may raise TransportNotPossible.
1511
            return
1534.4.26 by Robert Collins
Move working tree initialisation out from Branch.initialize, deprecated Branch.initialize to Branch.create.
1512
        # TODO make this consistent on all platforms:
1513
        # self.assertRaises(LockError, transport.lock_read, 'lock')
1514
        lock.unlock()
1185.85.80 by John Arbash Meinel
[merge] jam-integration 1527, including branch-formats, help text, misc bug fixes.
1515
1594.2.5 by Robert Collins
Readv patch from Johan Rydberg giving knits partial download support.
1516
    def test_readv(self):
1517
        transport = self.get_transport()
1518
        if transport.is_readonly():
1519
            file('a', 'w').write('0123456789')
1520
        else:
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
1521
            transport.put_bytes('a', '0123456789')
1185.85.76 by John Arbash Meinel
Adding an InvalidURL so transports can report they expect utf-8 quoted paths. Updated tests
1522
2004.1.22 by v.ladeuil+lp at free
Implements Range header handling for GET requests. Fix a test.
1523
        d = list(transport.readv('a', ((0, 1),)))
1524
        self.assertEqual(d[0], (0, '0'))
1525
1594.2.17 by Robert Collins
Better readv coalescing, now with test, and progress during knit index reading.
1526
        d = list(transport.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
1594.2.5 by Robert Collins
Readv patch from Johan Rydberg giving knits partial download support.
1527
        self.assertEqual(d[0], (0, '0'))
1594.2.17 by Robert Collins
Better readv coalescing, now with test, and progress during knit index reading.
1528
        self.assertEqual(d[1], (1, '1'))
1529
        self.assertEqual(d[2], (3, '34'))
1530
        self.assertEqual(d[3], (9, '9'))
1786.1.8 by John Arbash Meinel
[merge] Johan Rydberg test updates
1531
1864.5.2 by John Arbash Meinel
always read in sorted order, and return in requested order, but only cache what is currently out of order
1532
    def test_readv_out_of_order(self):
1533
        transport = self.get_transport()
1534
        if transport.is_readonly():
1535
            file('a', 'w').write('0123456789')
1536
        else:
1955.3.11 by John Arbash Meinel
Clean up the rest of the api calls to deprecated functions in the test suite, and make sure Transport._pump is only accepting files, not strings
1537
            transport.put_bytes('a', '01234567890')
1864.5.2 by John Arbash Meinel
always read in sorted order, and return in requested order, but only cache what is currently out of order
1538
1539
        d = list(transport.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
1540
        self.assertEqual(d[0], (1, '1'))
1541
        self.assertEqual(d[1], (9, '9'))
1542
        self.assertEqual(d[2], (0, '0'))
1543
        self.assertEqual(d[3], (3, '34'))
1752.2.40 by Andrew Bennetts
Merge from bzr.dev.
1544
2018.2.3 by Andrew Bennetts
Starting factoring out the smart server client "medium" from the protocol.
1545
    def test_get_smart_medium(self):
1546
        """All transports must either give a smart medium, or know they can't.
1547
        """
1548
        transport = self.get_transport()
1549
        try:
2018.5.2 by Andrew Bennetts
Start splitting bzrlib/transport/smart.py into a package.
1550
            client_medium = transport.get_smart_medium()
1551
            self.assertIsInstance(client_medium, medium.SmartClientMedium)
2018.2.3 by Andrew Bennetts
Starting factoring out the smart server client "medium" from the protocol.
1552
        except errors.NoSmartMedium:
1553
            # as long as we got it we're fine
1554
            pass
1555
2001.3.2 by John Arbash Meinel
Force all transports to raise ShortReadvError if they can
1556
    def test_readv_short_read(self):
1557
        transport = self.get_transport()
1558
        if transport.is_readonly():
1559
            file('a', 'w').write('0123456789')
1560
        else:
1561
            transport.put_bytes('a', '01234567890')
1562
1563
        # This is intentionally reading off the end of the file
1564
        # since we are sure that it cannot get there
2000.3.9 by v.ladeuil+lp at free
The tests that would have help avoid bug #73948 and all that mess :)
1565
        self.assertListRaises((errors.ShortReadvError, errors.InvalidRange,
1566
                               # Can be raised by paramiko
1567
                               AssertionError),
2001.3.2 by John Arbash Meinel
Force all transports to raise ShortReadvError if they can
1568
                              transport.readv, 'a', [(1,1), (8,10)])
1569
1570
        # This is trying to seek past the end of the file, it should
1571
        # also raise a special error
2000.3.9 by v.ladeuil+lp at free
The tests that would have help avoid bug #73948 and all that mess :)
1572
        self.assertListRaises((errors.ShortReadvError, errors.InvalidRange),
2001.3.2 by John Arbash Meinel
Force all transports to raise ShortReadvError if they can
1573
                              transport.readv, 'a', [(12,2)])