← 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: Robert Collins
  • Date: 2009-09-17 11:54:41 UTC
  • mto: This revision was merged to the branch mainline in revision 4700.
  • Revision ID: robertc@robertcollins.net-20090917115441-2ug57z6eyrnb6zim
Add stronger test isolation by interception BzrDir.open and checking the thing being opened is known to the test suite.

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
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)
 
109
            _set_protocol_handlers(saved_handlers)
108
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)
 
124
            _set_protocol_handlers(saved_handlers)
123
125
 
124
126
    def test_ssh_hints(self):
125
127
        """Transport ssh:// should raise an error pointing out bzr+ssh://"""
126
128
        try:
127
 
            transport.get_transport('ssh://fooserver/foo')
128
 
        except errors.UnsupportedProtocol, e:
 
129
            get_transport('ssh://fooserver/foo')
 
130
        except UnsupportedProtocol, e:
129
131
            e_str = str(e)
130
132
            self.assertEquals('Unsupported protocol'
131
133
                              ' for url "ssh://fooserver/foo":'
132
 
                              ' bzr supports bzr+ssh to operate over ssh,'
133
 
                              ' use "bzr+ssh://fooserver/foo".',
 
134
                              ' bzr supports bzr+ssh to operate over ssh, use "bzr+ssh://fooserver/foo".',
134
135
                              str(e))
135
136
        else:
136
137
            self.fail('Did not raise UnsupportedProtocol')
137
138
 
138
139
    def test_LateReadError(self):
139
140
        """The LateReadError helper should raise on read()."""
140
 
        a_file = transport.LateReadError('a path')
 
141
        a_file = LateReadError('a path')
141
142
        try:
142
143
            a_file.read()
143
 
        except errors.ReadError, error:
 
144
        except ReadError, error:
144
145
            self.assertEqual('a path', error.path)
145
 
        self.assertRaises(errors.ReadError, a_file.read, 40)
 
146
        self.assertRaises(ReadError, a_file.read, 40)
146
147
        a_file.close()
147
148
 
148
149
    def test__combine_paths(self):
149
 
        t = transport.Transport('/')
 
150
        t = Transport('/')
150
151
        self.assertEqual('/home/sarah/project/foo',
151
152
                         t._combine_paths('/home/sarah', 'project/foo'))
152
153
        self.assertEqual('/etc',
158
159
 
159
160
    def test_local_abspath_non_local_transport(self):
160
161
        # the base implementation should throw
161
 
        t = memory.MemoryTransport()
 
162
        t = MemoryTransport()
162
163
        e = self.assertRaises(errors.NotLocalUrl, t.local_abspath, 't')
163
164
        self.assertEqual('memory:///t is not a local path.', str(e))
164
165
 
165
166
 
166
 
class TestCoalesceOffsets(tests.TestCase):
 
167
class TestCoalesceOffsets(TestCase):
167
168
 
168
169
    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]
 
170
        coalesce = Transport._coalesce_offsets
 
171
        exp = [_CoalescedOffset(*x) for x in expected]
171
172
        out = list(coalesce(offsets, limit=limit, fudge_factor=fudge,
172
173
                            max_size=max_size))
173
174
        self.assertEqual(exp, out)
248
249
                   max_size=1*1024*1024*1024)
249
250
 
250
251
 
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):
 
252
class TestMemoryTransport(TestCase):
267
253
 
268
254
    def test_get_transport(self):
269
 
        memory.MemoryTransport()
 
255
        MemoryTransport()
270
256
 
271
257
    def test_clone(self):
272
 
        t = memory.MemoryTransport()
273
 
        self.assertTrue(isinstance(t, memory.MemoryTransport))
274
 
        self.assertEqual("memory:///", t.clone("/").base)
 
258
        transport = MemoryTransport()
 
259
        self.assertTrue(isinstance(transport, MemoryTransport))
 
260
        self.assertEqual("memory:///", transport.clone("/").base)
275
261
 
276
262
    def test_abspath(self):
277
 
        t = memory.MemoryTransport()
278
 
        self.assertEqual("memory:///relpath", t.abspath('relpath'))
 
263
        transport = MemoryTransport()
 
264
        self.assertEqual("memory:///relpath", transport.abspath('relpath'))
279
265
 
280
266
    def test_abspath_of_root(self):
281
 
        t = memory.MemoryTransport()
282
 
        self.assertEqual("memory:///", t.base)
283
 
        self.assertEqual("memory:///", t.abspath('/'))
 
267
        transport = MemoryTransport()
 
268
        self.assertEqual("memory:///", transport.base)
 
269
        self.assertEqual("memory:///", transport.abspath('/'))
284
270
 
285
271
    def test_abspath_of_relpath_starting_at_root(self):
286
 
        t = memory.MemoryTransport()
287
 
        self.assertEqual("memory:///foo", t.abspath('/foo'))
 
272
        transport = MemoryTransport()
 
273
        self.assertEqual("memory:///foo", transport.abspath('/foo'))
288
274
 
289
275
    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')
 
276
        transport = MemoryTransport()
 
277
        transport.append_bytes('path', 'content')
 
278
        self.assertEqual(transport.get('path').read(), 'content')
 
279
        transport.append_file('path', StringIO('content'))
 
280
        self.assertEqual(transport.get('path').read(), 'contentcontent')
295
281
 
296
282
    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')
 
283
        transport = MemoryTransport()
 
284
        transport.put_file('path', StringIO('content'))
 
285
        self.assertEqual(transport.get('path').read(), 'content')
 
286
        transport.put_bytes('path', 'content')
 
287
        self.assertEqual(transport.get('path').read(), 'content')
302
288
 
303
289
    def test_append_without_dir_fails(self):
304
 
        t = memory.MemoryTransport()
305
 
        self.assertRaises(errors.NoSuchFile,
306
 
                          t.append_bytes, 'dir/path', 'content')
 
290
        transport = MemoryTransport()
 
291
        self.assertRaises(NoSuchFile,
 
292
                          transport.append_bytes, 'dir/path', 'content')
307
293
 
308
294
    def test_put_without_dir_fails(self):
309
 
        t = memory.MemoryTransport()
310
 
        self.assertRaises(errors.NoSuchFile,
311
 
                          t.put_file, 'dir/path', StringIO('content'))
 
295
        transport = MemoryTransport()
 
296
        self.assertRaises(NoSuchFile,
 
297
                          transport.put_file, 'dir/path', StringIO('content'))
312
298
 
313
299
    def test_get_missing(self):
314
 
        transport = memory.MemoryTransport()
315
 
        self.assertRaises(errors.NoSuchFile, transport.get, 'foo')
 
300
        transport = MemoryTransport()
 
301
        self.assertRaises(NoSuchFile, transport.get, 'foo')
316
302
 
317
303
    def test_has_missing(self):
318
 
        t = memory.MemoryTransport()
319
 
        self.assertEquals(False, t.has('foo'))
 
304
        transport = MemoryTransport()
 
305
        self.assertEquals(False, transport.has('foo'))
320
306
 
321
307
    def test_has_present(self):
322
 
        t = memory.MemoryTransport()
323
 
        t.append_bytes('foo', 'content')
324
 
        self.assertEquals(True, t.has('foo'))
 
308
        transport = MemoryTransport()
 
309
        transport.append_bytes('foo', 'content')
 
310
        self.assertEquals(True, transport.has('foo'))
325
311
 
326
312
    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')
 
313
        transport = MemoryTransport()
 
314
        transport.put_bytes('foo', 'content')
 
315
        transport.mkdir('dir')
 
316
        transport.put_bytes('dir/subfoo', 'content')
 
317
        transport.put_bytes('dirlike', 'content')
332
318
 
333
 
        self.assertEquals(['dir', 'dirlike', 'foo'], sorted(t.list_dir('.')))
334
 
        self.assertEquals(['subfoo'], sorted(t.list_dir('dir')))
 
319
        self.assertEquals(['dir', 'dirlike', 'foo'], sorted(transport.list_dir('.')))
 
320
        self.assertEquals(['subfoo'], sorted(transport.list_dir('dir')))
335
321
 
336
322
    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')
 
323
        transport = MemoryTransport()
 
324
        transport.mkdir('dir')
 
325
        transport.append_bytes('dir/path', 'content')
 
326
        self.assertEqual(transport.get('dir/path').read(), 'content')
341
327
 
342
328
    def test_mkdir_missing_parent(self):
343
 
        t = memory.MemoryTransport()
344
 
        self.assertRaises(errors.NoSuchFile, t.mkdir, 'dir/dir')
 
329
        transport = MemoryTransport()
 
330
        self.assertRaises(NoSuchFile,
 
331
                          transport.mkdir, 'dir/dir')
345
332
 
346
333
    def test_mkdir_twice(self):
347
 
        t = memory.MemoryTransport()
348
 
        t.mkdir('dir')
349
 
        self.assertRaises(errors.FileExists, t.mkdir, 'dir')
 
334
        transport = MemoryTransport()
 
335
        transport.mkdir('dir')
 
336
        self.assertRaises(FileExists, transport.mkdir, 'dir')
350
337
 
351
338
    def test_parameters(self):
352
 
        t = memory.MemoryTransport()
353
 
        self.assertEqual(True, t.listable())
354
 
        self.assertEqual(False, t.is_readonly())
 
339
        transport = MemoryTransport()
 
340
        self.assertEqual(True, transport.listable())
 
341
        self.assertEqual(False, transport.is_readonly())
355
342
 
356
343
    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())
 
344
        transport = MemoryTransport()
 
345
        transport.mkdir('dir')
 
346
        transport.put_bytes('dir/foo', 'content')
 
347
        transport.put_bytes('dir/bar', 'content')
 
348
        transport.put_bytes('bar', 'content')
 
349
        paths = set(transport.iter_files_recursive())
363
350
        self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
364
351
 
365
352
    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):
 
353
        transport = MemoryTransport()
 
354
        transport.put_bytes('foo', 'content')
 
355
        transport.put_bytes('bar', 'phowar')
 
356
        self.assertEqual(7, transport.stat('foo').st_size)
 
357
        self.assertEqual(6, transport.stat('bar').st_size)
 
358
 
 
359
 
 
360
class ChrootDecoratorTransportTest(TestCase):
374
361
    """Chroot decoration specific tests."""
375
362
 
376
363
    def test_abspath(self):
377
364
        # The abspath is always relative to the chroot_url.
378
 
        server = chroot.ChrootServer(
379
 
            transport.get_transport('memory:///foo/bar/'))
 
365
        server = ChrootServer(get_transport('memory:///foo/bar/'))
380
366
        self.start_server(server)
381
 
        t = transport.get_transport(server.get_url())
382
 
        self.assertEqual(server.get_url(), t.abspath('/'))
 
367
        transport = get_transport(server.get_url())
 
368
        self.assertEqual(server.get_url(), transport.abspath('/'))
383
369
 
384
 
        subdir_t = t.clone('subdir')
385
 
        self.assertEqual(server.get_url(), subdir_t.abspath('/'))
 
370
        subdir_transport = transport.clone('subdir')
 
371
        self.assertEqual(server.get_url(), subdir_transport.abspath('/'))
386
372
 
387
373
    def test_clone(self):
388
 
        server = chroot.ChrootServer(
389
 
            transport.get_transport('memory:///foo/bar/'))
 
374
        server = ChrootServer(get_transport('memory:///foo/bar/'))
390
375
        self.start_server(server)
391
 
        t = transport.get_transport(server.get_url())
 
376
        transport = get_transport(server.get_url())
392
377
        # relpath from root and root path are the same
393
 
        relpath_cloned = t.clone('foo')
394
 
        abspath_cloned = t.clone('/foo')
 
378
        relpath_cloned = transport.clone('foo')
 
379
        abspath_cloned = transport.clone('/foo')
395
380
        self.assertEqual(server, relpath_cloned.server)
396
381
        self.assertEqual(server, abspath_cloned.server)
397
382
 
403
388
        This is so that it is not possible to escape a chroot by doing::
404
389
            url = chroot_transport.base
405
390
            parent_url = urlutils.join(url, '..')
406
 
            new_t = transport.get_transport(parent_url)
 
391
            new_transport = get_transport(parent_url)
407
392
        """
408
 
        server = chroot.ChrootServer(
409
 
            transport.get_transport('memory:///path/subpath'))
 
393
        server = ChrootServer(get_transport('memory:///path/subpath'))
410
394
        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)
 
395
        transport = get_transport(server.get_url())
 
396
        new_transport = get_transport(transport.base)
 
397
        self.assertEqual(transport.server, new_transport.server)
 
398
        self.assertEqual(transport.base, new_transport.base)
415
399
 
416
400
    def test_urljoin_preserves_chroot(self):
417
401
        """Using urlutils.join(url, '..') on a chroot URL should not produce a
420
404
        This is so that it is not possible to escape a chroot by doing::
421
405
            url = chroot_transport.base
422
406
            parent_url = urlutils.join(url, '..')
423
 
            new_t = transport.get_transport(parent_url)
 
407
            new_transport = get_transport(parent_url)
424
408
        """
425
 
        server = chroot.ChrootServer(transport.get_transport('memory:///path/'))
 
409
        server = ChrootServer(get_transport('memory:///path/'))
426
410
        self.start_server(server)
427
 
        t = transport.get_transport(server.get_url())
 
411
        transport = get_transport(server.get_url())
428
412
        self.assertRaises(
429
 
            errors.InvalidURLJoin, urlutils.join, t.base, '..')
430
 
 
431
 
 
432
 
class TestChrootServer(tests.TestCase):
 
413
            InvalidURLJoin, urlutils.join, transport.base, '..')
 
414
 
 
415
 
 
416
class ChrootServerTest(TestCase):
433
417
 
434
418
    def test_construct(self):
435
 
        backing_transport = memory.MemoryTransport()
436
 
        server = chroot.ChrootServer(backing_transport)
 
419
        backing_transport = MemoryTransport()
 
420
        server = ChrootServer(backing_transport)
437
421
        self.assertEqual(backing_transport, server.backing_transport)
438
422
 
439
423
    def test_setUp(self):
440
 
        backing_transport = memory.MemoryTransport()
441
 
        server = chroot.ChrootServer(backing_transport)
442
 
        server.start_server()
 
424
        backing_transport = MemoryTransport()
 
425
        server = ChrootServer(backing_transport)
 
426
        server.setUp()
443
427
        try:
444
 
            self.assertTrue(server.scheme
445
 
                            in transport._get_protocol_handlers().keys())
 
428
            self.assertTrue(server.scheme in _get_protocol_handlers().keys())
446
429
        finally:
447
 
            server.stop_server()
 
430
            server.tearDown()
448
431
 
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())
 
432
    def test_tearDown(self):
 
433
        backing_transport = MemoryTransport()
 
434
        server = ChrootServer(backing_transport)
 
435
        server.setUp()
 
436
        server.tearDown()
 
437
        self.assertFalse(server.scheme in _get_protocol_handlers().keys())
456
438
 
457
439
    def test_get_url(self):
458
 
        backing_transport = memory.MemoryTransport()
459
 
        server = chroot.ChrootServer(backing_transport)
460
 
        server.start_server()
 
440
        backing_transport = MemoryTransport()
 
441
        server = ChrootServer(backing_transport)
 
442
        server.setUp()
461
443
        try:
462
444
            self.assertEqual('chroot-%d:///' % id(server), server.get_url())
463
445
        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):
 
446
            server.tearDown()
 
447
 
 
448
 
 
449
class ReadonlyDecoratorTransportTest(TestCase):
552
450
    """Readonly decoration specific tests."""
553
451
 
554
452
    def test_local_parameters(self):
 
453
        import bzrlib.transport.readonly as readonly
555
454
        # connect to . in readonly mode
556
 
        t = readonly.ReadonlyTransportDecorator('readonly+.')
557
 
        self.assertEqual(True, t.listable())
558
 
        self.assertEqual(True, t.is_readonly())
 
455
        transport = readonly.ReadonlyTransportDecorator('readonly+.')
 
456
        self.assertEqual(True, transport.listable())
 
457
        self.assertEqual(True, transport.is_readonly())
559
458
 
560
459
    def test_http_parameters(self):
561
460
        from bzrlib.tests.http_server import HttpServer
 
461
        import bzrlib.transport.readonly as readonly
562
462
        # connect to '.' via http which is not listable
563
463
        server = HttpServer()
564
464
        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):
 
465
        transport = get_transport('readonly+' + server.get_url())
 
466
        self.failUnless(isinstance(transport,
 
467
                                   readonly.ReadonlyTransportDecorator))
 
468
        self.assertEqual(False, transport.listable())
 
469
        self.assertEqual(True, transport.is_readonly())
 
470
 
 
471
 
 
472
class FakeNFSDecoratorTests(TestCaseInTempDir):
572
473
    """NFS decorator specific tests."""
573
474
 
574
475
    def get_nfs_transport(self, url):
 
476
        import bzrlib.transport.fakenfs as fakenfs
575
477
        # connect to url with nfs decoration
576
478
        return fakenfs.FakeNFSTransportDecorator('fakenfs+' + url)
577
479
 
578
480
    def test_local_parameters(self):
579
481
        # the listable and is_readonly parameters
580
482
        # 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())
 
483
        transport = self.get_nfs_transport('.')
 
484
        self.assertEqual(True, transport.listable())
 
485
        self.assertEqual(False, transport.is_readonly())
584
486
 
585
487
    def test_http_parameters(self):
586
488
        # the listable and is_readonly parameters
589
491
        # connect to '.' via http which is not listable
590
492
        server = HttpServer()
591
493
        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())
 
494
        transport = self.get_nfs_transport(server.get_url())
 
495
        self.assertIsInstance(
 
496
            transport, bzrlib.transport.fakenfs.FakeNFSTransportDecorator)
 
497
        self.assertEqual(False, transport.listable())
 
498
        self.assertEqual(True, transport.is_readonly())
596
499
 
597
500
    def test_fakenfs_server_default(self):
598
501
        # a FakeNFSServer() should bring up a local relpath server for itself
599
 
        server = test_server.FakeNFSServer()
 
502
        import bzrlib.transport.fakenfs as fakenfs
 
503
        server = fakenfs.FakeNFSServer()
600
504
        self.start_server(server)
601
505
        # the url should be decorated appropriately
602
506
        self.assertStartsWith(server.get_url(), 'fakenfs+')
603
507
        # and we should be able to get a transport for it
604
 
        t = transport.get_transport(server.get_url())
 
508
        transport = get_transport(server.get_url())
605
509
        # which must be a FakeNFSTransportDecorator instance.
606
 
        self.assertIsInstance(t, fakenfs.FakeNFSTransportDecorator)
 
510
        self.assertIsInstance(transport, fakenfs.FakeNFSTransportDecorator)
607
511
 
608
512
    def test_fakenfs_rename_semantics(self):
609
513
        # a FakeNFS transport must mangle the way rename errors occur to
610
514
        # look like NFS problems.
611
 
        t = self.get_nfs_transport('.')
 
515
        transport = self.get_nfs_transport('.')
612
516
        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):
 
517
                        transport=transport)
 
518
        self.assertRaises(errors.ResourceBusy,
 
519
                          transport.rename, 'from', 'to')
 
520
 
 
521
 
 
522
class FakeVFATDecoratorTests(TestCaseInTempDir):
618
523
    """Tests for simulation of VFAT restrictions"""
619
524
 
620
525
    def get_vfat_transport(self, url):
624
529
 
625
530
    def test_transport_creation(self):
626
531
        from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
627
 
        t = self.get_vfat_transport('.')
628
 
        self.assertIsInstance(t, FakeVFATTransportDecorator)
 
532
        transport = self.get_vfat_transport('.')
 
533
        self.assertIsInstance(transport, FakeVFATTransportDecorator)
629
534
 
630
535
    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'))
 
536
        transport = self.get_vfat_transport('.')
 
537
        transport.mkdir('HELLO')
 
538
        self.assertTrue(transport.has('hello'))
 
539
        self.assertTrue(transport.has('Hello'))
635
540
 
636
541
    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):
 
542
        transport = self.get_vfat_transport('.')
 
543
        self.assertRaises(ValueError, transport.has, "<NU>")
 
544
 
 
545
 
 
546
class BadTransportHandler(Transport):
642
547
    def __init__(self, base_url):
643
 
        raise errors.DependencyNotPresent('some_lib',
644
 
                                          'testing missing dependency')
645
 
 
646
 
 
647
 
class BackupTransportHandler(transport.Transport):
 
548
        raise DependencyNotPresent('some_lib', 'testing missing dependency')
 
549
 
 
550
 
 
551
class BackupTransportHandler(Transport):
648
552
    """Test transport that works as a backup for the BadTransportHandler"""
649
553
    pass
650
554
 
651
555
 
652
 
class TestTransportImplementation(tests.TestCaseInTempDir):
 
556
class TestTransportImplementation(TestCaseInTempDir):
653
557
    """Implementation verification for transports.
654
558
 
655
559
    To verify a transport we need a server factory, which is a callable
684
588
        base_url = self._server.get_url()
685
589
        url = self._adjust_url(base_url, relpath)
686
590
        # try getting the transport via the regular interface:
687
 
        t = transport.get_transport(url)
 
591
        t = get_transport(url)
688
592
        # vila--20070607 if the following are commented out the test suite
689
593
        # still pass. Is this really still needed or was it a forgotten
690
594
        # temporary fix ?
695
599
        return t
696
600
 
697
601
 
698
 
class TestLocalTransports(tests.TestCase):
 
602
class TestLocalTransports(TestCase):
699
603
 
700
604
    def test_get_transport_from_abspath(self):
701
605
        here = osutils.abspath('.')
702
 
        t = transport.get_transport(here)
703
 
        self.assertIsInstance(t, local.LocalTransport)
 
606
        t = get_transport(here)
 
607
        self.assertIsInstance(t, LocalTransport)
704
608
        self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
705
609
 
706
610
    def test_get_transport_from_relpath(self):
707
611
        here = osutils.abspath('.')
708
 
        t = transport.get_transport('.')
709
 
        self.assertIsInstance(t, local.LocalTransport)
 
612
        t = get_transport('.')
 
613
        self.assertIsInstance(t, LocalTransport)
710
614
        self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
711
615
 
712
616
    def test_get_transport_from_local_url(self):
713
617
        here = osutils.abspath('.')
714
618
        here_url = urlutils.local_path_to_url(here) + '/'
715
 
        t = transport.get_transport(here_url)
716
 
        self.assertIsInstance(t, local.LocalTransport)
 
619
        t = get_transport(here_url)
 
620
        self.assertIsInstance(t, LocalTransport)
717
621
        self.assertEquals(t.base, here_url)
718
622
 
719
623
    def test_local_abspath(self):
720
624
        here = osutils.abspath('.')
721
 
        t = transport.get_transport(here)
 
625
        t = get_transport(here)
722
626
        self.assertEquals(t.local_abspath(''), here)
723
627
 
724
628
 
725
 
class TestWin32LocalTransport(tests.TestCase):
 
629
class TestWin32LocalTransport(TestCase):
726
630
 
727
631
    def test_unc_clone_to_root(self):
728
632
        # Win32 UNC path like \\HOST\path
729
633
        # clone to root should stop at least at \\HOST part
730
634
        # not on \\
731
 
        t = local.EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
 
635
        t = EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
732
636
        for i in xrange(4):
733
637
            t = t.clone('..')
734
638
        self.assertEquals(t.base, 'file://HOST/')
737
641
        self.assertEquals(t.base, 'file://HOST/')
738
642
 
739
643
 
740
 
class TestConnectedTransport(tests.TestCase):
 
644
class TestConnectedTransport(TestCase):
741
645
    """Tests for connected to remote server transports"""
742
646
 
743
647
    def test_parse_url(self):
744
 
        t = transport.ConnectedTransport(
745
 
            'http://simple.example.com/home/source')
 
648
        t = ConnectedTransport('http://simple.example.com/home/source')
746
649
        self.assertEquals(t._host, 'simple.example.com')
747
650
        self.assertEquals(t._port, None)
748
651
        self.assertEquals(t._path, '/home/source/')
753
656
 
754
657
    def test_parse_url_with_at_in_user(self):
755
658
        # Bug 228058
756
 
        t = transport.ConnectedTransport('ftp://user@host.com@www.host.com/')
 
659
        t = ConnectedTransport('ftp://user@host.com@www.host.com/')
757
660
        self.assertEquals(t._user, 'user@host.com')
758
661
 
759
662
    def test_parse_quoted_url(self):
760
 
        t = transport.ConnectedTransport(
761
 
            'http://ro%62ey:h%40t@ex%41mple.com:2222/path')
 
663
        t = ConnectedTransport('http://ro%62ey:h%40t@ex%41mple.com:2222/path')
762
664
        self.assertEquals(t._host, 'exAmple.com')
763
665
        self.assertEquals(t._port, 2222)
764
666
        self.assertEquals(t._user, 'robey')
770
672
 
771
673
    def test_parse_invalid_url(self):
772
674
        self.assertRaises(errors.InvalidURL,
773
 
                          transport.ConnectedTransport,
 
675
                          ConnectedTransport,
774
676
                          'sftp://lily.org:~janneke/public/bzr/gub')
775
677
 
776
678
    def test_relpath(self):
777
 
        t = transport.ConnectedTransport('sftp://user@host.com/abs/path')
 
679
        t = ConnectedTransport('sftp://user@host.com/abs/path')
778
680
 
779
681
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
780
682
        self.assertRaises(errors.PathNotChild, t.relpath,
786
688
        self.assertRaises(errors.PathNotChild, t.relpath,
787
689
                          'sftp://user@host.com:33/abs/path/sub')
788
690
        # Make sure it works when we don't supply a username
789
 
        t = transport.ConnectedTransport('sftp://host.com/abs/path')
 
691
        t = ConnectedTransport('sftp://host.com/abs/path')
790
692
        self.assertEquals(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
791
693
 
792
694
        # Make sure it works when parts of the path will be url encoded
793
 
        t = transport.ConnectedTransport('sftp://host.com/dev/%path')
 
695
        t = ConnectedTransport('sftp://host.com/dev/%path')
794
696
        self.assertEquals(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
795
697
 
796
698
    def test_connection_sharing_propagate_credentials(self):
797
 
        t = transport.ConnectedTransport('ftp://user@host.com/abs/path')
 
699
        t = ConnectedTransport('ftp://user@host.com/abs/path')
798
700
        self.assertEquals('user', t._user)
799
701
        self.assertEquals('host.com', t._host)
800
702
        self.assertIs(None, t._get_connection())
821
723
        self.assertIs(new_password, c._get_credentials())
822
724
 
823
725
 
824
 
class TestReusedTransports(tests.TestCase):
 
726
class TestReusedTransports(TestCase):
825
727
    """Tests for transport reuse"""
826
728
 
827
729
    def test_reuse_same_transport(self):
828
730
        possible_transports = []
829
 
        t1 = transport.get_transport('http://foo/',
830
 
                                     possible_transports=possible_transports)
 
731
        t1 = get_transport('http://foo/',
 
732
                           possible_transports=possible_transports)
831
733
        self.assertEqual([t1], possible_transports)
832
 
        t2 = transport.get_transport('http://foo/',
833
 
                                     possible_transports=[t1])
 
734
        t2 = get_transport('http://foo/', possible_transports=[t1])
834
735
        self.assertIs(t1, t2)
835
736
 
836
737
        # 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])
 
738
        t3 = get_transport('http://foo/path/')
 
739
        t4 = get_transport('http://foo/path', possible_transports=[t3])
840
740
        self.assertIs(t3, t4)
841
741
 
842
 
        t5 = transport.get_transport('http://foo/path')
843
 
        t6 = transport.get_transport('http://foo/path/',
844
 
                                     possible_transports=[t5])
 
742
        t5 = get_transport('http://foo/path')
 
743
        t6 = get_transport('http://foo/path/', possible_transports=[t5])
845
744
        self.assertIs(t5, t6)
846
745
 
847
746
    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])
 
747
        t1 = get_transport('http://foo/path')
 
748
        t2 = get_transport('http://bar/path', possible_transports=[t1])
851
749
        self.assertIsNot(t1, t2)
852
750
 
853
751
 
854
 
class TestTransportTrace(tests.TestCase):
 
752
class TestTransportTrace(TestCase):
855
753
 
856
754
    def test_get(self):
857
 
        t = transport.get_transport('trace+memory://')
858
 
        self.assertIsInstance(t, bzrlib.transport.trace.TransportTraceDecorator)
 
755
        transport = get_transport('trace+memory://')
 
756
        self.assertIsInstance(
 
757
            transport, bzrlib.transport.trace.TransportTraceDecorator)
859
758
 
860
759
    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)
 
760
        transport = get_transport('trace+memory://')
 
761
        transport2 = transport.clone('.')
 
762
        self.assertTrue(transport is not transport2)
 
763
        self.assertTrue(transport._activity is transport2._activity)
865
764
 
866
765
    # the following specific tests are for the operations that have made use of
867
766
    # logging in tests; we could test every single operation but doing that
868
767
    # still won't cause a test failure when the top level Transport API
869
768
    # changes; so there is little return doing that.
870
769
    def test_get(self):
871
 
        t = transport.get_transport('trace+memory:///')
872
 
        t.put_bytes('foo', 'barish')
873
 
        t.get('foo')
 
770
        transport = get_transport('trace+memory:///')
 
771
        transport.put_bytes('foo', 'barish')
 
772
        transport.get('foo')
874
773
        expected_result = []
875
774
        # put_bytes records the bytes, not the content to avoid memory
876
775
        # pressure.
877
776
        expected_result.append(('put_bytes', 'foo', 6, None))
878
777
        # get records the file name only.
879
778
        expected_result.append(('get', 'foo'))
880
 
        self.assertEqual(expected_result, t._activity)
 
779
        self.assertEqual(expected_result, transport._activity)
881
780
 
882
781
    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))
 
782
        transport = get_transport('trace+memory:///')
 
783
        transport.put_bytes('foo', 'barish')
 
784
        list(transport.readv('foo', [(0, 1), (3, 2)], adjust_for_latency=True,
 
785
            upper_limit=6))
887
786
        expected_result = []
888
787
        # put_bytes records the bytes, not the content to avoid memory
889
788
        # pressure.
890
789
        expected_result.append(('put_bytes', 'foo', 6, None))
891
790
        # readv records the supplied offset request
892
791
        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
 
        # This test actually causes a bzr instance to be invoked, which is very
904
 
        # expensive: it should be the only such test in the test suite.
905
 
        # A reasonable evolution for this would be to simply check inside
906
 
        # check_channel_exec_request that the command is appropriate, and then
907
 
        # satisfy requests in-process.
908
 
        self.requireFeature(features.paramiko)
909
 
        # SFTPFullAbsoluteServer has a get_url method, and doesn't
910
 
        # override the interface (doesn't change self._vendor).
911
 
        # Note that this does encryption, so can be slow.
912
 
        from bzrlib.tests import stub_sftp
913
 
 
914
 
        # Start an SSH server
915
 
        self.command_executed = []
916
 
        # XXX: This is horrible -- we define a really dumb SSH server that
917
 
        # executes commands, and manage the hooking up of stdin/out/err to the
918
 
        # SSH channel ourselves.  Surely this has already been implemented
919
 
        # elsewhere?
920
 
        started = []
921
 
        class StubSSHServer(stub_sftp.StubServer):
922
 
 
923
 
            test = self
924
 
 
925
 
            def check_channel_exec_request(self, channel, command):
926
 
                self.test.command_executed.append(command)
927
 
                proc = subprocess.Popen(
928
 
                    command, shell=True, stdin=subprocess.PIPE,
929
 
                    stdout=subprocess.PIPE, stderr=subprocess.PIPE)
930
 
 
931
 
                # XXX: horribly inefficient, not to mention ugly.
932
 
                # Start a thread for each of stdin/out/err, and relay bytes from
933
 
                # the subprocess to channel and vice versa.
934
 
                def ferry_bytes(read, write, close):
935
 
                    while True:
936
 
                        bytes = read(1)
937
 
                        if bytes == '':
938
 
                            close()
939
 
                            break
940
 
                        write(bytes)
941
 
 
942
 
                file_functions = [
943
 
                    (channel.recv, proc.stdin.write, proc.stdin.close),
944
 
                    (proc.stdout.read, channel.sendall, channel.close),
945
 
                    (proc.stderr.read, channel.sendall_stderr, channel.close)]
946
 
                started.append(proc)
947
 
                for read, write, close in file_functions:
948
 
                    t = threading.Thread(
949
 
                        target=ferry_bytes, args=(read, write, close))
950
 
                    t.start()
951
 
                    started.append(t)
952
 
 
953
 
                return True
954
 
 
955
 
        ssh_server = stub_sftp.SFTPFullAbsoluteServer(StubSSHServer)
956
 
        # We *don't* want to override the default SSH vendor: the detected one
957
 
        # is the one to use.
958
 
        self.start_server(ssh_server)
959
 
        port = ssh_server._listener.port
960
 
 
961
 
        if sys.platform == 'win32':
962
 
            bzr_remote_path = sys.executable + ' ' + self.get_bzr_path()
963
 
        else:
964
 
            bzr_remote_path = self.get_bzr_path()
965
 
        os.environ['BZR_REMOTE_PATH'] = bzr_remote_path
966
 
 
967
 
        # Access the branch via a bzr+ssh URL.  The BZR_REMOTE_PATH environment
968
 
        # variable is used to tell bzr what command to run on the remote end.
969
 
        path_to_branch = osutils.abspath('.')
970
 
        if sys.platform == 'win32':
971
 
            # On Windows, we export all drives as '/C:/, etc. So we need to
972
 
            # prefix a '/' to get the right path.
973
 
            path_to_branch = '/' + path_to_branch
974
 
        url = 'bzr+ssh://fred:secret@localhost:%d%s' % (port, path_to_branch)
975
 
        t = transport.get_transport(url)
976
 
        self.permit_url(t.base)
977
 
        t.mkdir('foo')
978
 
 
979
 
        self.assertEqual(
980
 
            ['%s serve --inet --directory=/ --allow-writes' % bzr_remote_path],
981
 
            self.command_executed)
982
 
        # Make sure to disconnect, so that the remote process can stop, and we
983
 
        # can cleanup. Then pause the test until everything is shutdown
984
 
        t._client._medium.disconnect()
985
 
        if not started:
986
 
            return
987
 
        # First wait for the subprocess
988
 
        started[0].wait()
989
 
        # And the rest are threads
990
 
        for t in started[1:]:
991
 
            t.join()
 
792
        self.assertEqual(expected_result, transport._activity)