← 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: Daniel Watkins
  • Date: 2009-02-08 14:26:35 UTC
  • mto: (3989.1.1 bzr.ab.integration)
  • mto: This revision was merged to the branch mainline in revision 3990.
  • Revision ID: daniel@daniel-watkins.co.uk-20090208142635-zyzycxxz7i6p3dih
Reverted some irrelevant changes.

Show diffs side-by-side

added added

removed removed

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