← 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: Canonical.com Patch Queue Manager
  • Date: 2010-09-01 08:02:42 UTC
  • mfrom: (5390.3.3 faster-revert-593560)
  • Revision ID: pqm@pqm.ubuntu.com-20100901080242-esg62ody4frwmy66
(spiv) Avoid repeatedly calling self.target.all_file_ids() in
 InterTree.iter_changes. (Andrew Bennetts)

Show diffs side-by-side

added added

removed removed

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