← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

  • Committer: Frank Aspell
  • Date: 2009-02-17 11:40:05 UTC
  • mto: (4054.1.1 doc)
  • mto: This revision was merged to the branch mainline in revision 4056.
  • Revision ID: frankaspell@googlemail.com-20090217114005-ojufrp6rqht664um
Fixed typos.

Fixed some typos in bzr doc's using "aspell -l en -c FILENAME".

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
1
 
# Copyright (C) 2005-2010 Canonical Ltd
 
1
# Copyright (C) 2004, 2005, 2006, 2007 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
 
18
18
from cStringIO import StringIO
19
 
import os
20
 
import subprocess
21
 
import sys
22
 
import threading
23
19
 
 
20
import bzrlib
24
21
from bzrlib import (
25
22
    errors,
26
23
    osutils,
27
 
    tests,
28
 
    transport,
29
24
    urlutils,
30
25
    )
31
 
from bzrlib.transport import (
32
 
    chroot,
33
 
    fakenfs,
34
 
    local,
35
 
    memory,
36
 
    pathfilter,
37
 
    readonly,
38
 
    )
39
 
from bzrlib.tests import (
40
 
    features,
41
 
    test_server,
42
 
    )
 
26
from bzrlib.errors import (DependencyNotPresent,
 
27
                           FileExists,
 
28
                           InvalidURLJoin,
 
29
                           NoSuchFile,
 
30
                           PathNotChild,
 
31
                           ReadError,
 
32
                           UnsupportedProtocol,
 
33
                           )
 
34
from bzrlib.tests import TestCase, TestCaseInTempDir
 
35
from bzrlib.transport import (_clear_protocol_handlers,
 
36
                              _CoalescedOffset,
 
37
                              ConnectedTransport,
 
38
                              _get_protocol_handlers,
 
39
                              _set_protocol_handlers,
 
40
                              _get_transport_modules,
 
41
                              get_transport,
 
42
                              LateReadError,
 
43
                              register_lazy_transport,
 
44
                              register_transport_proto,
 
45
                              Transport,
 
46
                              )
 
47
from bzrlib.transport.chroot import ChrootServer
 
48
from bzrlib.transport.memory import MemoryTransport
 
49
from bzrlib.transport.local import (LocalTransport,
 
50
                                    EmulatedWin32LocalTransport)
43
51
 
44
52
 
45
53
# TODO: Should possibly split transport-specific tests into their own files.
46
54
 
47
55
 
48
 
class TestTransport(tests.TestCase):
 
56
class TestTransport(TestCase):
49
57
    """Test the non transport-concrete class functionality."""
50
58
 
51
 
    # FIXME: These tests should use addCleanup() and/or overrideAttr() instead
52
 
    # of try/finally -- vila 20100205
53
 
 
54
59
    def test__get_set_protocol_handlers(self):
55
 
        handlers = transport._get_protocol_handlers()
 
60
        handlers = _get_protocol_handlers()
56
61
        self.assertNotEqual([], handlers.keys( ))
57
62
        try:
58
 
            transport._clear_protocol_handlers()
59
 
            self.assertEqual([], transport._get_protocol_handlers().keys())
 
63
            _clear_protocol_handlers()
 
64
            self.assertEqual([], _get_protocol_handlers().keys())
60
65
        finally:
61
 
            transport._set_protocol_handlers(handlers)
 
66
            _set_protocol_handlers(handlers)
62
67
 
63
68
    def test_get_transport_modules(self):
64
 
        handlers = transport._get_protocol_handlers()
 
69
        handlers = _get_protocol_handlers()
65
70
        # don't pollute the current handlers
66
 
        transport._clear_protocol_handlers()
 
71
        _clear_protocol_handlers()
67
72
        class SampleHandler(object):
68
73
            """I exist, isnt that enough?"""
69
74
        try:
70
 
            transport._clear_protocol_handlers()
71
 
            transport.register_transport_proto('foo')
72
 
            transport.register_lazy_transport('foo',
73
 
                                              'bzrlib.tests.test_transport',
74
 
                                              'TestTransport.SampleHandler')
75
 
            transport.register_transport_proto('bar')
76
 
            transport.register_lazy_transport('bar',
77
 
                                              'bzrlib.tests.test_transport',
78
 
                                              'TestTransport.SampleHandler')
 
75
            _clear_protocol_handlers()
 
76
            register_transport_proto('foo')
 
77
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
 
78
                                    'TestTransport.SampleHandler')
 
79
            register_transport_proto('bar')
 
80
            register_lazy_transport('bar', 'bzrlib.tests.test_transport',
 
81
                                    'TestTransport.SampleHandler')
79
82
            self.assertEqual([SampleHandler.__module__,
80
 
                              'bzrlib.transport.chroot',
81
 
                              'bzrlib.transport.pathfilter'],
82
 
                             transport._get_transport_modules())
 
83
                              'bzrlib.transport.chroot'],
 
84
                             _get_transport_modules())
83
85
        finally:
84
 
            transport._set_protocol_handlers(handlers)
 
86
            _set_protocol_handlers(handlers)
85
87
 
86
88
    def test_transport_dependency(self):
87
89
        """Transport with missing dependency causes no error"""
88
 
        saved_handlers = transport._get_protocol_handlers()
 
90
        saved_handlers = _get_protocol_handlers()
89
91
        # don't pollute the current handlers
90
 
        transport._clear_protocol_handlers()
 
92
        _clear_protocol_handlers()
91
93
        try:
92
 
            transport.register_transport_proto('foo')
93
 
            transport.register_lazy_transport(
94
 
                'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
 
94
            register_transport_proto('foo')
 
95
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
 
96
                    'BadTransportHandler')
95
97
            try:
96
 
                transport.get_transport('foo://fooserver/foo')
97
 
            except errors.UnsupportedProtocol, e:
 
98
                get_transport('foo://fooserver/foo')
 
99
            except UnsupportedProtocol, e:
98
100
                e_str = str(e)
99
101
                self.assertEquals('Unsupported protocol'
100
102
                                  ' for url "foo://fooserver/foo":'
104
106
                self.fail('Did not raise UnsupportedProtocol')
105
107
        finally:
106
108
            # restore original values
107
 
            transport._set_protocol_handlers(saved_handlers)
108
 
 
 
109
            _set_protocol_handlers(saved_handlers)
 
110
            
109
111
    def test_transport_fallback(self):
110
112
        """Transport with missing dependency causes no error"""
111
 
        saved_handlers = transport._get_protocol_handlers()
 
113
        saved_handlers = _get_protocol_handlers()
112
114
        try:
113
 
            transport._clear_protocol_handlers()
114
 
            transport.register_transport_proto('foo')
115
 
            transport.register_lazy_transport(
116
 
                'foo', 'bzrlib.tests.test_transport', 'BackupTransportHandler')
117
 
            transport.register_lazy_transport(
118
 
                'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
119
 
            t = transport.get_transport('foo://fooserver/foo')
 
115
            _clear_protocol_handlers()
 
116
            register_transport_proto('foo')
 
117
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
 
118
                    'BackupTransportHandler')
 
119
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
 
120
                    'BadTransportHandler')
 
121
            t = get_transport('foo://fooserver/foo')
120
122
            self.assertTrue(isinstance(t, BackupTransportHandler))
121
123
        finally:
122
 
            transport._set_protocol_handlers(saved_handlers)
123
 
 
124
 
    def test_ssh_hints(self):
125
 
        """Transport ssh:// should raise an error pointing out bzr+ssh://"""
126
 
        try:
127
 
            transport.get_transport('ssh://fooserver/foo')
128
 
        except errors.UnsupportedProtocol, e:
129
 
            e_str = str(e)
130
 
            self.assertEquals('Unsupported protocol'
131
 
                              ' for url "ssh://fooserver/foo":'
132
 
                              ' bzr supports bzr+ssh to operate over ssh,'
133
 
                              ' use "bzr+ssh://fooserver/foo".',
134
 
                              str(e))
135
 
        else:
136
 
            self.fail('Did not raise UnsupportedProtocol')
 
124
            _set_protocol_handlers(saved_handlers)
137
125
 
138
126
    def test_LateReadError(self):
139
127
        """The LateReadError helper should raise on read()."""
140
 
        a_file = transport.LateReadError('a path')
 
128
        a_file = LateReadError('a path')
141
129
        try:
142
130
            a_file.read()
143
 
        except errors.ReadError, error:
 
131
        except ReadError, error:
144
132
            self.assertEqual('a path', error.path)
145
 
        self.assertRaises(errors.ReadError, a_file.read, 40)
 
133
        self.assertRaises(ReadError, a_file.read, 40)
146
134
        a_file.close()
147
135
 
148
136
    def test__combine_paths(self):
149
 
        t = transport.Transport('/')
 
137
        t = Transport('/')
150
138
        self.assertEqual('/home/sarah/project/foo',
151
139
                         t._combine_paths('/home/sarah', 'project/foo'))
152
140
        self.assertEqual('/etc',
158
146
 
159
147
    def test_local_abspath_non_local_transport(self):
160
148
        # the base implementation should throw
161
 
        t = memory.MemoryTransport()
 
149
        t = MemoryTransport()
162
150
        e = self.assertRaises(errors.NotLocalUrl, t.local_abspath, 't')
163
151
        self.assertEqual('memory:///t is not a local path.', str(e))
164
152
 
165
153
 
166
 
class TestCoalesceOffsets(tests.TestCase):
 
154
class TestCoalesceOffsets(TestCase):
167
155
 
168
156
    def check(self, expected, offsets, limit=0, max_size=0, fudge=0):
169
 
        coalesce = transport.Transport._coalesce_offsets
170
 
        exp = [transport._CoalescedOffset(*x) for x in expected]
 
157
        coalesce = Transport._coalesce_offsets
 
158
        exp = [_CoalescedOffset(*x) for x in expected]
171
159
        out = list(coalesce(offsets, limit=limit, fudge_factor=fudge,
172
160
                            max_size=max_size))
173
161
        self.assertEqual(exp, out)
248
236
                   max_size=1*1024*1024*1024)
249
237
 
250
238
 
251
 
class TestMemoryServer(tests.TestCase):
252
 
 
253
 
    def test_create_server(self):
254
 
        server = memory.MemoryServer()
255
 
        server.start_server()
256
 
        url = server.get_url()
257
 
        self.assertTrue(url in transport.transport_list_registry)
258
 
        t = transport.get_transport(url)
259
 
        del t
260
 
        server.stop_server()
261
 
        self.assertFalse(url in transport.transport_list_registry)
262
 
        self.assertRaises(errors.UnsupportedProtocol,
263
 
                          transport.get_transport, url)
264
 
 
265
 
 
266
 
class TestMemoryTransport(tests.TestCase):
 
239
class TestMemoryTransport(TestCase):
267
240
 
268
241
    def test_get_transport(self):
269
 
        memory.MemoryTransport()
 
242
        MemoryTransport()
270
243
 
271
244
    def test_clone(self):
272
 
        t = memory.MemoryTransport()
273
 
        self.assertTrue(isinstance(t, memory.MemoryTransport))
274
 
        self.assertEqual("memory:///", t.clone("/").base)
 
245
        transport = MemoryTransport()
 
246
        self.assertTrue(isinstance(transport, MemoryTransport))
 
247
        self.assertEqual("memory:///", transport.clone("/").base)
275
248
 
276
249
    def test_abspath(self):
277
 
        t = memory.MemoryTransport()
278
 
        self.assertEqual("memory:///relpath", t.abspath('relpath'))
 
250
        transport = MemoryTransport()
 
251
        self.assertEqual("memory:///relpath", transport.abspath('relpath'))
279
252
 
280
253
    def test_abspath_of_root(self):
281
 
        t = memory.MemoryTransport()
282
 
        self.assertEqual("memory:///", t.base)
283
 
        self.assertEqual("memory:///", t.abspath('/'))
 
254
        transport = MemoryTransport()
 
255
        self.assertEqual("memory:///", transport.base)
 
256
        self.assertEqual("memory:///", transport.abspath('/'))
284
257
 
285
258
    def test_abspath_of_relpath_starting_at_root(self):
286
 
        t = memory.MemoryTransport()
287
 
        self.assertEqual("memory:///foo", t.abspath('/foo'))
 
259
        transport = MemoryTransport()
 
260
        self.assertEqual("memory:///foo", transport.abspath('/foo'))
288
261
 
289
262
    def test_append_and_get(self):
290
 
        t = memory.MemoryTransport()
291
 
        t.append_bytes('path', 'content')
292
 
        self.assertEqual(t.get('path').read(), 'content')
293
 
        t.append_file('path', StringIO('content'))
294
 
        self.assertEqual(t.get('path').read(), 'contentcontent')
 
263
        transport = MemoryTransport()
 
264
        transport.append_bytes('path', 'content')
 
265
        self.assertEqual(transport.get('path').read(), 'content')
 
266
        transport.append_file('path', StringIO('content'))
 
267
        self.assertEqual(transport.get('path').read(), 'contentcontent')
295
268
 
296
269
    def test_put_and_get(self):
297
 
        t = memory.MemoryTransport()
298
 
        t.put_file('path', StringIO('content'))
299
 
        self.assertEqual(t.get('path').read(), 'content')
300
 
        t.put_bytes('path', 'content')
301
 
        self.assertEqual(t.get('path').read(), 'content')
 
270
        transport = MemoryTransport()
 
271
        transport.put_file('path', StringIO('content'))
 
272
        self.assertEqual(transport.get('path').read(), 'content')
 
273
        transport.put_bytes('path', 'content')
 
274
        self.assertEqual(transport.get('path').read(), 'content')
302
275
 
303
276
    def test_append_without_dir_fails(self):
304
 
        t = memory.MemoryTransport()
305
 
        self.assertRaises(errors.NoSuchFile,
306
 
                          t.append_bytes, 'dir/path', 'content')
 
277
        transport = MemoryTransport()
 
278
        self.assertRaises(NoSuchFile,
 
279
                          transport.append_bytes, 'dir/path', 'content')
307
280
 
308
281
    def test_put_without_dir_fails(self):
309
 
        t = memory.MemoryTransport()
310
 
        self.assertRaises(errors.NoSuchFile,
311
 
                          t.put_file, 'dir/path', StringIO('content'))
 
282
        transport = MemoryTransport()
 
283
        self.assertRaises(NoSuchFile,
 
284
                          transport.put_file, 'dir/path', StringIO('content'))
312
285
 
313
286
    def test_get_missing(self):
314
 
        transport = memory.MemoryTransport()
315
 
        self.assertRaises(errors.NoSuchFile, transport.get, 'foo')
 
287
        transport = MemoryTransport()
 
288
        self.assertRaises(NoSuchFile, transport.get, 'foo')
316
289
 
317
290
    def test_has_missing(self):
318
 
        t = memory.MemoryTransport()
319
 
        self.assertEquals(False, t.has('foo'))
 
291
        transport = MemoryTransport()
 
292
        self.assertEquals(False, transport.has('foo'))
320
293
 
321
294
    def test_has_present(self):
322
 
        t = memory.MemoryTransport()
323
 
        t.append_bytes('foo', 'content')
324
 
        self.assertEquals(True, t.has('foo'))
 
295
        transport = MemoryTransport()
 
296
        transport.append_bytes('foo', 'content')
 
297
        self.assertEquals(True, transport.has('foo'))
325
298
 
326
299
    def test_list_dir(self):
327
 
        t = memory.MemoryTransport()
328
 
        t.put_bytes('foo', 'content')
329
 
        t.mkdir('dir')
330
 
        t.put_bytes('dir/subfoo', 'content')
331
 
        t.put_bytes('dirlike', 'content')
 
300
        transport = MemoryTransport()
 
301
        transport.put_bytes('foo', 'content')
 
302
        transport.mkdir('dir')
 
303
        transport.put_bytes('dir/subfoo', 'content')
 
304
        transport.put_bytes('dirlike', 'content')
332
305
 
333
 
        self.assertEquals(['dir', 'dirlike', 'foo'], sorted(t.list_dir('.')))
334
 
        self.assertEquals(['subfoo'], sorted(t.list_dir('dir')))
 
306
        self.assertEquals(['dir', 'dirlike', 'foo'], sorted(transport.list_dir('.')))
 
307
        self.assertEquals(['subfoo'], sorted(transport.list_dir('dir')))
335
308
 
336
309
    def test_mkdir(self):
337
 
        t = memory.MemoryTransport()
338
 
        t.mkdir('dir')
339
 
        t.append_bytes('dir/path', 'content')
340
 
        self.assertEqual(t.get('dir/path').read(), 'content')
 
310
        transport = MemoryTransport()
 
311
        transport.mkdir('dir')
 
312
        transport.append_bytes('dir/path', 'content')
 
313
        self.assertEqual(transport.get('dir/path').read(), 'content')
341
314
 
342
315
    def test_mkdir_missing_parent(self):
343
 
        t = memory.MemoryTransport()
344
 
        self.assertRaises(errors.NoSuchFile, t.mkdir, 'dir/dir')
 
316
        transport = MemoryTransport()
 
317
        self.assertRaises(NoSuchFile,
 
318
                          transport.mkdir, 'dir/dir')
345
319
 
346
320
    def test_mkdir_twice(self):
347
 
        t = memory.MemoryTransport()
348
 
        t.mkdir('dir')
349
 
        self.assertRaises(errors.FileExists, t.mkdir, 'dir')
 
321
        transport = MemoryTransport()
 
322
        transport.mkdir('dir')
 
323
        self.assertRaises(FileExists, transport.mkdir, 'dir')
350
324
 
351
325
    def test_parameters(self):
352
 
        t = memory.MemoryTransport()
353
 
        self.assertEqual(True, t.listable())
354
 
        self.assertEqual(False, t.is_readonly())
 
326
        transport = MemoryTransport()
 
327
        self.assertEqual(True, transport.listable())
 
328
        self.assertEqual(False, transport.is_readonly())
355
329
 
356
330
    def test_iter_files_recursive(self):
357
 
        t = memory.MemoryTransport()
358
 
        t.mkdir('dir')
359
 
        t.put_bytes('dir/foo', 'content')
360
 
        t.put_bytes('dir/bar', 'content')
361
 
        t.put_bytes('bar', 'content')
362
 
        paths = set(t.iter_files_recursive())
 
331
        transport = MemoryTransport()
 
332
        transport.mkdir('dir')
 
333
        transport.put_bytes('dir/foo', 'content')
 
334
        transport.put_bytes('dir/bar', 'content')
 
335
        transport.put_bytes('bar', 'content')
 
336
        paths = set(transport.iter_files_recursive())
363
337
        self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
364
338
 
365
339
    def test_stat(self):
366
 
        t = memory.MemoryTransport()
367
 
        t.put_bytes('foo', 'content')
368
 
        t.put_bytes('bar', 'phowar')
369
 
        self.assertEqual(7, t.stat('foo').st_size)
370
 
        self.assertEqual(6, t.stat('bar').st_size)
371
 
 
372
 
 
373
 
class ChrootDecoratorTransportTest(tests.TestCase):
 
340
        transport = MemoryTransport()
 
341
        transport.put_bytes('foo', 'content')
 
342
        transport.put_bytes('bar', 'phowar')
 
343
        self.assertEqual(7, transport.stat('foo').st_size)
 
344
        self.assertEqual(6, transport.stat('bar').st_size)
 
345
 
 
346
 
 
347
class ChrootDecoratorTransportTest(TestCase):
374
348
    """Chroot decoration specific tests."""
375
349
 
376
350
    def test_abspath(self):
377
351
        # The abspath is always relative to the chroot_url.
378
 
        server = chroot.ChrootServer(
379
 
            transport.get_transport('memory:///foo/bar/'))
380
 
        self.start_server(server)
381
 
        t = transport.get_transport(server.get_url())
382
 
        self.assertEqual(server.get_url(), t.abspath('/'))
 
352
        server = ChrootServer(get_transport('memory:///foo/bar/'))
 
353
        server.setUp()
 
354
        transport = get_transport(server.get_url())
 
355
        self.assertEqual(server.get_url(), transport.abspath('/'))
383
356
 
384
 
        subdir_t = t.clone('subdir')
385
 
        self.assertEqual(server.get_url(), subdir_t.abspath('/'))
 
357
        subdir_transport = transport.clone('subdir')
 
358
        self.assertEqual(server.get_url(), subdir_transport.abspath('/'))
 
359
        server.tearDown()
386
360
 
387
361
    def test_clone(self):
388
 
        server = chroot.ChrootServer(
389
 
            transport.get_transport('memory:///foo/bar/'))
390
 
        self.start_server(server)
391
 
        t = transport.get_transport(server.get_url())
 
362
        server = ChrootServer(get_transport('memory:///foo/bar/'))
 
363
        server.setUp()
 
364
        transport = get_transport(server.get_url())
392
365
        # relpath from root and root path are the same
393
 
        relpath_cloned = t.clone('foo')
394
 
        abspath_cloned = t.clone('/foo')
 
366
        relpath_cloned = transport.clone('foo')
 
367
        abspath_cloned = transport.clone('/foo')
395
368
        self.assertEqual(server, relpath_cloned.server)
396
369
        self.assertEqual(server, abspath_cloned.server)
397
 
 
 
370
        server.tearDown()
 
371
    
398
372
    def test_chroot_url_preserves_chroot(self):
399
373
        """Calling get_transport on a chroot transport's base should produce a
400
374
        transport with exactly the same behaviour as the original chroot
403
377
        This is so that it is not possible to escape a chroot by doing::
404
378
            url = chroot_transport.base
405
379
            parent_url = urlutils.join(url, '..')
406
 
            new_t = transport.get_transport(parent_url)
 
380
            new_transport = get_transport(parent_url)
407
381
        """
408
 
        server = chroot.ChrootServer(
409
 
            transport.get_transport('memory:///path/subpath'))
410
 
        self.start_server(server)
411
 
        t = transport.get_transport(server.get_url())
412
 
        new_t = transport.get_transport(t.base)
413
 
        self.assertEqual(t.server, new_t.server)
414
 
        self.assertEqual(t.base, new_t.base)
415
 
 
 
382
        server = ChrootServer(get_transport('memory:///path/subpath'))
 
383
        server.setUp()
 
384
        transport = get_transport(server.get_url())
 
385
        new_transport = get_transport(transport.base)
 
386
        self.assertEqual(transport.server, new_transport.server)
 
387
        self.assertEqual(transport.base, new_transport.base)
 
388
        server.tearDown()
 
389
        
416
390
    def test_urljoin_preserves_chroot(self):
417
391
        """Using urlutils.join(url, '..') on a chroot URL should not produce a
418
392
        URL that escapes the intended chroot.
420
394
        This is so that it is not possible to escape a chroot by doing::
421
395
            url = chroot_transport.base
422
396
            parent_url = urlutils.join(url, '..')
423
 
            new_t = transport.get_transport(parent_url)
 
397
            new_transport = get_transport(parent_url)
424
398
        """
425
 
        server = chroot.ChrootServer(transport.get_transport('memory:///path/'))
426
 
        self.start_server(server)
427
 
        t = transport.get_transport(server.get_url())
 
399
        server = ChrootServer(get_transport('memory:///path/'))
 
400
        server.setUp()
 
401
        transport = get_transport(server.get_url())
428
402
        self.assertRaises(
429
 
            errors.InvalidURLJoin, urlutils.join, t.base, '..')
430
 
 
431
 
 
432
 
class TestChrootServer(tests.TestCase):
 
403
            InvalidURLJoin, urlutils.join, transport.base, '..')
 
404
        server.tearDown()
 
405
 
 
406
 
 
407
class ChrootServerTest(TestCase):
433
408
 
434
409
    def test_construct(self):
435
 
        backing_transport = memory.MemoryTransport()
436
 
        server = chroot.ChrootServer(backing_transport)
 
410
        backing_transport = MemoryTransport()
 
411
        server = ChrootServer(backing_transport)
437
412
        self.assertEqual(backing_transport, server.backing_transport)
438
413
 
439
414
    def test_setUp(self):
440
 
        backing_transport = memory.MemoryTransport()
441
 
        server = chroot.ChrootServer(backing_transport)
442
 
        server.start_server()
443
 
        try:
444
 
            self.assertTrue(server.scheme
445
 
                            in transport._get_protocol_handlers().keys())
446
 
        finally:
447
 
            server.stop_server()
 
415
        backing_transport = MemoryTransport()
 
416
        server = ChrootServer(backing_transport)
 
417
        server.setUp()
 
418
        self.assertTrue(server.scheme in _get_protocol_handlers().keys())
448
419
 
449
 
    def test_stop_server(self):
450
 
        backing_transport = memory.MemoryTransport()
451
 
        server = chroot.ChrootServer(backing_transport)
452
 
        server.start_server()
453
 
        server.stop_server()
454
 
        self.assertFalse(server.scheme
455
 
                         in transport._get_protocol_handlers().keys())
 
420
    def test_tearDown(self):
 
421
        backing_transport = MemoryTransport()
 
422
        server = ChrootServer(backing_transport)
 
423
        server.setUp()
 
424
        server.tearDown()
 
425
        self.assertFalse(server.scheme in _get_protocol_handlers().keys())
456
426
 
457
427
    def test_get_url(self):
458
 
        backing_transport = memory.MemoryTransport()
459
 
        server = chroot.ChrootServer(backing_transport)
460
 
        server.start_server()
461
 
        try:
462
 
            self.assertEqual('chroot-%d:///' % id(server), server.get_url())
463
 
        finally:
464
 
            server.stop_server()
465
 
 
466
 
 
467
 
class PathFilteringDecoratorTransportTest(tests.TestCase):
468
 
    """Pathfilter decoration specific tests."""
469
 
 
470
 
    def test_abspath(self):
471
 
        # The abspath is always relative to the base of the backing transport.
472
 
        server = pathfilter.PathFilteringServer(
473
 
            transport.get_transport('memory:///foo/bar/'),
474
 
            lambda x: x)
475
 
        server.start_server()
476
 
        t = transport.get_transport(server.get_url())
477
 
        self.assertEqual(server.get_url(), t.abspath('/'))
478
 
 
479
 
        subdir_t = t.clone('subdir')
480
 
        self.assertEqual(server.get_url(), subdir_t.abspath('/'))
481
 
        server.stop_server()
482
 
 
483
 
    def make_pf_transport(self, filter_func=None):
484
 
        """Make a PathFilteringTransport backed by a MemoryTransport.
485
 
        
486
 
        :param filter_func: by default this will be a no-op function.  Use this
487
 
            parameter to override it."""
488
 
        if filter_func is None:
489
 
            filter_func = lambda x: x
490
 
        server = pathfilter.PathFilteringServer(
491
 
            transport.get_transport('memory:///foo/bar/'), filter_func)
492
 
        server.start_server()
493
 
        self.addCleanup(server.stop_server)
494
 
        return transport.get_transport(server.get_url())
495
 
 
496
 
    def test__filter(self):
497
 
        # _filter (with an identity func as filter_func) always returns
498
 
        # paths relative to the base of the backing transport.
499
 
        t = self.make_pf_transport()
500
 
        self.assertEqual('foo', t._filter('foo'))
501
 
        self.assertEqual('foo/bar', t._filter('foo/bar'))
502
 
        self.assertEqual('', t._filter('..'))
503
 
        self.assertEqual('', t._filter('/'))
504
 
        # The base of the pathfiltering transport is taken into account too.
505
 
        t = t.clone('subdir1/subdir2')
506
 
        self.assertEqual('subdir1/subdir2/foo', t._filter('foo'))
507
 
        self.assertEqual('subdir1/subdir2/foo/bar', t._filter('foo/bar'))
508
 
        self.assertEqual('subdir1', t._filter('..'))
509
 
        self.assertEqual('', t._filter('/'))
510
 
 
511
 
    def test_filter_invocation(self):
512
 
        filter_log = []
513
 
        def filter(path):
514
 
            filter_log.append(path)
515
 
            return path
516
 
        t = self.make_pf_transport(filter)
517
 
        t.has('abc')
518
 
        self.assertEqual(['abc'], filter_log)
519
 
        del filter_log[:]
520
 
        t.clone('abc').has('xyz')
521
 
        self.assertEqual(['abc/xyz'], filter_log)
522
 
        del filter_log[:]
523
 
        t.has('/abc')
524
 
        self.assertEqual(['abc'], filter_log)
525
 
 
526
 
    def test_clone(self):
527
 
        t = self.make_pf_transport()
528
 
        # relpath from root and root path are the same
529
 
        relpath_cloned = t.clone('foo')
530
 
        abspath_cloned = t.clone('/foo')
531
 
        self.assertEqual(t.server, relpath_cloned.server)
532
 
        self.assertEqual(t.server, abspath_cloned.server)
533
 
 
534
 
    def test_url_preserves_pathfiltering(self):
535
 
        """Calling get_transport on a pathfiltered transport's base should
536
 
        produce a transport with exactly the same behaviour as the original
537
 
        pathfiltered transport.
538
 
 
539
 
        This is so that it is not possible to escape (accidentally or
540
 
        otherwise) the filtering by doing::
541
 
            url = filtered_transport.base
542
 
            parent_url = urlutils.join(url, '..')
543
 
            new_t = transport.get_transport(parent_url)
544
 
        """
545
 
        t = self.make_pf_transport()
546
 
        new_t = transport.get_transport(t.base)
547
 
        self.assertEqual(t.server, new_t.server)
548
 
        self.assertEqual(t.base, new_t.base)
549
 
 
550
 
 
551
 
class ReadonlyDecoratorTransportTest(tests.TestCase):
 
428
        backing_transport = MemoryTransport()
 
429
        server = ChrootServer(backing_transport)
 
430
        server.setUp()
 
431
        self.assertEqual('chroot-%d:///' % id(server), server.get_url())
 
432
        server.tearDown()
 
433
 
 
434
 
 
435
class ReadonlyDecoratorTransportTest(TestCase):
552
436
    """Readonly decoration specific tests."""
553
437
 
554
438
    def test_local_parameters(self):
 
439
        import bzrlib.transport.readonly as readonly
555
440
        # connect to . in readonly mode
556
 
        t = readonly.ReadonlyTransportDecorator('readonly+.')
557
 
        self.assertEqual(True, t.listable())
558
 
        self.assertEqual(True, t.is_readonly())
 
441
        transport = readonly.ReadonlyTransportDecorator('readonly+.')
 
442
        self.assertEqual(True, transport.listable())
 
443
        self.assertEqual(True, transport.is_readonly())
559
444
 
560
445
    def test_http_parameters(self):
561
446
        from bzrlib.tests.http_server import HttpServer
 
447
        import bzrlib.transport.readonly as readonly
562
448
        # connect to '.' via http which is not listable
563
449
        server = HttpServer()
564
 
        self.start_server(server)
565
 
        t = transport.get_transport('readonly+' + server.get_url())
566
 
        self.failUnless(isinstance(t, readonly.ReadonlyTransportDecorator))
567
 
        self.assertEqual(False, t.listable())
568
 
        self.assertEqual(True, t.is_readonly())
569
 
 
570
 
 
571
 
class FakeNFSDecoratorTests(tests.TestCaseInTempDir):
 
450
        server.setUp()
 
451
        try:
 
452
            transport = get_transport('readonly+' + server.get_url())
 
453
            self.failUnless(isinstance(transport,
 
454
                                       readonly.ReadonlyTransportDecorator))
 
455
            self.assertEqual(False, transport.listable())
 
456
            self.assertEqual(True, transport.is_readonly())
 
457
        finally:
 
458
            server.tearDown()
 
459
 
 
460
 
 
461
class FakeNFSDecoratorTests(TestCaseInTempDir):
572
462
    """NFS decorator specific tests."""
573
463
 
574
464
    def get_nfs_transport(self, url):
 
465
        import bzrlib.transport.fakenfs as fakenfs
575
466
        # connect to url with nfs decoration
576
467
        return fakenfs.FakeNFSTransportDecorator('fakenfs+' + url)
577
468
 
578
469
    def test_local_parameters(self):
579
470
        # the listable and is_readonly parameters
580
471
        # are not changed by the fakenfs decorator
581
 
        t = self.get_nfs_transport('.')
582
 
        self.assertEqual(True, t.listable())
583
 
        self.assertEqual(False, t.is_readonly())
 
472
        transport = self.get_nfs_transport('.')
 
473
        self.assertEqual(True, transport.listable())
 
474
        self.assertEqual(False, transport.is_readonly())
584
475
 
585
476
    def test_http_parameters(self):
586
477
        # the listable and is_readonly parameters
588
479
        from bzrlib.tests.http_server import HttpServer
589
480
        # connect to '.' via http which is not listable
590
481
        server = HttpServer()
591
 
        self.start_server(server)
592
 
        t = self.get_nfs_transport(server.get_url())
593
 
        self.assertIsInstance(t, fakenfs.FakeNFSTransportDecorator)
594
 
        self.assertEqual(False, t.listable())
595
 
        self.assertEqual(True, t.is_readonly())
 
482
        server.setUp()
 
483
        try:
 
484
            transport = self.get_nfs_transport(server.get_url())
 
485
            self.assertIsInstance(
 
486
                transport, bzrlib.transport.fakenfs.FakeNFSTransportDecorator)
 
487
            self.assertEqual(False, transport.listable())
 
488
            self.assertEqual(True, transport.is_readonly())
 
489
        finally:
 
490
            server.tearDown()
596
491
 
597
492
    def test_fakenfs_server_default(self):
598
493
        # a FakeNFSServer() should bring up a local relpath server for itself
599
 
        server = test_server.FakeNFSServer()
600
 
        self.start_server(server)
601
 
        # the url should be decorated appropriately
602
 
        self.assertStartsWith(server.get_url(), 'fakenfs+')
603
 
        # and we should be able to get a transport for it
604
 
        t = transport.get_transport(server.get_url())
605
 
        # which must be a FakeNFSTransportDecorator instance.
606
 
        self.assertIsInstance(t, fakenfs.FakeNFSTransportDecorator)
 
494
        import bzrlib.transport.fakenfs as fakenfs
 
495
        server = fakenfs.FakeNFSServer()
 
496
        server.setUp()
 
497
        try:
 
498
            # the url should be decorated appropriately
 
499
            self.assertStartsWith(server.get_url(), 'fakenfs+')
 
500
            # and we should be able to get a transport for it
 
501
            transport = get_transport(server.get_url())
 
502
            # which must be a FakeNFSTransportDecorator instance.
 
503
            self.assertIsInstance(
 
504
                transport, fakenfs.FakeNFSTransportDecorator)
 
505
        finally:
 
506
            server.tearDown()
607
507
 
608
508
    def test_fakenfs_rename_semantics(self):
609
509
        # a FakeNFS transport must mangle the way rename errors occur to
610
510
        # look like NFS problems.
611
 
        t = self.get_nfs_transport('.')
 
511
        transport = self.get_nfs_transport('.')
612
512
        self.build_tree(['from/', 'from/foo', 'to/', 'to/bar'],
613
 
                        transport=t)
614
 
        self.assertRaises(errors.ResourceBusy, t.rename, 'from', 'to')
615
 
 
616
 
 
617
 
class FakeVFATDecoratorTests(tests.TestCaseInTempDir):
 
513
                        transport=transport)
 
514
        self.assertRaises(errors.ResourceBusy,
 
515
                          transport.rename, 'from', 'to')
 
516
 
 
517
 
 
518
class FakeVFATDecoratorTests(TestCaseInTempDir):
618
519
    """Tests for simulation of VFAT restrictions"""
619
520
 
620
521
    def get_vfat_transport(self, url):
624
525
 
625
526
    def test_transport_creation(self):
626
527
        from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
627
 
        t = self.get_vfat_transport('.')
628
 
        self.assertIsInstance(t, FakeVFATTransportDecorator)
 
528
        transport = self.get_vfat_transport('.')
 
529
        self.assertIsInstance(transport, FakeVFATTransportDecorator)
629
530
 
630
531
    def test_transport_mkdir(self):
631
 
        t = self.get_vfat_transport('.')
632
 
        t.mkdir('HELLO')
633
 
        self.assertTrue(t.has('hello'))
634
 
        self.assertTrue(t.has('Hello'))
 
532
        transport = self.get_vfat_transport('.')
 
533
        transport.mkdir('HELLO')
 
534
        self.assertTrue(transport.has('hello'))
 
535
        self.assertTrue(transport.has('Hello'))
635
536
 
636
537
    def test_forbidden_chars(self):
637
 
        t = self.get_vfat_transport('.')
638
 
        self.assertRaises(ValueError, t.has, "<NU>")
639
 
 
640
 
 
641
 
class BadTransportHandler(transport.Transport):
 
538
        transport = self.get_vfat_transport('.')
 
539
        self.assertRaises(ValueError, transport.has, "<NU>")
 
540
 
 
541
 
 
542
class BadTransportHandler(Transport):
642
543
    def __init__(self, base_url):
643
 
        raise errors.DependencyNotPresent('some_lib',
644
 
                                          'testing missing dependency')
645
 
 
646
 
 
647
 
class BackupTransportHandler(transport.Transport):
 
544
        raise DependencyNotPresent('some_lib', 'testing missing dependency')
 
545
 
 
546
 
 
547
class BackupTransportHandler(Transport):
648
548
    """Test transport that works as a backup for the BadTransportHandler"""
649
549
    pass
650
550
 
651
551
 
652
 
class TestTransportImplementation(tests.TestCaseInTempDir):
 
552
class TestTransportImplementation(TestCaseInTempDir):
653
553
    """Implementation verification for transports.
654
 
 
 
554
    
655
555
    To verify a transport we need a server factory, which is a callable
656
556
    that accepts no parameters and returns an implementation of
657
557
    bzrlib.transport.Server.
658
 
 
 
558
    
659
559
    That Server is then used to construct transport instances and test
660
560
    the transport via loopback activity.
661
561
 
662
 
    Currently this assumes that the Transport object is connected to the
663
 
    current working directory.  So that whatever is done
664
 
    through the transport, should show up in the working
 
562
    Currently this assumes that the Transport object is connected to the 
 
563
    current working directory.  So that whatever is done 
 
564
    through the transport, should show up in the working 
665
565
    directory, and vice-versa. This is a bug, because its possible to have
666
 
    URL schemes which provide access to something that may not be
667
 
    result in storage on the local disk, i.e. due to file system limits, or
 
566
    URL schemes which provide access to something that may not be 
 
567
    result in storage on the local disk, i.e. due to file system limits, or 
668
568
    due to it being a database or some other non-filesystem tool.
669
569
 
670
570
    This also tests to make sure that the functions work with both
671
571
    generators and lists (assuming iter(list) is effectively a generator)
672
572
    """
673
 
 
 
573
    
674
574
    def setUp(self):
675
575
        super(TestTransportImplementation, self).setUp()
676
576
        self._server = self.transport_server()
677
 
        self.start_server(self._server)
 
577
        self._server.setUp()
 
578
        self.addCleanup(self._server.tearDown)
678
579
 
679
580
    def get_transport(self, relpath=None):
680
581
        """Return a connected transport to the local directory.
684
585
        base_url = self._server.get_url()
685
586
        url = self._adjust_url(base_url, relpath)
686
587
        # try getting the transport via the regular interface:
687
 
        t = transport.get_transport(url)
 
588
        t = get_transport(url)
688
589
        # vila--20070607 if the following are commented out the test suite
689
590
        # still pass. Is this really still needed or was it a forgotten
690
591
        # temporary fix ?
695
596
        return t
696
597
 
697
598
 
698
 
class TestLocalTransports(tests.TestCase):
 
599
class TestLocalTransports(TestCase):
699
600
 
700
601
    def test_get_transport_from_abspath(self):
701
602
        here = osutils.abspath('.')
702
 
        t = transport.get_transport(here)
703
 
        self.assertIsInstance(t, local.LocalTransport)
 
603
        t = get_transport(here)
 
604
        self.assertIsInstance(t, LocalTransport)
704
605
        self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
705
606
 
706
607
    def test_get_transport_from_relpath(self):
707
608
        here = osutils.abspath('.')
708
 
        t = transport.get_transport('.')
709
 
        self.assertIsInstance(t, local.LocalTransport)
 
609
        t = get_transport('.')
 
610
        self.assertIsInstance(t, LocalTransport)
710
611
        self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
711
612
 
712
613
    def test_get_transport_from_local_url(self):
713
614
        here = osutils.abspath('.')
714
615
        here_url = urlutils.local_path_to_url(here) + '/'
715
 
        t = transport.get_transport(here_url)
716
 
        self.assertIsInstance(t, local.LocalTransport)
 
616
        t = get_transport(here_url)
 
617
        self.assertIsInstance(t, LocalTransport)
717
618
        self.assertEquals(t.base, here_url)
718
619
 
719
620
    def test_local_abspath(self):
720
621
        here = osutils.abspath('.')
721
 
        t = transport.get_transport(here)
 
622
        t = get_transport(here)
722
623
        self.assertEquals(t.local_abspath(''), here)
723
624
 
724
625
 
725
 
class TestWin32LocalTransport(tests.TestCase):
 
626
class TestWin32LocalTransport(TestCase):
726
627
 
727
628
    def test_unc_clone_to_root(self):
728
629
        # Win32 UNC path like \\HOST\path
729
630
        # clone to root should stop at least at \\HOST part
730
631
        # not on \\
731
 
        t = local.EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
 
632
        t = EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
732
633
        for i in xrange(4):
733
634
            t = t.clone('..')
734
635
        self.assertEquals(t.base, 'file://HOST/')
737
638
        self.assertEquals(t.base, 'file://HOST/')
738
639
 
739
640
 
740
 
class TestConnectedTransport(tests.TestCase):
 
641
class TestConnectedTransport(TestCase):
741
642
    """Tests for connected to remote server transports"""
742
643
 
743
644
    def test_parse_url(self):
744
 
        t = transport.ConnectedTransport(
745
 
            'http://simple.example.com/home/source')
 
645
        t = ConnectedTransport('http://simple.example.com/home/source')
746
646
        self.assertEquals(t._host, 'simple.example.com')
747
647
        self.assertEquals(t._port, None)
748
648
        self.assertEquals(t._path, '/home/source/')
753
653
 
754
654
    def test_parse_url_with_at_in_user(self):
755
655
        # Bug 228058
756
 
        t = transport.ConnectedTransport('ftp://user@host.com@www.host.com/')
 
656
        t = ConnectedTransport('ftp://user@host.com@www.host.com/')
757
657
        self.assertEquals(t._user, 'user@host.com')
758
658
 
759
659
    def test_parse_quoted_url(self):
760
 
        t = transport.ConnectedTransport(
761
 
            'http://ro%62ey:h%40t@ex%41mple.com:2222/path')
 
660
        t = ConnectedTransport('http://ro%62ey:h%40t@ex%41mple.com:2222/path')
762
661
        self.assertEquals(t._host, 'exAmple.com')
763
662
        self.assertEquals(t._port, 2222)
764
663
        self.assertEquals(t._user, 'robey')
770
669
 
771
670
    def test_parse_invalid_url(self):
772
671
        self.assertRaises(errors.InvalidURL,
773
 
                          transport.ConnectedTransport,
 
672
                          ConnectedTransport,
774
673
                          'sftp://lily.org:~janneke/public/bzr/gub')
775
674
 
776
675
    def test_relpath(self):
777
 
        t = transport.ConnectedTransport('sftp://user@host.com/abs/path')
 
676
        t = ConnectedTransport('sftp://user@host.com/abs/path')
778
677
 
779
678
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
780
679
        self.assertRaises(errors.PathNotChild, t.relpath,
786
685
        self.assertRaises(errors.PathNotChild, t.relpath,
787
686
                          'sftp://user@host.com:33/abs/path/sub')
788
687
        # Make sure it works when we don't supply a username
789
 
        t = transport.ConnectedTransport('sftp://host.com/abs/path')
 
688
        t = ConnectedTransport('sftp://host.com/abs/path')
790
689
        self.assertEquals(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
791
690
 
792
691
        # Make sure it works when parts of the path will be url encoded
793
 
        t = transport.ConnectedTransport('sftp://host.com/dev/%path')
 
692
        t = ConnectedTransport('sftp://host.com/dev/%path')
794
693
        self.assertEquals(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
795
694
 
796
695
    def test_connection_sharing_propagate_credentials(self):
797
 
        t = transport.ConnectedTransport('ftp://user@host.com/abs/path')
 
696
        t = ConnectedTransport('ftp://user@host.com/abs/path')
798
697
        self.assertEquals('user', t._user)
799
698
        self.assertEquals('host.com', t._host)
800
699
        self.assertIs(None, t._get_connection())
821
720
        self.assertIs(new_password, c._get_credentials())
822
721
 
823
722
 
824
 
class TestReusedTransports(tests.TestCase):
 
723
class TestReusedTransports(TestCase):
825
724
    """Tests for transport reuse"""
826
725
 
827
726
    def test_reuse_same_transport(self):
828
727
        possible_transports = []
829
 
        t1 = transport.get_transport('http://foo/',
830
 
                                     possible_transports=possible_transports)
 
728
        t1 = get_transport('http://foo/',
 
729
                           possible_transports=possible_transports)
831
730
        self.assertEqual([t1], possible_transports)
832
 
        t2 = transport.get_transport('http://foo/',
833
 
                                     possible_transports=[t1])
 
731
        t2 = get_transport('http://foo/', possible_transports=[t1])
834
732
        self.assertIs(t1, t2)
835
733
 
836
734
        # Also check that final '/' are handled correctly
837
 
        t3 = transport.get_transport('http://foo/path/')
838
 
        t4 = transport.get_transport('http://foo/path',
839
 
                                     possible_transports=[t3])
 
735
        t3 = get_transport('http://foo/path/')
 
736
        t4 = get_transport('http://foo/path', possible_transports=[t3])
840
737
        self.assertIs(t3, t4)
841
738
 
842
 
        t5 = transport.get_transport('http://foo/path')
843
 
        t6 = transport.get_transport('http://foo/path/',
844
 
                                     possible_transports=[t5])
 
739
        t5 = get_transport('http://foo/path')
 
740
        t6 = get_transport('http://foo/path/', possible_transports=[t5])
845
741
        self.assertIs(t5, t6)
846
742
 
847
743
    def test_don_t_reuse_different_transport(self):
848
 
        t1 = transport.get_transport('http://foo/path')
849
 
        t2 = transport.get_transport('http://bar/path',
850
 
                                     possible_transports=[t1])
 
744
        t1 = get_transport('http://foo/path')
 
745
        t2 = get_transport('http://bar/path', possible_transports=[t1])
851
746
        self.assertIsNot(t1, t2)
852
747
 
853
748
 
854
 
class TestTransportTrace(tests.TestCase):
 
749
class TestTransportTrace(TestCase):
855
750
 
856
751
    def test_get(self):
857
 
        t = transport.get_transport('trace+memory://')
858
 
        self.assertIsInstance(t, bzrlib.transport.trace.TransportTraceDecorator)
 
752
        transport = get_transport('trace+memory://')
 
753
        self.assertIsInstance(
 
754
            transport, bzrlib.transport.trace.TransportTraceDecorator)
859
755
 
860
756
    def test_clone_preserves_activity(self):
861
 
        t = transport.get_transport('trace+memory://')
862
 
        t2 = t.clone('.')
863
 
        self.assertTrue(t is not t2)
864
 
        self.assertTrue(t._activity is t2._activity)
 
757
        transport = get_transport('trace+memory://')
 
758
        transport2 = transport.clone('.')
 
759
        self.assertTrue(transport is not transport2)
 
760
        self.assertTrue(transport._activity is transport2._activity)
865
761
 
866
762
    # the following specific tests are for the operations that have made use of
867
763
    # logging in tests; we could test every single operation but doing that
868
764
    # still won't cause a test failure when the top level Transport API
869
765
    # changes; so there is little return doing that.
870
766
    def test_get(self):
871
 
        t = transport.get_transport('trace+memory:///')
872
 
        t.put_bytes('foo', 'barish')
873
 
        t.get('foo')
 
767
        transport = get_transport('trace+memory:///')
 
768
        transport.put_bytes('foo', 'barish')
 
769
        transport.get('foo')
874
770
        expected_result = []
875
771
        # put_bytes records the bytes, not the content to avoid memory
876
772
        # pressure.
877
773
        expected_result.append(('put_bytes', 'foo', 6, None))
878
774
        # get records the file name only.
879
775
        expected_result.append(('get', 'foo'))
880
 
        self.assertEqual(expected_result, t._activity)
 
776
        self.assertEqual(expected_result, transport._activity)
881
777
 
882
778
    def test_readv(self):
883
 
        t = transport.get_transport('trace+memory:///')
884
 
        t.put_bytes('foo', 'barish')
885
 
        list(t.readv('foo', [(0, 1), (3, 2)],
886
 
                     adjust_for_latency=True, upper_limit=6))
 
779
        transport = get_transport('trace+memory:///')
 
780
        transport.put_bytes('foo', 'barish')
 
781
        list(transport.readv('foo', [(0, 1), (3, 2)], adjust_for_latency=True,
 
782
            upper_limit=6))
887
783
        expected_result = []
888
784
        # put_bytes records the bytes, not the content to avoid memory
889
785
        # pressure.
890
786
        expected_result.append(('put_bytes', 'foo', 6, None))
891
787
        # readv records the supplied offset request
892
788
        expected_result.append(('readv', 'foo', [(0, 1), (3, 2)], True, 6))
893
 
        self.assertEqual(expected_result, t._activity)
894
 
 
895
 
 
896
 
class TestSSHConnections(tests.TestCaseWithTransport):
897
 
 
898
 
    def test_bzr_connect_to_bzr_ssh(self):
899
 
        """User acceptance that get_transport of a bzr+ssh:// behaves correctly.
900
 
 
901
 
        bzr+ssh:// should cause bzr to run a remote bzr smart server over SSH.
902
 
        """
903
 
        raise tests.TestSkipped('this test was recently broken,'
904
 
                                ' see bug #626876')
905
 
        # This test actually causes a bzr instance to be invoked, which is very
906
 
        # expensive: it should be the only such test in the test suite.
907
 
        # A reasonable evolution for this would be to simply check inside
908
 
        # check_channel_exec_request that the command is appropriate, and then
909
 
        # satisfy requests in-process.
910
 
        self.requireFeature(features.paramiko)
911
 
        # SFTPFullAbsoluteServer has a get_url method, and doesn't
912
 
        # override the interface (doesn't change self._vendor).
913
 
        # Note that this does encryption, so can be slow.
914
 
        from bzrlib.tests import stub_sftp
915
 
 
916
 
        # Start an SSH server
917
 
        self.command_executed = []
918
 
        # XXX: This is horrible -- we define a really dumb SSH server that
919
 
        # executes commands, and manage the hooking up of stdin/out/err to the
920
 
        # SSH channel ourselves.  Surely this has already been implemented
921
 
        # elsewhere?
922
 
        started = []
923
 
        class StubSSHServer(stub_sftp.StubServer):
924
 
 
925
 
            test = self
926
 
 
927
 
            def check_channel_exec_request(self, channel, command):
928
 
                self.test.command_executed.append(command)
929
 
                proc = subprocess.Popen(
930
 
                    command, shell=True, stdin=subprocess.PIPE,
931
 
                    stdout=subprocess.PIPE, stderr=subprocess.PIPE)
932
 
 
933
 
                # XXX: horribly inefficient, not to mention ugly.
934
 
                # Start a thread for each of stdin/out/err, and relay bytes from
935
 
                # the subprocess to channel and vice versa.
936
 
                def ferry_bytes(read, write, close):
937
 
                    while True:
938
 
                        bytes = read(1)
939
 
                        if bytes == '':
940
 
                            close()
941
 
                            break
942
 
                        write(bytes)
943
 
 
944
 
                file_functions = [
945
 
                    (channel.recv, proc.stdin.write, proc.stdin.close),
946
 
                    (proc.stdout.read, channel.sendall, channel.close),
947
 
                    (proc.stderr.read, channel.sendall_stderr, channel.close)]
948
 
                started.append(proc)
949
 
                for read, write, close in file_functions:
950
 
                    t = threading.Thread(
951
 
                        target=ferry_bytes, args=(read, write, close))
952
 
                    t.start()
953
 
                    started.append(t)
954
 
 
955
 
                return True
956
 
 
957
 
        ssh_server = stub_sftp.SFTPFullAbsoluteServer(StubSSHServer)
958
 
        # We *don't* want to override the default SSH vendor: the detected one
959
 
        # is the one to use.
960
 
 
961
 
        # FIXME: I don't understand the above comment, SFTPFullAbsoluteServer
962
 
        # inherits from SFTPServer which forces the SSH vendor to
963
 
        # ssh.ParamikoVendor(). So it's forced, not detected. --vila 20100623
964
 
        self.start_server(ssh_server)
965
 
        port = ssh_server.port
966
 
 
967
 
        if sys.platform == 'win32':
968
 
            bzr_remote_path = sys.executable + ' ' + self.get_bzr_path()
969
 
        else:
970
 
            bzr_remote_path = self.get_bzr_path()
971
 
        os.environ['BZR_REMOTE_PATH'] = bzr_remote_path
972
 
 
973
 
        # Access the branch via a bzr+ssh URL.  The BZR_REMOTE_PATH environment
974
 
        # variable is used to tell bzr what command to run on the remote end.
975
 
        path_to_branch = osutils.abspath('.')
976
 
        if sys.platform == 'win32':
977
 
            # On Windows, we export all drives as '/C:/, etc. So we need to
978
 
            # prefix a '/' to get the right path.
979
 
            path_to_branch = '/' + path_to_branch
980
 
        url = 'bzr+ssh://fred:secret@localhost:%d%s' % (port, path_to_branch)
981
 
        t = transport.get_transport(url)
982
 
        self.permit_url(t.base)
983
 
        t.mkdir('foo')
984
 
 
985
 
        self.assertEqual(
986
 
            ['%s serve --inet --directory=/ --allow-writes' % bzr_remote_path],
987
 
            self.command_executed)
988
 
        # Make sure to disconnect, so that the remote process can stop, and we
989
 
        # can cleanup. Then pause the test until everything is shutdown
990
 
        t._client._medium.disconnect()
991
 
        if not started:
992
 
            return
993
 
        # First wait for the subprocess
994
 
        started[0].wait()
995
 
        # And the rest are threads
996
 
        for t in started[1:]:
997
 
            t.join()
 
789
        self.assertEqual(expected_result, transport._activity)