← 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: Martin Pool
  • Date: 2009-07-10 06:46:10 UTC
  • mto: (4525.1.1 integration)
  • mto: This revision was merged to the branch mainline in revision 4526.
  • Revision ID: mbp@sourcefrog.net-20090710064610-sqviksbqp5i34sw2
Rename to per_interrepository

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/'))
380
 
        self.start_server(server)
381
 
        t = transport.get_transport(server.get_url())
382
 
        self.assertEqual(server.get_url(), t.abspath('/'))
 
365
        server = ChrootServer(get_transport('memory:///foo/bar/'))
 
366
        server.setUp()
 
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('/'))
 
372
        server.tearDown()
386
373
 
387
374
    def test_clone(self):
388
 
        server = chroot.ChrootServer(
389
 
            transport.get_transport('memory:///foo/bar/'))
390
 
        self.start_server(server)
391
 
        t = transport.get_transport(server.get_url())
 
375
        server = ChrootServer(get_transport('memory:///foo/bar/'))
 
376
        server.setUp()
 
377
        transport = get_transport(server.get_url())
392
378
        # relpath from root and root path are the same
393
 
        relpath_cloned = t.clone('foo')
394
 
        abspath_cloned = t.clone('/foo')
 
379
        relpath_cloned = transport.clone('foo')
 
380
        abspath_cloned = transport.clone('/foo')
395
381
        self.assertEqual(server, relpath_cloned.server)
396
382
        self.assertEqual(server, abspath_cloned.server)
 
383
        server.tearDown()
397
384
 
398
385
    def test_chroot_url_preserves_chroot(self):
399
386
        """Calling get_transport on a chroot transport's base should produce a
403
390
        This is so that it is not possible to escape a chroot by doing::
404
391
            url = chroot_transport.base
405
392
            parent_url = urlutils.join(url, '..')
406
 
            new_t = transport.get_transport(parent_url)
 
393
            new_transport = get_transport(parent_url)
407
394
        """
408
 
        server = chroot.ChrootServer(
409
 
            transport.get_transport('memory:///path/subpath'))
410
 
        self.start_server(server)
411
 
        t = transport.get_transport(server.get_url())
412
 
        new_t = transport.get_transport(t.base)
413
 
        self.assertEqual(t.server, new_t.server)
414
 
        self.assertEqual(t.base, new_t.base)
 
395
        server = ChrootServer(get_transport('memory:///path/subpath'))
 
396
        server.setUp()
 
397
        transport = get_transport(server.get_url())
 
398
        new_transport = get_transport(transport.base)
 
399
        self.assertEqual(transport.server, new_transport.server)
 
400
        self.assertEqual(transport.base, new_transport.base)
 
401
        server.tearDown()
415
402
 
416
403
    def test_urljoin_preserves_chroot(self):
417
404
        """Using urlutils.join(url, '..') on a chroot URL should not produce a
420
407
        This is so that it is not possible to escape a chroot by doing::
421
408
            url = chroot_transport.base
422
409
            parent_url = urlutils.join(url, '..')
423
 
            new_t = transport.get_transport(parent_url)
 
410
            new_transport = get_transport(parent_url)
424
411
        """
425
 
        server = chroot.ChrootServer(transport.get_transport('memory:///path/'))
426
 
        self.start_server(server)
427
 
        t = transport.get_transport(server.get_url())
 
412
        server = ChrootServer(get_transport('memory:///path/'))
 
413
        server.setUp()
 
414
        transport = get_transport(server.get_url())
428
415
        self.assertRaises(
429
 
            errors.InvalidURLJoin, urlutils.join, t.base, '..')
430
 
 
431
 
 
432
 
class TestChrootServer(tests.TestCase):
 
416
            InvalidURLJoin, urlutils.join, transport.base, '..')
 
417
        server.tearDown()
 
418
 
 
419
 
 
420
class ChrootServerTest(TestCase):
433
421
 
434
422
    def test_construct(self):
435
 
        backing_transport = memory.MemoryTransport()
436
 
        server = chroot.ChrootServer(backing_transport)
 
423
        backing_transport = MemoryTransport()
 
424
        server = ChrootServer(backing_transport)
437
425
        self.assertEqual(backing_transport, server.backing_transport)
438
426
 
439
427
    def test_setUp(self):
440
 
        backing_transport = memory.MemoryTransport()
441
 
        server = chroot.ChrootServer(backing_transport)
442
 
        server.start_server()
443
 
        try:
444
 
            self.assertTrue(server.scheme
445
 
                            in transport._get_protocol_handlers().keys())
446
 
        finally:
447
 
            server.stop_server()
 
428
        backing_transport = MemoryTransport()
 
429
        server = ChrootServer(backing_transport)
 
430
        server.setUp()
 
431
        self.assertTrue(server.scheme in _get_protocol_handlers().keys())
448
432
 
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())
 
433
    def test_tearDown(self):
 
434
        backing_transport = MemoryTransport()
 
435
        server = ChrootServer(backing_transport)
 
436
        server.setUp()
 
437
        server.tearDown()
 
438
        self.assertFalse(server.scheme in _get_protocol_handlers().keys())
456
439
 
457
440
    def test_get_url(self):
458
 
        backing_transport = memory.MemoryTransport()
459
 
        server = chroot.ChrootServer(backing_transport)
460
 
        server.start_server()
461
 
        try:
462
 
            self.assertEqual('chroot-%d:///' % id(server), server.get_url())
463
 
        finally:
464
 
            server.stop_server()
465
 
 
466
 
 
467
 
class PathFilteringDecoratorTransportTest(tests.TestCase):
468
 
    """Pathfilter decoration specific tests."""
469
 
 
470
 
    def test_abspath(self):
471
 
        # The abspath is always relative to the base of the backing transport.
472
 
        server = pathfilter.PathFilteringServer(
473
 
            transport.get_transport('memory:///foo/bar/'),
474
 
            lambda x: x)
475
 
        server.start_server()
476
 
        t = transport.get_transport(server.get_url())
477
 
        self.assertEqual(server.get_url(), t.abspath('/'))
478
 
 
479
 
        subdir_t = t.clone('subdir')
480
 
        self.assertEqual(server.get_url(), subdir_t.abspath('/'))
481
 
        server.stop_server()
482
 
 
483
 
    def make_pf_transport(self, filter_func=None):
484
 
        """Make a PathFilteringTransport backed by a MemoryTransport.
485
 
        
486
 
        :param filter_func: by default this will be a no-op function.  Use this
487
 
            parameter to override it."""
488
 
        if filter_func is None:
489
 
            filter_func = lambda x: x
490
 
        server = pathfilter.PathFilteringServer(
491
 
            transport.get_transport('memory:///foo/bar/'), filter_func)
492
 
        server.start_server()
493
 
        self.addCleanup(server.stop_server)
494
 
        return transport.get_transport(server.get_url())
495
 
 
496
 
    def test__filter(self):
497
 
        # _filter (with an identity func as filter_func) always returns
498
 
        # paths relative to the base of the backing transport.
499
 
        t = self.make_pf_transport()
500
 
        self.assertEqual('foo', t._filter('foo'))
501
 
        self.assertEqual('foo/bar', t._filter('foo/bar'))
502
 
        self.assertEqual('', t._filter('..'))
503
 
        self.assertEqual('', t._filter('/'))
504
 
        # The base of the pathfiltering transport is taken into account too.
505
 
        t = t.clone('subdir1/subdir2')
506
 
        self.assertEqual('subdir1/subdir2/foo', t._filter('foo'))
507
 
        self.assertEqual('subdir1/subdir2/foo/bar', t._filter('foo/bar'))
508
 
        self.assertEqual('subdir1', t._filter('..'))
509
 
        self.assertEqual('', t._filter('/'))
510
 
 
511
 
    def test_filter_invocation(self):
512
 
        filter_log = []
513
 
        def filter(path):
514
 
            filter_log.append(path)
515
 
            return path
516
 
        t = self.make_pf_transport(filter)
517
 
        t.has('abc')
518
 
        self.assertEqual(['abc'], filter_log)
519
 
        del filter_log[:]
520
 
        t.clone('abc').has('xyz')
521
 
        self.assertEqual(['abc/xyz'], filter_log)
522
 
        del filter_log[:]
523
 
        t.has('/abc')
524
 
        self.assertEqual(['abc'], filter_log)
525
 
 
526
 
    def test_clone(self):
527
 
        t = self.make_pf_transport()
528
 
        # relpath from root and root path are the same
529
 
        relpath_cloned = t.clone('foo')
530
 
        abspath_cloned = t.clone('/foo')
531
 
        self.assertEqual(t.server, relpath_cloned.server)
532
 
        self.assertEqual(t.server, abspath_cloned.server)
533
 
 
534
 
    def test_url_preserves_pathfiltering(self):
535
 
        """Calling get_transport on a pathfiltered transport's base should
536
 
        produce a transport with exactly the same behaviour as the original
537
 
        pathfiltered transport.
538
 
 
539
 
        This is so that it is not possible to escape (accidentally or
540
 
        otherwise) the filtering by doing::
541
 
            url = filtered_transport.base
542
 
            parent_url = urlutils.join(url, '..')
543
 
            new_t = transport.get_transport(parent_url)
544
 
        """
545
 
        t = self.make_pf_transport()
546
 
        new_t = transport.get_transport(t.base)
547
 
        self.assertEqual(t.server, new_t.server)
548
 
        self.assertEqual(t.base, new_t.base)
549
 
 
550
 
 
551
 
class ReadonlyDecoratorTransportTest(tests.TestCase):
 
441
        backing_transport = MemoryTransport()
 
442
        server = ChrootServer(backing_transport)
 
443
        server.setUp()
 
444
        self.assertEqual('chroot-%d:///' % id(server), server.get_url())
 
445
        server.tearDown()
 
446
 
 
447
 
 
448
class ReadonlyDecoratorTransportTest(TestCase):
552
449
    """Readonly decoration specific tests."""
553
450
 
554
451
    def test_local_parameters(self):
 
452
        import bzrlib.transport.readonly as readonly
555
453
        # connect to . in readonly mode
556
 
        t = readonly.ReadonlyTransportDecorator('readonly+.')
557
 
        self.assertEqual(True, t.listable())
558
 
        self.assertEqual(True, t.is_readonly())
 
454
        transport = readonly.ReadonlyTransportDecorator('readonly+.')
 
455
        self.assertEqual(True, transport.listable())
 
456
        self.assertEqual(True, transport.is_readonly())
559
457
 
560
458
    def test_http_parameters(self):
561
459
        from bzrlib.tests.http_server import HttpServer
 
460
        import bzrlib.transport.readonly as readonly
562
461
        # connect to '.' via http which is not listable
563
462
        server = HttpServer()
564
 
        self.start_server(server)
565
 
        t = transport.get_transport('readonly+' + server.get_url())
566
 
        self.failUnless(isinstance(t, readonly.ReadonlyTransportDecorator))
567
 
        self.assertEqual(False, t.listable())
568
 
        self.assertEqual(True, t.is_readonly())
569
 
 
570
 
 
571
 
class FakeNFSDecoratorTests(tests.TestCaseInTempDir):
 
463
        server.setUp()
 
464
        try:
 
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
        finally:
 
471
            server.tearDown()
 
472
 
 
473
 
 
474
class FakeNFSDecoratorTests(TestCaseInTempDir):
572
475
    """NFS decorator specific tests."""
573
476
 
574
477
    def get_nfs_transport(self, url):
 
478
        import bzrlib.transport.fakenfs as fakenfs
575
479
        # connect to url with nfs decoration
576
480
        return fakenfs.FakeNFSTransportDecorator('fakenfs+' + url)
577
481
 
578
482
    def test_local_parameters(self):
579
483
        # the listable and is_readonly parameters
580
484
        # 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())
 
485
        transport = self.get_nfs_transport('.')
 
486
        self.assertEqual(True, transport.listable())
 
487
        self.assertEqual(False, transport.is_readonly())
584
488
 
585
489
    def test_http_parameters(self):
586
490
        # the listable and is_readonly parameters
588
492
        from bzrlib.tests.http_server import HttpServer
589
493
        # connect to '.' via http which is not listable
590
494
        server = HttpServer()
591
 
        self.start_server(server)
592
 
        t = self.get_nfs_transport(server.get_url())
593
 
        self.assertIsInstance(t, fakenfs.FakeNFSTransportDecorator)
594
 
        self.assertEqual(False, t.listable())
595
 
        self.assertEqual(True, t.is_readonly())
 
495
        server.setUp()
 
496
        try:
 
497
            transport = self.get_nfs_transport(server.get_url())
 
498
            self.assertIsInstance(
 
499
                transport, bzrlib.transport.fakenfs.FakeNFSTransportDecorator)
 
500
            self.assertEqual(False, transport.listable())
 
501
            self.assertEqual(True, transport.is_readonly())
 
502
        finally:
 
503
            server.tearDown()
596
504
 
597
505
    def test_fakenfs_server_default(self):
598
506
        # a FakeNFSServer() should bring up a local relpath server for itself
599
 
        server = test_server.FakeNFSServer()
600
 
        self.start_server(server)
601
 
        # the url should be decorated appropriately
602
 
        self.assertStartsWith(server.get_url(), 'fakenfs+')
603
 
        # and we should be able to get a transport for it
604
 
        t = transport.get_transport(server.get_url())
605
 
        # which must be a FakeNFSTransportDecorator instance.
606
 
        self.assertIsInstance(t, fakenfs.FakeNFSTransportDecorator)
 
507
        import bzrlib.transport.fakenfs as fakenfs
 
508
        server = fakenfs.FakeNFSServer()
 
509
        server.setUp()
 
510
        try:
 
511
            # the url should be decorated appropriately
 
512
            self.assertStartsWith(server.get_url(), 'fakenfs+')
 
513
            # and we should be able to get a transport for it
 
514
            transport = get_transport(server.get_url())
 
515
            # which must be a FakeNFSTransportDecorator instance.
 
516
            self.assertIsInstance(
 
517
                transport, fakenfs.FakeNFSTransportDecorator)
 
518
        finally:
 
519
            server.tearDown()
607
520
 
608
521
    def test_fakenfs_rename_semantics(self):
609
522
        # a FakeNFS transport must mangle the way rename errors occur to
610
523
        # look like NFS problems.
611
 
        t = self.get_nfs_transport('.')
 
524
        transport = self.get_nfs_transport('.')
612
525
        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):
 
526
                        transport=transport)
 
527
        self.assertRaises(errors.ResourceBusy,
 
528
                          transport.rename, 'from', 'to')
 
529
 
 
530
 
 
531
class FakeVFATDecoratorTests(TestCaseInTempDir):
618
532
    """Tests for simulation of VFAT restrictions"""
619
533
 
620
534
    def get_vfat_transport(self, url):
624
538
 
625
539
    def test_transport_creation(self):
626
540
        from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
627
 
        t = self.get_vfat_transport('.')
628
 
        self.assertIsInstance(t, FakeVFATTransportDecorator)
 
541
        transport = self.get_vfat_transport('.')
 
542
        self.assertIsInstance(transport, FakeVFATTransportDecorator)
629
543
 
630
544
    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'))
 
545
        transport = self.get_vfat_transport('.')
 
546
        transport.mkdir('HELLO')
 
547
        self.assertTrue(transport.has('hello'))
 
548
        self.assertTrue(transport.has('Hello'))
635
549
 
636
550
    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):
 
551
        transport = self.get_vfat_transport('.')
 
552
        self.assertRaises(ValueError, transport.has, "<NU>")
 
553
 
 
554
 
 
555
class BadTransportHandler(Transport):
642
556
    def __init__(self, base_url):
643
 
        raise errors.DependencyNotPresent('some_lib',
644
 
                                          'testing missing dependency')
645
 
 
646
 
 
647
 
class BackupTransportHandler(transport.Transport):
 
557
        raise DependencyNotPresent('some_lib', 'testing missing dependency')
 
558
 
 
559
 
 
560
class BackupTransportHandler(Transport):
648
561
    """Test transport that works as a backup for the BadTransportHandler"""
649
562
    pass
650
563
 
651
564
 
652
 
class TestTransportImplementation(tests.TestCaseInTempDir):
 
565
class TestTransportImplementation(TestCaseInTempDir):
653
566
    """Implementation verification for transports.
654
567
 
655
568
    To verify a transport we need a server factory, which is a callable
674
587
    def setUp(self):
675
588
        super(TestTransportImplementation, self).setUp()
676
589
        self._server = self.transport_server()
677
 
        self.start_server(self._server)
 
590
        self._server.setUp()
 
591
        self.addCleanup(self._server.tearDown)
678
592
 
679
593
    def get_transport(self, relpath=None):
680
594
        """Return a connected transport to the local directory.
684
598
        base_url = self._server.get_url()
685
599
        url = self._adjust_url(base_url, relpath)
686
600
        # try getting the transport via the regular interface:
687
 
        t = transport.get_transport(url)
 
601
        t = get_transport(url)
688
602
        # vila--20070607 if the following are commented out the test suite
689
603
        # still pass. Is this really still needed or was it a forgotten
690
604
        # temporary fix ?
695
609
        return t
696
610
 
697
611
 
698
 
class TestLocalTransports(tests.TestCase):
 
612
class TestLocalTransports(TestCase):
699
613
 
700
614
    def test_get_transport_from_abspath(self):
701
615
        here = osutils.abspath('.')
702
 
        t = transport.get_transport(here)
703
 
        self.assertIsInstance(t, local.LocalTransport)
 
616
        t = get_transport(here)
 
617
        self.assertIsInstance(t, LocalTransport)
704
618
        self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
705
619
 
706
620
    def test_get_transport_from_relpath(self):
707
621
        here = osutils.abspath('.')
708
 
        t = transport.get_transport('.')
709
 
        self.assertIsInstance(t, local.LocalTransport)
 
622
        t = get_transport('.')
 
623
        self.assertIsInstance(t, LocalTransport)
710
624
        self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
711
625
 
712
626
    def test_get_transport_from_local_url(self):
713
627
        here = osutils.abspath('.')
714
628
        here_url = urlutils.local_path_to_url(here) + '/'
715
 
        t = transport.get_transport(here_url)
716
 
        self.assertIsInstance(t, local.LocalTransport)
 
629
        t = get_transport(here_url)
 
630
        self.assertIsInstance(t, LocalTransport)
717
631
        self.assertEquals(t.base, here_url)
718
632
 
719
633
    def test_local_abspath(self):
720
634
        here = osutils.abspath('.')
721
 
        t = transport.get_transport(here)
 
635
        t = get_transport(here)
722
636
        self.assertEquals(t.local_abspath(''), here)
723
637
 
724
638
 
725
 
class TestWin32LocalTransport(tests.TestCase):
 
639
class TestWin32LocalTransport(TestCase):
726
640
 
727
641
    def test_unc_clone_to_root(self):
728
642
        # Win32 UNC path like \\HOST\path
729
643
        # clone to root should stop at least at \\HOST part
730
644
        # not on \\
731
 
        t = local.EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
 
645
        t = EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
732
646
        for i in xrange(4):
733
647
            t = t.clone('..')
734
648
        self.assertEquals(t.base, 'file://HOST/')
737
651
        self.assertEquals(t.base, 'file://HOST/')
738
652
 
739
653
 
740
 
class TestConnectedTransport(tests.TestCase):
 
654
class TestConnectedTransport(TestCase):
741
655
    """Tests for connected to remote server transports"""
742
656
 
743
657
    def test_parse_url(self):
744
 
        t = transport.ConnectedTransport(
745
 
            'http://simple.example.com/home/source')
 
658
        t = ConnectedTransport('http://simple.example.com/home/source')
746
659
        self.assertEquals(t._host, 'simple.example.com')
747
660
        self.assertEquals(t._port, None)
748
661
        self.assertEquals(t._path, '/home/source/')
753
666
 
754
667
    def test_parse_url_with_at_in_user(self):
755
668
        # Bug 228058
756
 
        t = transport.ConnectedTransport('ftp://user@host.com@www.host.com/')
 
669
        t = ConnectedTransport('ftp://user@host.com@www.host.com/')
757
670
        self.assertEquals(t._user, 'user@host.com')
758
671
 
759
672
    def test_parse_quoted_url(self):
760
 
        t = transport.ConnectedTransport(
761
 
            'http://ro%62ey:h%40t@ex%41mple.com:2222/path')
 
673
        t = ConnectedTransport('http://ro%62ey:h%40t@ex%41mple.com:2222/path')
762
674
        self.assertEquals(t._host, 'exAmple.com')
763
675
        self.assertEquals(t._port, 2222)
764
676
        self.assertEquals(t._user, 'robey')
770
682
 
771
683
    def test_parse_invalid_url(self):
772
684
        self.assertRaises(errors.InvalidURL,
773
 
                          transport.ConnectedTransport,
 
685
                          ConnectedTransport,
774
686
                          'sftp://lily.org:~janneke/public/bzr/gub')
775
687
 
776
688
    def test_relpath(self):
777
 
        t = transport.ConnectedTransport('sftp://user@host.com/abs/path')
 
689
        t = ConnectedTransport('sftp://user@host.com/abs/path')
778
690
 
779
691
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
780
692
        self.assertRaises(errors.PathNotChild, t.relpath,
786
698
        self.assertRaises(errors.PathNotChild, t.relpath,
787
699
                          'sftp://user@host.com:33/abs/path/sub')
788
700
        # Make sure it works when we don't supply a username
789
 
        t = transport.ConnectedTransport('sftp://host.com/abs/path')
 
701
        t = ConnectedTransport('sftp://host.com/abs/path')
790
702
        self.assertEquals(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
791
703
 
792
704
        # Make sure it works when parts of the path will be url encoded
793
 
        t = transport.ConnectedTransport('sftp://host.com/dev/%path')
 
705
        t = ConnectedTransport('sftp://host.com/dev/%path')
794
706
        self.assertEquals(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
795
707
 
796
708
    def test_connection_sharing_propagate_credentials(self):
797
 
        t = transport.ConnectedTransport('ftp://user@host.com/abs/path')
 
709
        t = ConnectedTransport('ftp://user@host.com/abs/path')
798
710
        self.assertEquals('user', t._user)
799
711
        self.assertEquals('host.com', t._host)
800
712
        self.assertIs(None, t._get_connection())
821
733
        self.assertIs(new_password, c._get_credentials())
822
734
 
823
735
 
824
 
class TestReusedTransports(tests.TestCase):
 
736
class TestReusedTransports(TestCase):
825
737
    """Tests for transport reuse"""
826
738
 
827
739
    def test_reuse_same_transport(self):
828
740
        possible_transports = []
829
 
        t1 = transport.get_transport('http://foo/',
830
 
                                     possible_transports=possible_transports)
 
741
        t1 = get_transport('http://foo/',
 
742
                           possible_transports=possible_transports)
831
743
        self.assertEqual([t1], possible_transports)
832
 
        t2 = transport.get_transport('http://foo/',
833
 
                                     possible_transports=[t1])
 
744
        t2 = get_transport('http://foo/', possible_transports=[t1])
834
745
        self.assertIs(t1, t2)
835
746
 
836
747
        # 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])
 
748
        t3 = get_transport('http://foo/path/')
 
749
        t4 = get_transport('http://foo/path', possible_transports=[t3])
840
750
        self.assertIs(t3, t4)
841
751
 
842
 
        t5 = transport.get_transport('http://foo/path')
843
 
        t6 = transport.get_transport('http://foo/path/',
844
 
                                     possible_transports=[t5])
 
752
        t5 = get_transport('http://foo/path')
 
753
        t6 = get_transport('http://foo/path/', possible_transports=[t5])
845
754
        self.assertIs(t5, t6)
846
755
 
847
756
    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])
 
757
        t1 = get_transport('http://foo/path')
 
758
        t2 = get_transport('http://bar/path', possible_transports=[t1])
851
759
        self.assertIsNot(t1, t2)
852
760
 
853
761
 
854
 
class TestTransportTrace(tests.TestCase):
 
762
class TestTransportTrace(TestCase):
855
763
 
856
764
    def test_get(self):
857
 
        t = transport.get_transport('trace+memory://')
858
 
        self.assertIsInstance(t, bzrlib.transport.trace.TransportTraceDecorator)
 
765
        transport = get_transport('trace+memory://')
 
766
        self.assertIsInstance(
 
767
            transport, bzrlib.transport.trace.TransportTraceDecorator)
859
768
 
860
769
    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)
 
770
        transport = get_transport('trace+memory://')
 
771
        transport2 = transport.clone('.')
 
772
        self.assertTrue(transport is not transport2)
 
773
        self.assertTrue(transport._activity is transport2._activity)
865
774
 
866
775
    # the following specific tests are for the operations that have made use of
867
776
    # logging in tests; we could test every single operation but doing that
868
777
    # still won't cause a test failure when the top level Transport API
869
778
    # changes; so there is little return doing that.
870
779
    def test_get(self):
871
 
        t = transport.get_transport('trace+memory:///')
872
 
        t.put_bytes('foo', 'barish')
873
 
        t.get('foo')
 
780
        transport = get_transport('trace+memory:///')
 
781
        transport.put_bytes('foo', 'barish')
 
782
        transport.get('foo')
874
783
        expected_result = []
875
784
        # put_bytes records the bytes, not the content to avoid memory
876
785
        # pressure.
877
786
        expected_result.append(('put_bytes', 'foo', 6, None))
878
787
        # get records the file name only.
879
788
        expected_result.append(('get', 'foo'))
880
 
        self.assertEqual(expected_result, t._activity)
 
789
        self.assertEqual(expected_result, transport._activity)
881
790
 
882
791
    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))
 
792
        transport = get_transport('trace+memory:///')
 
793
        transport.put_bytes('foo', 'barish')
 
794
        list(transport.readv('foo', [(0, 1), (3, 2)], adjust_for_latency=True,
 
795
            upper_limit=6))
887
796
        expected_result = []
888
797
        # put_bytes records the bytes, not the content to avoid memory
889
798
        # pressure.
890
799
        expected_result.append(('put_bytes', 'foo', 6, None))
891
800
        # readv records the supplied offset request
892
801
        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
 
 
959
 
        # FIXME: I don't understand the above comment, SFTPFullAbsoluteServer
960
 
        # inherits from SFTPServer which forces the SSH vendor to
961
 
        # ssh.ParamikoVendor(). So it's forced, not detected. --vila 20100623
962
 
        self.start_server(ssh_server)
963
 
        port = ssh_server.port
964
 
 
965
 
        if sys.platform == 'win32':
966
 
            bzr_remote_path = sys.executable + ' ' + self.get_bzr_path()
967
 
        else:
968
 
            bzr_remote_path = self.get_bzr_path()
969
 
        os.environ['BZR_REMOTE_PATH'] = bzr_remote_path
970
 
 
971
 
        # Access the branch via a bzr+ssh URL.  The BZR_REMOTE_PATH environment
972
 
        # variable is used to tell bzr what command to run on the remote end.
973
 
        path_to_branch = osutils.abspath('.')
974
 
        if sys.platform == 'win32':
975
 
            # On Windows, we export all drives as '/C:/, etc. So we need to
976
 
            # prefix a '/' to get the right path.
977
 
            path_to_branch = '/' + path_to_branch
978
 
        url = 'bzr+ssh://fred:secret@localhost:%d%s' % (port, path_to_branch)
979
 
        t = transport.get_transport(url)
980
 
        self.permit_url(t.base)
981
 
        t.mkdir('foo')
982
 
 
983
 
        self.assertEqual(
984
 
            ['%s serve --inet --directory=/ --allow-writes' % bzr_remote_path],
985
 
            self.command_executed)
986
 
        # Make sure to disconnect, so that the remote process can stop, and we
987
 
        # can cleanup. Then pause the test until everything is shutdown
988
 
        t._client._medium.disconnect()
989
 
        if not started:
990
 
            return
991
 
        # First wait for the subprocess
992
 
        started[0].wait()
993
 
        # And the rest are threads
994
 
        for t in started[1:]:
995
 
            t.join()
 
802
        self.assertEqual(expected_result, transport._activity)