/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

Add a Transport.is_readonly remote call, let {Branch,Repository}.lock_write remote call return UnlockableTransport, and miscellaneous test fixes.

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