← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

(jelmer) Use the absolute_import feature everywhere in bzrlib,
 and add a source test to make sure it's used everywhere. (Jelmer Vernooij)

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-2011 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 errno
19
20
import os
20
21
import subprocess
21
22
import sys
22
23
import threading
23
24
 
24
 
import bzrlib
25
25
from bzrlib import (
26
26
    errors,
27
27
    osutils,
28
28
    tests,
 
29
    transport,
29
30
    urlutils,
30
31
    )
31
 
from bzrlib.errors import (DependencyNotPresent,
32
 
                           FileExists,
33
 
                           InvalidURLJoin,
34
 
                           NoSuchFile,
35
 
                           PathNotChild,
36
 
                           ReadError,
37
 
                           UnsupportedProtocol,
38
 
                           )
39
 
from bzrlib.tests import ParamikoFeature, TestCase, TestCaseInTempDir
40
 
from bzrlib.transport import (_clear_protocol_handlers,
41
 
                              _CoalescedOffset,
42
 
                              ConnectedTransport,
43
 
                              _get_protocol_handlers,
44
 
                              _set_protocol_handlers,
45
 
                              _get_transport_modules,
46
 
                              get_transport,
47
 
                              LateReadError,
48
 
                              register_lazy_transport,
49
 
                              register_transport_proto,
50
 
                              Transport,
51
 
                              )
52
 
from bzrlib.transport.chroot import ChrootServer
53
 
from bzrlib.transport.memory import MemoryTransport
54
 
from bzrlib.transport.local import (LocalTransport,
55
 
                                    EmulatedWin32LocalTransport)
56
 
from bzrlib.transport.pathfilter import PathFilteringServer
 
32
from bzrlib.directory_service import directories
 
33
from bzrlib.transport import (
 
34
    chroot,
 
35
    fakenfs,
 
36
    http,
 
37
    local,
 
38
    location_to_url,
 
39
    memory,
 
40
    pathfilter,
 
41
    readonly,
 
42
    )
 
43
import bzrlib.transport.trace
 
44
from bzrlib.tests import (
 
45
    features,
 
46
    test_server,
 
47
    )
57
48
 
58
49
 
59
50
# TODO: Should possibly split transport-specific tests into their own files.
60
51
 
61
52
 
62
 
class TestTransport(TestCase):
 
53
class TestTransport(tests.TestCase):
63
54
    """Test the non transport-concrete class functionality."""
64
55
 
65
56
    def test__get_set_protocol_handlers(self):
66
 
        handlers = _get_protocol_handlers()
67
 
        self.assertNotEqual([], handlers.keys( ))
68
 
        try:
69
 
            _clear_protocol_handlers()
70
 
            self.assertEqual([], _get_protocol_handlers().keys())
71
 
        finally:
72
 
            _set_protocol_handlers(handlers)
 
57
        handlers = transport._get_protocol_handlers()
 
58
        self.assertNotEqual([], handlers.keys())
 
59
        transport._clear_protocol_handlers()
 
60
        self.addCleanup(transport._set_protocol_handlers, handlers)
 
61
        self.assertEqual([], transport._get_protocol_handlers().keys())
73
62
 
74
63
    def test_get_transport_modules(self):
75
 
        handlers = _get_protocol_handlers()
 
64
        handlers = transport._get_protocol_handlers()
 
65
        self.addCleanup(transport._set_protocol_handlers, handlers)
76
66
        # don't pollute the current handlers
77
 
        _clear_protocol_handlers()
 
67
        transport._clear_protocol_handlers()
 
68
 
78
69
        class SampleHandler(object):
79
70
            """I exist, isnt that enough?"""
80
 
        try:
81
 
            _clear_protocol_handlers()
82
 
            register_transport_proto('foo')
83
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
84
 
                                    'TestTransport.SampleHandler')
85
 
            register_transport_proto('bar')
86
 
            register_lazy_transport('bar', 'bzrlib.tests.test_transport',
87
 
                                    'TestTransport.SampleHandler')
88
 
            self.assertEqual([SampleHandler.__module__,
89
 
                              'bzrlib.transport.chroot',
90
 
                              'bzrlib.transport.pathfilter'],
91
 
                             _get_transport_modules())
92
 
        finally:
93
 
            _set_protocol_handlers(handlers)
 
71
        transport._clear_protocol_handlers()
 
72
        transport.register_transport_proto('foo')
 
73
        transport.register_lazy_transport('foo',
 
74
                                            'bzrlib.tests.test_transport',
 
75
                                            'TestTransport.SampleHandler')
 
76
        transport.register_transport_proto('bar')
 
77
        transport.register_lazy_transport('bar',
 
78
                                            'bzrlib.tests.test_transport',
 
79
                                            'TestTransport.SampleHandler')
 
80
        self.assertEqual([SampleHandler.__module__,
 
81
                            'bzrlib.transport.chroot',
 
82
                            'bzrlib.transport.pathfilter'],
 
83
                            transport._get_transport_modules())
94
84
 
95
85
    def test_transport_dependency(self):
96
86
        """Transport with missing dependency causes no error"""
97
 
        saved_handlers = _get_protocol_handlers()
 
87
        saved_handlers = transport._get_protocol_handlers()
 
88
        self.addCleanup(transport._set_protocol_handlers, saved_handlers)
98
89
        # don't pollute the current handlers
99
 
        _clear_protocol_handlers()
 
90
        transport._clear_protocol_handlers()
 
91
        transport.register_transport_proto('foo')
 
92
        transport.register_lazy_transport(
 
93
            'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
100
94
        try:
101
 
            register_transport_proto('foo')
102
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
103
 
                    'BadTransportHandler')
104
 
            try:
105
 
                get_transport('foo://fooserver/foo')
106
 
            except UnsupportedProtocol, e:
107
 
                e_str = str(e)
108
 
                self.assertEquals('Unsupported protocol'
109
 
                                  ' for url "foo://fooserver/foo":'
110
 
                                  ' Unable to import library "some_lib":'
111
 
                                  ' testing missing dependency', str(e))
112
 
            else:
113
 
                self.fail('Did not raise UnsupportedProtocol')
114
 
        finally:
115
 
            # restore original values
116
 
            _set_protocol_handlers(saved_handlers)
 
95
            transport.get_transport_from_url('foo://fooserver/foo')
 
96
        except errors.UnsupportedProtocol, e:
 
97
            e_str = str(e)
 
98
            self.assertEquals('Unsupported protocol'
 
99
                                ' for url "foo://fooserver/foo":'
 
100
                                ' Unable to import library "some_lib":'
 
101
                                ' testing missing dependency', str(e))
 
102
        else:
 
103
            self.fail('Did not raise UnsupportedProtocol')
117
104
 
118
105
    def test_transport_fallback(self):
119
106
        """Transport with missing dependency causes no error"""
120
 
        saved_handlers = _get_protocol_handlers()
121
 
        try:
122
 
            _clear_protocol_handlers()
123
 
            register_transport_proto('foo')
124
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
125
 
                    'BackupTransportHandler')
126
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
127
 
                    'BadTransportHandler')
128
 
            t = get_transport('foo://fooserver/foo')
129
 
            self.assertTrue(isinstance(t, BackupTransportHandler))
130
 
        finally:
131
 
            _set_protocol_handlers(saved_handlers)
 
107
        saved_handlers = transport._get_protocol_handlers()
 
108
        self.addCleanup(transport._set_protocol_handlers, saved_handlers)
 
109
        transport._clear_protocol_handlers()
 
110
        transport.register_transport_proto('foo')
 
111
        transport.register_lazy_transport(
 
112
            'foo', 'bzrlib.tests.test_transport', 'BackupTransportHandler')
 
113
        transport.register_lazy_transport(
 
114
            'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
 
115
        t = transport.get_transport_from_url('foo://fooserver/foo')
 
116
        self.assertTrue(isinstance(t, BackupTransportHandler))
132
117
 
133
118
    def test_ssh_hints(self):
134
119
        """Transport ssh:// should raise an error pointing out bzr+ssh://"""
135
120
        try:
136
 
            get_transport('ssh://fooserver/foo')
137
 
        except UnsupportedProtocol, e:
 
121
            transport.get_transport_from_url('ssh://fooserver/foo')
 
122
        except errors.UnsupportedProtocol, e:
138
123
            e_str = str(e)
139
124
            self.assertEquals('Unsupported protocol'
140
125
                              ' for url "ssh://fooserver/foo":'
141
 
                              ' bzr supports bzr+ssh to operate over ssh, use "bzr+ssh://fooserver/foo".',
 
126
                              ' bzr supports bzr+ssh to operate over ssh,'
 
127
                              ' use "bzr+ssh://fooserver/foo".',
142
128
                              str(e))
143
129
        else:
144
130
            self.fail('Did not raise UnsupportedProtocol')
145
131
 
146
132
    def test_LateReadError(self):
147
133
        """The LateReadError helper should raise on read()."""
148
 
        a_file = LateReadError('a path')
 
134
        a_file = transport.LateReadError('a path')
149
135
        try:
150
136
            a_file.read()
151
 
        except ReadError, error:
 
137
        except errors.ReadError, error:
152
138
            self.assertEqual('a path', error.path)
153
 
        self.assertRaises(ReadError, a_file.read, 40)
 
139
        self.assertRaises(errors.ReadError, a_file.read, 40)
154
140
        a_file.close()
155
141
 
156
 
    def test__combine_paths(self):
157
 
        t = Transport('/')
158
 
        self.assertEqual('/home/sarah/project/foo',
159
 
                         t._combine_paths('/home/sarah', 'project/foo'))
160
 
        self.assertEqual('/etc',
161
 
                         t._combine_paths('/home/sarah', '../../etc'))
162
 
        self.assertEqual('/etc',
163
 
                         t._combine_paths('/home/sarah', '../../../etc'))
164
 
        self.assertEqual('/etc',
165
 
                         t._combine_paths('/home/sarah', '/etc'))
166
 
 
167
142
    def test_local_abspath_non_local_transport(self):
168
143
        # the base implementation should throw
169
 
        t = MemoryTransport()
 
144
        t = memory.MemoryTransport()
170
145
        e = self.assertRaises(errors.NotLocalUrl, t.local_abspath, 't')
171
146
        self.assertEqual('memory:///t is not a local path.', str(e))
172
147
 
173
148
 
174
 
class TestCoalesceOffsets(TestCase):
 
149
class TestCoalesceOffsets(tests.TestCase):
175
150
 
176
151
    def check(self, expected, offsets, limit=0, max_size=0, fudge=0):
177
 
        coalesce = Transport._coalesce_offsets
178
 
        exp = [_CoalescedOffset(*x) for x in expected]
 
152
        coalesce = transport.Transport._coalesce_offsets
 
153
        exp = [transport._CoalescedOffset(*x) for x in expected]
179
154
        out = list(coalesce(offsets, limit=limit, fudge_factor=fudge,
180
155
                            max_size=max_size))
181
156
        self.assertEqual(exp, out)
226
201
 
227
202
    def test_coalesce_fudge(self):
228
203
        self.check([(10, 30, [(0, 10), (20, 10)]),
229
 
                    (100, 10, [(0, 10),]),
 
204
                    (100, 10, [(0, 10)]),
230
205
                   ], [(10, 10), (30, 10), (100, 10)],
231
 
                   fudge=10
232
 
                  )
 
206
                   fudge=10)
 
207
 
233
208
    def test_coalesce_max_size(self):
234
209
        self.check([(10, 20, [(0, 10), (10, 10)]),
235
210
                    (30, 50, [(0, 50)]),
236
211
                    # If one range is above max_size, it gets its own coalesced
237
212
                    # offset
238
 
                    (100, 80, [(0, 80),]),],
 
213
                    (100, 80, [(0, 80)]),],
239
214
                   [(10, 10), (20, 10), (30, 50), (100, 80)],
240
 
                   max_size=50
241
 
                  )
 
215
                   max_size=50)
242
216
 
243
217
    def test_coalesce_no_max_size(self):
244
 
        self.check([(10, 170, [(0, 10), (10, 10), (20, 50), (70, 100)]),],
 
218
        self.check([(10, 170, [(0, 10), (10, 10), (20, 50), (70, 100)])],
245
219
                   [(10, 10), (20, 10), (30, 50), (80, 100)],
246
220
                  )
247
221
 
248
222
    def test_coalesce_default_limit(self):
249
223
        # By default we use a 100MB max size.
250
 
        ten_mb = 10*1024*1024
251
 
        self.check([(0, 10*ten_mb, [(i*ten_mb, ten_mb) for i in range(10)]),
 
224
        ten_mb = 10 * 1024 * 1024
 
225
        self.check([(0, 10 * ten_mb, [(i * ten_mb, ten_mb) for i in range(10)]),
252
226
                    (10*ten_mb, ten_mb, [(0, ten_mb)])],
253
227
                   [(i*ten_mb, ten_mb) for i in range(11)])
254
 
        self.check([(0, 11*ten_mb, [(i*ten_mb, ten_mb) for i in range(11)]),],
255
 
                   [(i*ten_mb, ten_mb) for i in range(11)],
 
228
        self.check([(0, 11 * ten_mb, [(i * ten_mb, ten_mb) for i in range(11)])],
 
229
                   [(i * ten_mb, ten_mb) for i in range(11)],
256
230
                   max_size=1*1024*1024*1024)
257
231
 
258
232
 
259
 
class TestMemoryTransport(TestCase):
 
233
class TestMemoryServer(tests.TestCase):
 
234
 
 
235
    def test_create_server(self):
 
236
        server = memory.MemoryServer()
 
237
        server.start_server()
 
238
        url = server.get_url()
 
239
        self.assertTrue(url in transport.transport_list_registry)
 
240
        t = transport.get_transport_from_url(url)
 
241
        del t
 
242
        server.stop_server()
 
243
        self.assertFalse(url in transport.transport_list_registry)
 
244
        self.assertRaises(errors.UnsupportedProtocol,
 
245
                          transport.get_transport, url)
 
246
 
 
247
 
 
248
class TestMemoryTransport(tests.TestCase):
260
249
 
261
250
    def test_get_transport(self):
262
 
        MemoryTransport()
 
251
        memory.MemoryTransport()
263
252
 
264
253
    def test_clone(self):
265
 
        transport = MemoryTransport()
266
 
        self.assertTrue(isinstance(transport, MemoryTransport))
267
 
        self.assertEqual("memory:///", transport.clone("/").base)
 
254
        t = memory.MemoryTransport()
 
255
        self.assertTrue(isinstance(t, memory.MemoryTransport))
 
256
        self.assertEqual("memory:///", t.clone("/").base)
268
257
 
269
258
    def test_abspath(self):
270
 
        transport = MemoryTransport()
271
 
        self.assertEqual("memory:///relpath", transport.abspath('relpath'))
 
259
        t = memory.MemoryTransport()
 
260
        self.assertEqual("memory:///relpath", t.abspath('relpath'))
272
261
 
273
262
    def test_abspath_of_root(self):
274
 
        transport = MemoryTransport()
275
 
        self.assertEqual("memory:///", transport.base)
276
 
        self.assertEqual("memory:///", transport.abspath('/'))
 
263
        t = memory.MemoryTransport()
 
264
        self.assertEqual("memory:///", t.base)
 
265
        self.assertEqual("memory:///", t.abspath('/'))
277
266
 
278
267
    def test_abspath_of_relpath_starting_at_root(self):
279
 
        transport = MemoryTransport()
280
 
        self.assertEqual("memory:///foo", transport.abspath('/foo'))
 
268
        t = memory.MemoryTransport()
 
269
        self.assertEqual("memory:///foo", t.abspath('/foo'))
281
270
 
282
271
    def test_append_and_get(self):
283
 
        transport = MemoryTransport()
284
 
        transport.append_bytes('path', 'content')
285
 
        self.assertEqual(transport.get('path').read(), 'content')
286
 
        transport.append_file('path', StringIO('content'))
287
 
        self.assertEqual(transport.get('path').read(), 'contentcontent')
 
272
        t = memory.MemoryTransport()
 
273
        t.append_bytes('path', 'content')
 
274
        self.assertEqual(t.get('path').read(), 'content')
 
275
        t.append_file('path', StringIO('content'))
 
276
        self.assertEqual(t.get('path').read(), 'contentcontent')
288
277
 
289
278
    def test_put_and_get(self):
290
 
        transport = MemoryTransport()
291
 
        transport.put_file('path', StringIO('content'))
292
 
        self.assertEqual(transport.get('path').read(), 'content')
293
 
        transport.put_bytes('path', 'content')
294
 
        self.assertEqual(transport.get('path').read(), 'content')
 
279
        t = memory.MemoryTransport()
 
280
        t.put_file('path', StringIO('content'))
 
281
        self.assertEqual(t.get('path').read(), 'content')
 
282
        t.put_bytes('path', 'content')
 
283
        self.assertEqual(t.get('path').read(), 'content')
295
284
 
296
285
    def test_append_without_dir_fails(self):
297
 
        transport = MemoryTransport()
298
 
        self.assertRaises(NoSuchFile,
299
 
                          transport.append_bytes, 'dir/path', 'content')
 
286
        t = memory.MemoryTransport()
 
287
        self.assertRaises(errors.NoSuchFile,
 
288
                          t.append_bytes, 'dir/path', 'content')
300
289
 
301
290
    def test_put_without_dir_fails(self):
302
 
        transport = MemoryTransport()
303
 
        self.assertRaises(NoSuchFile,
304
 
                          transport.put_file, 'dir/path', StringIO('content'))
 
291
        t = memory.MemoryTransport()
 
292
        self.assertRaises(errors.NoSuchFile,
 
293
                          t.put_file, 'dir/path', StringIO('content'))
305
294
 
306
295
    def test_get_missing(self):
307
 
        transport = MemoryTransport()
308
 
        self.assertRaises(NoSuchFile, transport.get, 'foo')
 
296
        transport = memory.MemoryTransport()
 
297
        self.assertRaises(errors.NoSuchFile, transport.get, 'foo')
309
298
 
310
299
    def test_has_missing(self):
311
 
        transport = MemoryTransport()
312
 
        self.assertEquals(False, transport.has('foo'))
 
300
        t = memory.MemoryTransport()
 
301
        self.assertEquals(False, t.has('foo'))
313
302
 
314
303
    def test_has_present(self):
315
 
        transport = MemoryTransport()
316
 
        transport.append_bytes('foo', 'content')
317
 
        self.assertEquals(True, transport.has('foo'))
 
304
        t = memory.MemoryTransport()
 
305
        t.append_bytes('foo', 'content')
 
306
        self.assertEquals(True, t.has('foo'))
318
307
 
319
308
    def test_list_dir(self):
320
 
        transport = MemoryTransport()
321
 
        transport.put_bytes('foo', 'content')
322
 
        transport.mkdir('dir')
323
 
        transport.put_bytes('dir/subfoo', 'content')
324
 
        transport.put_bytes('dirlike', 'content')
 
309
        t = memory.MemoryTransport()
 
310
        t.put_bytes('foo', 'content')
 
311
        t.mkdir('dir')
 
312
        t.put_bytes('dir/subfoo', 'content')
 
313
        t.put_bytes('dirlike', 'content')
325
314
 
326
 
        self.assertEquals(['dir', 'dirlike', 'foo'], sorted(transport.list_dir('.')))
327
 
        self.assertEquals(['subfoo'], sorted(transport.list_dir('dir')))
 
315
        self.assertEquals(['dir', 'dirlike', 'foo'], sorted(t.list_dir('.')))
 
316
        self.assertEquals(['subfoo'], sorted(t.list_dir('dir')))
328
317
 
329
318
    def test_mkdir(self):
330
 
        transport = MemoryTransport()
331
 
        transport.mkdir('dir')
332
 
        transport.append_bytes('dir/path', 'content')
333
 
        self.assertEqual(transport.get('dir/path').read(), 'content')
 
319
        t = memory.MemoryTransport()
 
320
        t.mkdir('dir')
 
321
        t.append_bytes('dir/path', 'content')
 
322
        self.assertEqual(t.get('dir/path').read(), 'content')
334
323
 
335
324
    def test_mkdir_missing_parent(self):
336
 
        transport = MemoryTransport()
337
 
        self.assertRaises(NoSuchFile,
338
 
                          transport.mkdir, 'dir/dir')
 
325
        t = memory.MemoryTransport()
 
326
        self.assertRaises(errors.NoSuchFile, t.mkdir, 'dir/dir')
339
327
 
340
328
    def test_mkdir_twice(self):
341
 
        transport = MemoryTransport()
342
 
        transport.mkdir('dir')
343
 
        self.assertRaises(FileExists, transport.mkdir, 'dir')
 
329
        t = memory.MemoryTransport()
 
330
        t.mkdir('dir')
 
331
        self.assertRaises(errors.FileExists, t.mkdir, 'dir')
344
332
 
345
333
    def test_parameters(self):
346
 
        transport = MemoryTransport()
347
 
        self.assertEqual(True, transport.listable())
348
 
        self.assertEqual(False, transport.is_readonly())
 
334
        t = memory.MemoryTransport()
 
335
        self.assertEqual(True, t.listable())
 
336
        self.assertEqual(False, t.is_readonly())
349
337
 
350
338
    def test_iter_files_recursive(self):
351
 
        transport = MemoryTransport()
352
 
        transport.mkdir('dir')
353
 
        transport.put_bytes('dir/foo', 'content')
354
 
        transport.put_bytes('dir/bar', 'content')
355
 
        transport.put_bytes('bar', 'content')
356
 
        paths = set(transport.iter_files_recursive())
 
339
        t = memory.MemoryTransport()
 
340
        t.mkdir('dir')
 
341
        t.put_bytes('dir/foo', 'content')
 
342
        t.put_bytes('dir/bar', 'content')
 
343
        t.put_bytes('bar', 'content')
 
344
        paths = set(t.iter_files_recursive())
357
345
        self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
358
346
 
359
347
    def test_stat(self):
360
 
        transport = MemoryTransport()
361
 
        transport.put_bytes('foo', 'content')
362
 
        transport.put_bytes('bar', 'phowar')
363
 
        self.assertEqual(7, transport.stat('foo').st_size)
364
 
        self.assertEqual(6, transport.stat('bar').st_size)
365
 
 
366
 
 
367
 
class ChrootDecoratorTransportTest(TestCase):
 
348
        t = memory.MemoryTransport()
 
349
        t.put_bytes('foo', 'content')
 
350
        t.put_bytes('bar', 'phowar')
 
351
        self.assertEqual(7, t.stat('foo').st_size)
 
352
        self.assertEqual(6, t.stat('bar').st_size)
 
353
 
 
354
 
 
355
class ChrootDecoratorTransportTest(tests.TestCase):
368
356
    """Chroot decoration specific tests."""
369
357
 
370
358
    def test_abspath(self):
371
359
        # The abspath is always relative to the chroot_url.
372
 
        server = ChrootServer(get_transport('memory:///foo/bar/'))
 
360
        server = chroot.ChrootServer(
 
361
            transport.get_transport_from_url('memory:///foo/bar/'))
373
362
        self.start_server(server)
374
 
        transport = get_transport(server.get_url())
375
 
        self.assertEqual(server.get_url(), transport.abspath('/'))
 
363
        t = transport.get_transport_from_url(server.get_url())
 
364
        self.assertEqual(server.get_url(), t.abspath('/'))
376
365
 
377
 
        subdir_transport = transport.clone('subdir')
378
 
        self.assertEqual(server.get_url(), subdir_transport.abspath('/'))
 
366
        subdir_t = t.clone('subdir')
 
367
        self.assertEqual(server.get_url(), subdir_t.abspath('/'))
379
368
 
380
369
    def test_clone(self):
381
 
        server = ChrootServer(get_transport('memory:///foo/bar/'))
 
370
        server = chroot.ChrootServer(
 
371
            transport.get_transport_from_url('memory:///foo/bar/'))
382
372
        self.start_server(server)
383
 
        transport = get_transport(server.get_url())
 
373
        t = transport.get_transport_from_url(server.get_url())
384
374
        # relpath from root and root path are the same
385
 
        relpath_cloned = transport.clone('foo')
386
 
        abspath_cloned = transport.clone('/foo')
 
375
        relpath_cloned = t.clone('foo')
 
376
        abspath_cloned = t.clone('/foo')
387
377
        self.assertEqual(server, relpath_cloned.server)
388
378
        self.assertEqual(server, abspath_cloned.server)
389
379
 
395
385
        This is so that it is not possible to escape a chroot by doing::
396
386
            url = chroot_transport.base
397
387
            parent_url = urlutils.join(url, '..')
398
 
            new_transport = get_transport(parent_url)
 
388
            new_t = transport.get_transport_from_url(parent_url)
399
389
        """
400
 
        server = ChrootServer(get_transport('memory:///path/subpath'))
 
390
        server = chroot.ChrootServer(
 
391
            transport.get_transport_from_url('memory:///path/subpath'))
401
392
        self.start_server(server)
402
 
        transport = get_transport(server.get_url())
403
 
        new_transport = get_transport(transport.base)
404
 
        self.assertEqual(transport.server, new_transport.server)
405
 
        self.assertEqual(transport.base, new_transport.base)
 
393
        t = transport.get_transport_from_url(server.get_url())
 
394
        new_t = transport.get_transport_from_url(t.base)
 
395
        self.assertEqual(t.server, new_t.server)
 
396
        self.assertEqual(t.base, new_t.base)
406
397
 
407
398
    def test_urljoin_preserves_chroot(self):
408
399
        """Using urlutils.join(url, '..') on a chroot URL should not produce a
411
402
        This is so that it is not possible to escape a chroot by doing::
412
403
            url = chroot_transport.base
413
404
            parent_url = urlutils.join(url, '..')
414
 
            new_transport = get_transport(parent_url)
 
405
            new_t = transport.get_transport_from_url(parent_url)
415
406
        """
416
 
        server = ChrootServer(get_transport('memory:///path/'))
 
407
        server = chroot.ChrootServer(
 
408
            transport.get_transport_from_url('memory:///path/'))
417
409
        self.start_server(server)
418
 
        transport = get_transport(server.get_url())
 
410
        t = transport.get_transport_from_url(server.get_url())
419
411
        self.assertRaises(
420
 
            InvalidURLJoin, urlutils.join, transport.base, '..')
421
 
 
422
 
 
423
 
class ChrootServerTest(TestCase):
 
412
            errors.InvalidURLJoin, urlutils.join, t.base, '..')
 
413
 
 
414
 
 
415
class TestChrootServer(tests.TestCase):
424
416
 
425
417
    def test_construct(self):
426
 
        backing_transport = MemoryTransport()
427
 
        server = ChrootServer(backing_transport)
 
418
        backing_transport = memory.MemoryTransport()
 
419
        server = chroot.ChrootServer(backing_transport)
428
420
        self.assertEqual(backing_transport, server.backing_transport)
429
421
 
430
422
    def test_setUp(self):
431
 
        backing_transport = MemoryTransport()
432
 
        server = ChrootServer(backing_transport)
433
 
        server.setUp()
434
 
        try:
435
 
            self.assertTrue(server.scheme in _get_protocol_handlers().keys())
436
 
        finally:
437
 
            server.tearDown()
 
423
        backing_transport = memory.MemoryTransport()
 
424
        server = chroot.ChrootServer(backing_transport)
 
425
        server.start_server()
 
426
        self.addCleanup(server.stop_server)
 
427
        self.assertTrue(server.scheme
 
428
                        in transport._get_protocol_handlers().keys())
438
429
 
439
 
    def test_tearDown(self):
440
 
        backing_transport = MemoryTransport()
441
 
        server = ChrootServer(backing_transport)
442
 
        server.setUp()
443
 
        server.tearDown()
444
 
        self.assertFalse(server.scheme in _get_protocol_handlers().keys())
 
430
    def test_stop_server(self):
 
431
        backing_transport = memory.MemoryTransport()
 
432
        server = chroot.ChrootServer(backing_transport)
 
433
        server.start_server()
 
434
        server.stop_server()
 
435
        self.assertFalse(server.scheme
 
436
                         in transport._get_protocol_handlers().keys())
445
437
 
446
438
    def test_get_url(self):
447
 
        backing_transport = MemoryTransport()
448
 
        server = ChrootServer(backing_transport)
449
 
        server.setUp()
450
 
        try:
451
 
            self.assertEqual('chroot-%d:///' % id(server), server.get_url())
452
 
        finally:
453
 
            server.tearDown()
454
 
 
455
 
 
456
 
class PathFilteringDecoratorTransportTest(TestCase):
 
439
        backing_transport = memory.MemoryTransport()
 
440
        server = chroot.ChrootServer(backing_transport)
 
441
        server.start_server()
 
442
        self.addCleanup(server.stop_server)
 
443
        self.assertEqual('chroot-%d:///' % id(server), server.get_url())
 
444
 
 
445
 
 
446
class PathFilteringDecoratorTransportTest(tests.TestCase):
457
447
    """Pathfilter decoration specific tests."""
458
448
 
459
449
    def test_abspath(self):
460
450
        # The abspath is always relative to the base of the backing transport.
461
 
        server = PathFilteringServer(get_transport('memory:///foo/bar/'),
 
451
        server = pathfilter.PathFilteringServer(
 
452
            transport.get_transport_from_url('memory:///foo/bar/'),
462
453
            lambda x: x)
463
 
        server.setUp()
464
 
        transport = get_transport(server.get_url())
465
 
        self.assertEqual(server.get_url(), transport.abspath('/'))
 
454
        server.start_server()
 
455
        t = transport.get_transport_from_url(server.get_url())
 
456
        self.assertEqual(server.get_url(), t.abspath('/'))
466
457
 
467
 
        subdir_transport = transport.clone('subdir')
468
 
        self.assertEqual(server.get_url(), subdir_transport.abspath('/'))
469
 
        server.tearDown()
 
458
        subdir_t = t.clone('subdir')
 
459
        self.assertEqual(server.get_url(), subdir_t.abspath('/'))
 
460
        server.stop_server()
470
461
 
471
462
    def make_pf_transport(self, filter_func=None):
472
463
        """Make a PathFilteringTransport backed by a MemoryTransport.
473
 
        
 
464
 
474
465
        :param filter_func: by default this will be a no-op function.  Use this
475
466
            parameter to override it."""
476
467
        if filter_func is None:
477
468
            filter_func = lambda x: x
478
 
        server = PathFilteringServer(
479
 
            get_transport('memory:///foo/bar/'), filter_func)
480
 
        server.setUp()
481
 
        self.addCleanup(server.tearDown)
482
 
        return get_transport(server.get_url())
 
469
        server = pathfilter.PathFilteringServer(
 
470
            transport.get_transport_from_url('memory:///foo/bar/'), filter_func)
 
471
        server.start_server()
 
472
        self.addCleanup(server.stop_server)
 
473
        return transport.get_transport_from_url(server.get_url())
483
474
 
484
475
    def test__filter(self):
485
476
        # _filter (with an identity func as filter_func) always returns
486
477
        # paths relative to the base of the backing transport.
487
 
        transport = self.make_pf_transport()
488
 
        self.assertEqual('foo', transport._filter('foo'))
489
 
        self.assertEqual('foo/bar', transport._filter('foo/bar'))
490
 
        self.assertEqual('', transport._filter('..'))
491
 
        self.assertEqual('', transport._filter('/'))
 
478
        t = self.make_pf_transport()
 
479
        self.assertEqual('foo', t._filter('foo'))
 
480
        self.assertEqual('foo/bar', t._filter('foo/bar'))
 
481
        self.assertEqual('', t._filter('..'))
 
482
        self.assertEqual('', t._filter('/'))
492
483
        # The base of the pathfiltering transport is taken into account too.
493
 
        transport = transport.clone('subdir1/subdir2')
494
 
        self.assertEqual('subdir1/subdir2/foo', transport._filter('foo'))
495
 
        self.assertEqual(
496
 
            'subdir1/subdir2/foo/bar', transport._filter('foo/bar'))
497
 
        self.assertEqual('subdir1', transport._filter('..'))
498
 
        self.assertEqual('', transport._filter('/'))
 
484
        t = t.clone('subdir1/subdir2')
 
485
        self.assertEqual('subdir1/subdir2/foo', t._filter('foo'))
 
486
        self.assertEqual('subdir1/subdir2/foo/bar', t._filter('foo/bar'))
 
487
        self.assertEqual('subdir1', t._filter('..'))
 
488
        self.assertEqual('', t._filter('/'))
499
489
 
500
490
    def test_filter_invocation(self):
501
491
        filter_log = []
 
492
 
502
493
        def filter(path):
503
494
            filter_log.append(path)
504
495
            return path
505
 
        transport = self.make_pf_transport(filter)
506
 
        transport.has('abc')
 
496
        t = self.make_pf_transport(filter)
 
497
        t.has('abc')
507
498
        self.assertEqual(['abc'], filter_log)
508
499
        del filter_log[:]
509
 
        transport.clone('abc').has('xyz')
 
500
        t.clone('abc').has('xyz')
510
501
        self.assertEqual(['abc/xyz'], filter_log)
511
502
        del filter_log[:]
512
 
        transport.has('/abc')
 
503
        t.has('/abc')
513
504
        self.assertEqual(['abc'], filter_log)
514
505
 
515
506
    def test_clone(self):
516
 
        transport = self.make_pf_transport()
 
507
        t = self.make_pf_transport()
517
508
        # relpath from root and root path are the same
518
 
        relpath_cloned = transport.clone('foo')
519
 
        abspath_cloned = transport.clone('/foo')
520
 
        self.assertEqual(transport.server, relpath_cloned.server)
521
 
        self.assertEqual(transport.server, abspath_cloned.server)
 
509
        relpath_cloned = t.clone('foo')
 
510
        abspath_cloned = t.clone('/foo')
 
511
        self.assertEqual(t.server, relpath_cloned.server)
 
512
        self.assertEqual(t.server, abspath_cloned.server)
522
513
 
523
514
    def test_url_preserves_pathfiltering(self):
524
515
        """Calling get_transport on a pathfiltered transport's base should
529
520
        otherwise) the filtering by doing::
530
521
            url = filtered_transport.base
531
522
            parent_url = urlutils.join(url, '..')
532
 
            new_transport = get_transport(parent_url)
 
523
            new_t = transport.get_transport_from_url(parent_url)
533
524
        """
534
 
        transport = self.make_pf_transport()
535
 
        new_transport = get_transport(transport.base)
536
 
        self.assertEqual(transport.server, new_transport.server)
537
 
        self.assertEqual(transport.base, new_transport.base)
538
 
 
539
 
 
540
 
class ReadonlyDecoratorTransportTest(TestCase):
 
525
        t = self.make_pf_transport()
 
526
        new_t = transport.get_transport_from_url(t.base)
 
527
        self.assertEqual(t.server, new_t.server)
 
528
        self.assertEqual(t.base, new_t.base)
 
529
 
 
530
 
 
531
class ReadonlyDecoratorTransportTest(tests.TestCase):
541
532
    """Readonly decoration specific tests."""
542
533
 
543
534
    def test_local_parameters(self):
544
 
        import bzrlib.transport.readonly as readonly
545
535
        # connect to . in readonly mode
546
 
        transport = readonly.ReadonlyTransportDecorator('readonly+.')
547
 
        self.assertEqual(True, transport.listable())
548
 
        self.assertEqual(True, transport.is_readonly())
 
536
        t = readonly.ReadonlyTransportDecorator('readonly+.')
 
537
        self.assertEqual(True, t.listable())
 
538
        self.assertEqual(True, t.is_readonly())
549
539
 
550
540
    def test_http_parameters(self):
551
541
        from bzrlib.tests.http_server import HttpServer
552
 
        import bzrlib.transport.readonly as readonly
553
542
        # connect to '.' via http which is not listable
554
543
        server = HttpServer()
555
544
        self.start_server(server)
556
 
        transport = get_transport('readonly+' + server.get_url())
557
 
        self.failUnless(isinstance(transport,
558
 
                                   readonly.ReadonlyTransportDecorator))
559
 
        self.assertEqual(False, transport.listable())
560
 
        self.assertEqual(True, transport.is_readonly())
561
 
 
562
 
 
563
 
class FakeNFSDecoratorTests(TestCaseInTempDir):
 
545
        t = transport.get_transport_from_url('readonly+' + server.get_url())
 
546
        self.assertIsInstance(t, readonly.ReadonlyTransportDecorator)
 
547
        self.assertEqual(False, t.listable())
 
548
        self.assertEqual(True, t.is_readonly())
 
549
 
 
550
 
 
551
class FakeNFSDecoratorTests(tests.TestCaseInTempDir):
564
552
    """NFS decorator specific tests."""
565
553
 
566
554
    def get_nfs_transport(self, url):
567
 
        import bzrlib.transport.fakenfs as fakenfs
568
555
        # connect to url with nfs decoration
569
556
        return fakenfs.FakeNFSTransportDecorator('fakenfs+' + url)
570
557
 
571
558
    def test_local_parameters(self):
572
559
        # the listable and is_readonly parameters
573
560
        # are not changed by the fakenfs decorator
574
 
        transport = self.get_nfs_transport('.')
575
 
        self.assertEqual(True, transport.listable())
576
 
        self.assertEqual(False, transport.is_readonly())
 
561
        t = self.get_nfs_transport('.')
 
562
        self.assertEqual(True, t.listable())
 
563
        self.assertEqual(False, t.is_readonly())
577
564
 
578
565
    def test_http_parameters(self):
579
566
        # the listable and is_readonly parameters
582
569
        # connect to '.' via http which is not listable
583
570
        server = HttpServer()
584
571
        self.start_server(server)
585
 
        transport = self.get_nfs_transport(server.get_url())
586
 
        self.assertIsInstance(
587
 
            transport, bzrlib.transport.fakenfs.FakeNFSTransportDecorator)
588
 
        self.assertEqual(False, transport.listable())
589
 
        self.assertEqual(True, transport.is_readonly())
 
572
        t = self.get_nfs_transport(server.get_url())
 
573
        self.assertIsInstance(t, fakenfs.FakeNFSTransportDecorator)
 
574
        self.assertEqual(False, t.listable())
 
575
        self.assertEqual(True, t.is_readonly())
590
576
 
591
577
    def test_fakenfs_server_default(self):
592
578
        # a FakeNFSServer() should bring up a local relpath server for itself
593
 
        import bzrlib.transport.fakenfs as fakenfs
594
 
        server = fakenfs.FakeNFSServer()
 
579
        server = test_server.FakeNFSServer()
595
580
        self.start_server(server)
596
581
        # the url should be decorated appropriately
597
582
        self.assertStartsWith(server.get_url(), 'fakenfs+')
598
583
        # and we should be able to get a transport for it
599
 
        transport = get_transport(server.get_url())
 
584
        t = transport.get_transport_from_url(server.get_url())
600
585
        # which must be a FakeNFSTransportDecorator instance.
601
 
        self.assertIsInstance(transport, fakenfs.FakeNFSTransportDecorator)
 
586
        self.assertIsInstance(t, fakenfs.FakeNFSTransportDecorator)
602
587
 
603
588
    def test_fakenfs_rename_semantics(self):
604
589
        # a FakeNFS transport must mangle the way rename errors occur to
605
590
        # look like NFS problems.
606
 
        transport = self.get_nfs_transport('.')
 
591
        t = self.get_nfs_transport('.')
607
592
        self.build_tree(['from/', 'from/foo', 'to/', 'to/bar'],
608
 
                        transport=transport)
609
 
        self.assertRaises(errors.ResourceBusy,
610
 
                          transport.rename, 'from', 'to')
611
 
 
612
 
 
613
 
class FakeVFATDecoratorTests(TestCaseInTempDir):
 
593
                        transport=t)
 
594
        self.assertRaises(errors.ResourceBusy, t.rename, 'from', 'to')
 
595
 
 
596
 
 
597
class FakeVFATDecoratorTests(tests.TestCaseInTempDir):
614
598
    """Tests for simulation of VFAT restrictions"""
615
599
 
616
600
    def get_vfat_transport(self, url):
620
604
 
621
605
    def test_transport_creation(self):
622
606
        from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
623
 
        transport = self.get_vfat_transport('.')
624
 
        self.assertIsInstance(transport, FakeVFATTransportDecorator)
 
607
        t = self.get_vfat_transport('.')
 
608
        self.assertIsInstance(t, FakeVFATTransportDecorator)
625
609
 
626
610
    def test_transport_mkdir(self):
627
 
        transport = self.get_vfat_transport('.')
628
 
        transport.mkdir('HELLO')
629
 
        self.assertTrue(transport.has('hello'))
630
 
        self.assertTrue(transport.has('Hello'))
 
611
        t = self.get_vfat_transport('.')
 
612
        t.mkdir('HELLO')
 
613
        self.assertTrue(t.has('hello'))
 
614
        self.assertTrue(t.has('Hello'))
631
615
 
632
616
    def test_forbidden_chars(self):
633
 
        transport = self.get_vfat_transport('.')
634
 
        self.assertRaises(ValueError, transport.has, "<NU>")
635
 
 
636
 
 
637
 
class BadTransportHandler(Transport):
 
617
        t = self.get_vfat_transport('.')
 
618
        self.assertRaises(ValueError, t.has, "<NU>")
 
619
 
 
620
 
 
621
class BadTransportHandler(transport.Transport):
638
622
    def __init__(self, base_url):
639
 
        raise DependencyNotPresent('some_lib', 'testing missing dependency')
640
 
 
641
 
 
642
 
class BackupTransportHandler(Transport):
 
623
        raise errors.DependencyNotPresent('some_lib',
 
624
                                          'testing missing dependency')
 
625
 
 
626
 
 
627
class BackupTransportHandler(transport.Transport):
643
628
    """Test transport that works as a backup for the BadTransportHandler"""
644
629
    pass
645
630
 
646
631
 
647
 
class TestTransportImplementation(TestCaseInTempDir):
 
632
class TestTransportImplementation(tests.TestCaseInTempDir):
648
633
    """Implementation verification for transports.
649
634
 
650
635
    To verify a transport we need a server factory, which is a callable
679
664
        base_url = self._server.get_url()
680
665
        url = self._adjust_url(base_url, relpath)
681
666
        # try getting the transport via the regular interface:
682
 
        t = get_transport(url)
 
667
        t = transport.get_transport_from_url(url)
683
668
        # vila--20070607 if the following are commented out the test suite
684
669
        # still pass. Is this really still needed or was it a forgotten
685
670
        # temporary fix ?
690
675
        return t
691
676
 
692
677
 
693
 
class TestLocalTransports(TestCase):
 
678
class TestTransportFromPath(tests.TestCaseInTempDir):
 
679
 
 
680
    def test_with_path(self):
 
681
        t = transport.get_transport_from_path(self.test_dir)
 
682
        self.assertIsInstance(t, local.LocalTransport)
 
683
        self.assertEquals(t.base.rstrip("/"),
 
684
            urlutils.local_path_to_url(self.test_dir))
 
685
 
 
686
    def test_with_url(self):
 
687
        t = transport.get_transport_from_path("file:")
 
688
        self.assertIsInstance(t, local.LocalTransport)
 
689
        self.assertEquals(t.base.rstrip("/"),
 
690
            urlutils.local_path_to_url(os.path.join(self.test_dir, "file:")))
 
691
 
 
692
 
 
693
class TestTransportFromUrl(tests.TestCaseInTempDir):
 
694
 
 
695
    def test_with_path(self):
 
696
        self.assertRaises(errors.InvalidURL, transport.get_transport_from_url,
 
697
            self.test_dir)
 
698
 
 
699
    def test_with_url(self):
 
700
        url = urlutils.local_path_to_url(self.test_dir)
 
701
        t = transport.get_transport_from_url(url)
 
702
        self.assertIsInstance(t, local.LocalTransport)
 
703
        self.assertEquals(t.base.rstrip("/"), url)
 
704
 
 
705
    def test_with_url_and_segment_parameters(self):
 
706
        url = urlutils.local_path_to_url(self.test_dir)+",branch=foo"
 
707
        t = transport.get_transport_from_url(url)
 
708
        self.assertIsInstance(t, local.LocalTransport)
 
709
        self.assertEquals(t.base.rstrip("/"), url)
 
710
        with open(os.path.join(self.test_dir, "afile"), 'w') as f:
 
711
            f.write("data")
 
712
        self.assertTrue(t.has("afile"))
 
713
 
 
714
 
 
715
class TestLocalTransports(tests.TestCase):
694
716
 
695
717
    def test_get_transport_from_abspath(self):
696
718
        here = osutils.abspath('.')
697
 
        t = get_transport(here)
698
 
        self.assertIsInstance(t, LocalTransport)
 
719
        t = transport.get_transport(here)
 
720
        self.assertIsInstance(t, local.LocalTransport)
699
721
        self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
700
722
 
701
723
    def test_get_transport_from_relpath(self):
702
724
        here = osutils.abspath('.')
703
 
        t = get_transport('.')
704
 
        self.assertIsInstance(t, LocalTransport)
 
725
        t = transport.get_transport('.')
 
726
        self.assertIsInstance(t, local.LocalTransport)
705
727
        self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
706
728
 
707
729
    def test_get_transport_from_local_url(self):
708
730
        here = osutils.abspath('.')
709
731
        here_url = urlutils.local_path_to_url(here) + '/'
710
 
        t = get_transport(here_url)
711
 
        self.assertIsInstance(t, LocalTransport)
 
732
        t = transport.get_transport(here_url)
 
733
        self.assertIsInstance(t, local.LocalTransport)
712
734
        self.assertEquals(t.base, here_url)
713
735
 
714
736
    def test_local_abspath(self):
715
737
        here = osutils.abspath('.')
716
 
        t = get_transport(here)
 
738
        t = transport.get_transport(here)
717
739
        self.assertEquals(t.local_abspath(''), here)
718
740
 
719
741
 
720
 
class TestWin32LocalTransport(TestCase):
 
742
class TestLocalTransportMutation(tests.TestCaseInTempDir):
 
743
 
 
744
    def test_local_transport_mkdir(self):
 
745
        here = osutils.abspath('.')
 
746
        t = transport.get_transport(here)
 
747
        t.mkdir('test')
 
748
        self.assertTrue(os.path.exists('test'))
 
749
 
 
750
    def test_local_transport_mkdir_permission_denied(self):
 
751
        # See https://bugs.launchpad.net/bzr/+bug/606537
 
752
        here = osutils.abspath('.')
 
753
        t = transport.get_transport(here)
 
754
        def fake_chmod(path, mode):
 
755
            e = OSError('permission denied')
 
756
            e.errno = errno.EPERM
 
757
            raise e
 
758
        self.overrideAttr(os, 'chmod', fake_chmod)
 
759
        t.mkdir('test')
 
760
        t.mkdir('test2', mode=0707)
 
761
        self.assertTrue(os.path.exists('test'))
 
762
        self.assertTrue(os.path.exists('test2'))
 
763
 
 
764
 
 
765
class TestLocalTransportWriteStream(tests.TestCaseWithTransport):
 
766
 
 
767
    def test_local_fdatasync_calls_fdatasync(self):
 
768
        """Check fdatasync on a stream tries to flush the data to the OS.
 
769
        
 
770
        We can't easily observe the external effect but we can at least see
 
771
        it's called.
 
772
        """
 
773
        sentinel = object()
 
774
        fdatasync = getattr(os, 'fdatasync', sentinel)
 
775
        if fdatasync is sentinel:
 
776
            raise tests.TestNotApplicable('fdatasync not supported')
 
777
        t = self.get_transport('.')
 
778
        calls = self.recordCalls(os, 'fdatasync')
 
779
        w = t.open_write_stream('out')
 
780
        w.write('foo')
 
781
        w.fdatasync()
 
782
        with open('out', 'rb') as f:
 
783
            # Should have been flushed.
 
784
            self.assertEquals(f.read(), 'foo')
 
785
        self.assertEquals(len(calls), 1, calls)
 
786
 
 
787
    def test_missing_directory(self):
 
788
        t = self.get_transport('.')
 
789
        self.assertRaises(errors.NoSuchFile, t.open_write_stream, 'dir/foo')
 
790
 
 
791
 
 
792
class TestWin32LocalTransport(tests.TestCase):
721
793
 
722
794
    def test_unc_clone_to_root(self):
723
795
        # Win32 UNC path like \\HOST\path
724
796
        # clone to root should stop at least at \\HOST part
725
797
        # not on \\
726
 
        t = EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
 
798
        t = local.EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
727
799
        for i in xrange(4):
728
800
            t = t.clone('..')
729
801
        self.assertEquals(t.base, 'file://HOST/')
732
804
        self.assertEquals(t.base, 'file://HOST/')
733
805
 
734
806
 
735
 
class TestConnectedTransport(TestCase):
 
807
class TestConnectedTransport(tests.TestCase):
736
808
    """Tests for connected to remote server transports"""
737
809
 
738
810
    def test_parse_url(self):
739
 
        t = ConnectedTransport('http://simple.example.com/home/source')
740
 
        self.assertEquals(t._host, 'simple.example.com')
741
 
        self.assertEquals(t._port, None)
742
 
        self.assertEquals(t._path, '/home/source/')
743
 
        self.failUnless(t._user is None)
744
 
        self.failUnless(t._password is None)
 
811
        t = transport.ConnectedTransport(
 
812
            'http://simple.example.com/home/source')
 
813
        self.assertEquals(t._parsed_url.host, 'simple.example.com')
 
814
        self.assertEquals(t._parsed_url.port, None)
 
815
        self.assertEquals(t._parsed_url.path, '/home/source/')
 
816
        self.assertTrue(t._parsed_url.user is None)
 
817
        self.assertTrue(t._parsed_url.password is None)
745
818
 
746
819
        self.assertEquals(t.base, 'http://simple.example.com/home/source/')
747
820
 
748
821
    def test_parse_url_with_at_in_user(self):
749
822
        # Bug 228058
750
 
        t = ConnectedTransport('ftp://user@host.com@www.host.com/')
751
 
        self.assertEquals(t._user, 'user@host.com')
 
823
        t = transport.ConnectedTransport('ftp://user@host.com@www.host.com/')
 
824
        self.assertEquals(t._parsed_url.user, 'user@host.com')
752
825
 
753
826
    def test_parse_quoted_url(self):
754
 
        t = ConnectedTransport('http://ro%62ey:h%40t@ex%41mple.com:2222/path')
755
 
        self.assertEquals(t._host, 'exAmple.com')
756
 
        self.assertEquals(t._port, 2222)
757
 
        self.assertEquals(t._user, 'robey')
758
 
        self.assertEquals(t._password, 'h@t')
759
 
        self.assertEquals(t._path, '/path/')
 
827
        t = transport.ConnectedTransport(
 
828
            'http://ro%62ey:h%40t@ex%41mple.com:2222/path')
 
829
        self.assertEquals(t._parsed_url.host, 'exAmple.com')
 
830
        self.assertEquals(t._parsed_url.port, 2222)
 
831
        self.assertEquals(t._parsed_url.user, 'robey')
 
832
        self.assertEquals(t._parsed_url.password, 'h@t')
 
833
        self.assertEquals(t._parsed_url.path, '/path/')
760
834
 
761
835
        # Base should not keep track of the password
762
 
        self.assertEquals(t.base, 'http://robey@exAmple.com:2222/path/')
 
836
        self.assertEquals(t.base, 'http://ro%62ey@ex%41mple.com:2222/path/')
763
837
 
764
838
    def test_parse_invalid_url(self):
765
839
        self.assertRaises(errors.InvalidURL,
766
 
                          ConnectedTransport,
 
840
                          transport.ConnectedTransport,
767
841
                          'sftp://lily.org:~janneke/public/bzr/gub')
768
842
 
769
843
    def test_relpath(self):
770
 
        t = ConnectedTransport('sftp://user@host.com/abs/path')
 
844
        t = transport.ConnectedTransport('sftp://user@host.com/abs/path')
771
845
 
772
 
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
 
846
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'),
 
847
            'sub')
773
848
        self.assertRaises(errors.PathNotChild, t.relpath,
774
849
                          'http://user@host.com/abs/path/sub')
775
850
        self.assertRaises(errors.PathNotChild, t.relpath,
779
854
        self.assertRaises(errors.PathNotChild, t.relpath,
780
855
                          'sftp://user@host.com:33/abs/path/sub')
781
856
        # Make sure it works when we don't supply a username
782
 
        t = ConnectedTransport('sftp://host.com/abs/path')
 
857
        t = transport.ConnectedTransport('sftp://host.com/abs/path')
783
858
        self.assertEquals(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
784
859
 
785
860
        # Make sure it works when parts of the path will be url encoded
786
 
        t = ConnectedTransport('sftp://host.com/dev/%path')
 
861
        t = transport.ConnectedTransport('sftp://host.com/dev/%path')
787
862
        self.assertEquals(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
788
863
 
789
864
    def test_connection_sharing_propagate_credentials(self):
790
 
        t = ConnectedTransport('ftp://user@host.com/abs/path')
791
 
        self.assertEquals('user', t._user)
792
 
        self.assertEquals('host.com', t._host)
 
865
        t = transport.ConnectedTransport('ftp://user@host.com/abs/path')
 
866
        self.assertEquals('user', t._parsed_url.user)
 
867
        self.assertEquals('host.com', t._parsed_url.host)
793
868
        self.assertIs(None, t._get_connection())
794
 
        self.assertIs(None, t._password)
 
869
        self.assertIs(None, t._parsed_url.password)
795
870
        c = t.clone('subdir')
796
871
        self.assertIs(None, c._get_connection())
797
 
        self.assertIs(None, t._password)
 
872
        self.assertIs(None, t._parsed_url.password)
798
873
 
799
874
        # Simulate the user entering a password
800
875
        password = 'secret'
814
889
        self.assertIs(new_password, c._get_credentials())
815
890
 
816
891
 
817
 
class TestReusedTransports(TestCase):
 
892
class TestReusedTransports(tests.TestCase):
818
893
    """Tests for transport reuse"""
819
894
 
820
895
    def test_reuse_same_transport(self):
821
896
        possible_transports = []
822
 
        t1 = get_transport('http://foo/',
823
 
                           possible_transports=possible_transports)
 
897
        t1 = transport.get_transport_from_url('http://foo/',
 
898
                                     possible_transports=possible_transports)
824
899
        self.assertEqual([t1], possible_transports)
825
 
        t2 = get_transport('http://foo/', possible_transports=[t1])
 
900
        t2 = transport.get_transport_from_url('http://foo/',
 
901
                                     possible_transports=[t1])
826
902
        self.assertIs(t1, t2)
827
903
 
828
904
        # Also check that final '/' are handled correctly
829
 
        t3 = get_transport('http://foo/path/')
830
 
        t4 = get_transport('http://foo/path', possible_transports=[t3])
 
905
        t3 = transport.get_transport_from_url('http://foo/path/')
 
906
        t4 = transport.get_transport_from_url('http://foo/path',
 
907
                                     possible_transports=[t3])
831
908
        self.assertIs(t3, t4)
832
909
 
833
 
        t5 = get_transport('http://foo/path')
834
 
        t6 = get_transport('http://foo/path/', possible_transports=[t5])
 
910
        t5 = transport.get_transport_from_url('http://foo/path')
 
911
        t6 = transport.get_transport_from_url('http://foo/path/',
 
912
                                     possible_transports=[t5])
835
913
        self.assertIs(t5, t6)
836
914
 
837
915
    def test_don_t_reuse_different_transport(self):
838
 
        t1 = get_transport('http://foo/path')
839
 
        t2 = get_transport('http://bar/path', possible_transports=[t1])
 
916
        t1 = transport.get_transport_from_url('http://foo/path')
 
917
        t2 = transport.get_transport_from_url('http://bar/path',
 
918
                                     possible_transports=[t1])
840
919
        self.assertIsNot(t1, t2)
841
920
 
842
921
 
843
 
class TestTransportTrace(TestCase):
 
922
class TestTransportTrace(tests.TestCase):
844
923
 
845
 
    def test_get(self):
846
 
        transport = get_transport('trace+memory://')
 
924
    def test_decorator(self):
 
925
        t = transport.get_transport_from_url('trace+memory://')
847
926
        self.assertIsInstance(
848
 
            transport, bzrlib.transport.trace.TransportTraceDecorator)
 
927
            t, bzrlib.transport.trace.TransportTraceDecorator)
849
928
 
850
929
    def test_clone_preserves_activity(self):
851
 
        transport = get_transport('trace+memory://')
852
 
        transport2 = transport.clone('.')
853
 
        self.assertTrue(transport is not transport2)
854
 
        self.assertTrue(transport._activity is transport2._activity)
 
930
        t = transport.get_transport_from_url('trace+memory://')
 
931
        t2 = t.clone('.')
 
932
        self.assertTrue(t is not t2)
 
933
        self.assertTrue(t._activity is t2._activity)
855
934
 
856
935
    # the following specific tests are for the operations that have made use of
857
936
    # logging in tests; we could test every single operation but doing that
858
937
    # still won't cause a test failure when the top level Transport API
859
938
    # changes; so there is little return doing that.
860
939
    def test_get(self):
861
 
        transport = get_transport('trace+memory:///')
862
 
        transport.put_bytes('foo', 'barish')
863
 
        transport.get('foo')
 
940
        t = transport.get_transport_from_url('trace+memory:///')
 
941
        t.put_bytes('foo', 'barish')
 
942
        t.get('foo')
864
943
        expected_result = []
865
944
        # put_bytes records the bytes, not the content to avoid memory
866
945
        # pressure.
867
946
        expected_result.append(('put_bytes', 'foo', 6, None))
868
947
        # get records the file name only.
869
948
        expected_result.append(('get', 'foo'))
870
 
        self.assertEqual(expected_result, transport._activity)
 
949
        self.assertEqual(expected_result, t._activity)
871
950
 
872
951
    def test_readv(self):
873
 
        transport = get_transport('trace+memory:///')
874
 
        transport.put_bytes('foo', 'barish')
875
 
        list(transport.readv('foo', [(0, 1), (3, 2)], adjust_for_latency=True,
876
 
            upper_limit=6))
 
952
        t = transport.get_transport_from_url('trace+memory:///')
 
953
        t.put_bytes('foo', 'barish')
 
954
        list(t.readv('foo', [(0, 1), (3, 2)],
 
955
                     adjust_for_latency=True, upper_limit=6))
877
956
        expected_result = []
878
957
        # put_bytes records the bytes, not the content to avoid memory
879
958
        # pressure.
880
959
        expected_result.append(('put_bytes', 'foo', 6, None))
881
960
        # readv records the supplied offset request
882
961
        expected_result.append(('readv', 'foo', [(0, 1), (3, 2)], True, 6))
883
 
        self.assertEqual(expected_result, transport._activity)
 
962
        self.assertEqual(expected_result, t._activity)
884
963
 
885
964
 
886
965
class TestSSHConnections(tests.TestCaseWithTransport):
887
966
 
888
967
    def test_bzr_connect_to_bzr_ssh(self):
889
 
        """User acceptance that get_transport of a bzr+ssh:// behaves correctly.
 
968
        """get_transport of a bzr+ssh:// behaves correctly.
890
969
 
891
970
        bzr+ssh:// should cause bzr to run a remote bzr smart server over SSH.
892
971
        """
895
974
        # A reasonable evolution for this would be to simply check inside
896
975
        # check_channel_exec_request that the command is appropriate, and then
897
976
        # satisfy requests in-process.
898
 
        self.requireFeature(ParamikoFeature)
 
977
        self.requireFeature(features.paramiko)
899
978
        # SFTPFullAbsoluteServer has a get_url method, and doesn't
900
979
        # override the interface (doesn't change self._vendor).
901
980
        # Note that this does encryption, so can be slow.
902
 
        from bzrlib.transport.sftp import SFTPFullAbsoluteServer
903
 
        from bzrlib.tests.stub_sftp import StubServer
 
981
        from bzrlib.tests import stub_sftp
904
982
 
905
983
        # Start an SSH server
906
984
        self.command_executed = []
908
986
        # executes commands, and manage the hooking up of stdin/out/err to the
909
987
        # SSH channel ourselves.  Surely this has already been implemented
910
988
        # elsewhere?
911
 
        class StubSSHServer(StubServer):
 
989
        started = []
 
990
 
 
991
        class StubSSHServer(stub_sftp.StubServer):
912
992
 
913
993
            test = self
914
994
 
919
999
                    stdout=subprocess.PIPE, stderr=subprocess.PIPE)
920
1000
 
921
1001
                # XXX: horribly inefficient, not to mention ugly.
922
 
                # Start a thread for each of stdin/out/err, and relay bytes from
923
 
                # the subprocess to channel and vice versa.
 
1002
                # Start a thread for each of stdin/out/err, and relay bytes
 
1003
                # from the subprocess to channel and vice versa.
924
1004
                def ferry_bytes(read, write, close):
925
1005
                    while True:
926
1006
                        bytes = read(1)
933
1013
                    (channel.recv, proc.stdin.write, proc.stdin.close),
934
1014
                    (proc.stdout.read, channel.sendall, channel.close),
935
1015
                    (proc.stderr.read, channel.sendall_stderr, channel.close)]
 
1016
                started.append(proc)
936
1017
                for read, write, close in file_functions:
937
1018
                    t = threading.Thread(
938
1019
                        target=ferry_bytes, args=(read, write, close))
939
1020
                    t.start()
 
1021
                    started.append(t)
940
1022
 
941
1023
                return True
942
1024
 
943
 
        ssh_server = SFTPFullAbsoluteServer(StubSSHServer)
 
1025
        ssh_server = stub_sftp.SFTPFullAbsoluteServer(StubSSHServer)
944
1026
        # We *don't* want to override the default SSH vendor: the detected one
945
1027
        # is the one to use.
 
1028
 
 
1029
        # FIXME: I don't understand the above comment, SFTPFullAbsoluteServer
 
1030
        # inherits from SFTPServer which forces the SSH vendor to
 
1031
        # ssh.ParamikoVendor(). So it's forced, not detected. --vila 20100623
946
1032
        self.start_server(ssh_server)
947
 
        port = ssh_server._listener.port
 
1033
        port = ssh_server.port
948
1034
 
949
1035
        if sys.platform == 'win32':
950
1036
            bzr_remote_path = sys.executable + ' ' + self.get_bzr_path()
951
1037
        else:
952
1038
            bzr_remote_path = self.get_bzr_path()
953
 
        os.environ['BZR_REMOTE_PATH'] = bzr_remote_path
 
1039
        self.overrideEnv('BZR_REMOTE_PATH', bzr_remote_path)
954
1040
 
955
1041
        # Access the branch via a bzr+ssh URL.  The BZR_REMOTE_PATH environment
956
1042
        # variable is used to tell bzr what command to run on the remote end.
960
1046
            # prefix a '/' to get the right path.
961
1047
            path_to_branch = '/' + path_to_branch
962
1048
        url = 'bzr+ssh://fred:secret@localhost:%d%s' % (port, path_to_branch)
963
 
        t = get_transport(url)
 
1049
        t = transport.get_transport(url)
964
1050
        self.permit_url(t.base)
965
1051
        t.mkdir('foo')
966
1052
 
967
1053
        self.assertEqual(
968
1054
            ['%s serve --inet --directory=/ --allow-writes' % bzr_remote_path],
969
1055
            self.command_executed)
 
1056
        # Make sure to disconnect, so that the remote process can stop, and we
 
1057
        # can cleanup. Then pause the test until everything is shutdown
 
1058
        t._client._medium.disconnect()
 
1059
        if not started:
 
1060
            return
 
1061
        # First wait for the subprocess
 
1062
        started[0].wait()
 
1063
        # And the rest are threads
 
1064
        for t in started[1:]:
 
1065
            t.join()
 
1066
 
 
1067
 
 
1068
class TestUnhtml(tests.TestCase):
 
1069
 
 
1070
    """Tests for unhtml_roughly"""
 
1071
 
 
1072
    def test_truncation(self):
 
1073
        fake_html = "<p>something!\n" * 1000
 
1074
        result = http.unhtml_roughly(fake_html)
 
1075
        self.assertEquals(len(result), 1000)
 
1076
        self.assertStartsWith(result, " something!")
 
1077
 
 
1078
 
 
1079
class SomeDirectory(object):
 
1080
 
 
1081
    def look_up(self, name, url):
 
1082
        return "http://bar"
 
1083
 
 
1084
 
 
1085
class TestLocationToUrl(tests.TestCase):
 
1086
 
 
1087
    def get_base_location(self):
 
1088
        path = osutils.abspath('/foo/bar')
 
1089
        if path.startswith('/'):
 
1090
            url = 'file://%s' % (path,)
 
1091
        else:
 
1092
            # On Windows, abspaths start with the drive letter, so we have to
 
1093
            # add in the extra '/'
 
1094
            url = 'file:///%s' % (path,)
 
1095
        return path, url
 
1096
 
 
1097
    def test_regular_url(self):
 
1098
        self.assertEquals("file://foo", location_to_url("file://foo"))
 
1099
 
 
1100
    def test_directory(self):
 
1101
        directories.register("bar:", SomeDirectory, "Dummy directory")
 
1102
        self.addCleanup(directories.remove, "bar:")
 
1103
        self.assertEquals("http://bar", location_to_url("bar:"))
 
1104
 
 
1105
    def test_unicode_url(self):
 
1106
        self.assertRaises(errors.InvalidURL, location_to_url,
 
1107
            "http://fo/\xc3\xaf".decode("utf-8"))
 
1108
 
 
1109
    def test_unicode_path(self):
 
1110
        path, url = self.get_base_location()
 
1111
        location = path + "\xc3\xaf".decode("utf-8")
 
1112
        url += '%C3%AF'
 
1113
        self.assertEquals(url, location_to_url(location))
 
1114
 
 
1115
    def test_path(self):
 
1116
        path, url = self.get_base_location()
 
1117
        self.assertEquals(url, location_to_url(path))
 
1118
 
 
1119
    def test_relative_file_url(self):
 
1120
        self.assertEquals(urlutils.local_path_to_url(".") + "/bar",
 
1121
            location_to_url("file:bar"))
 
1122
 
 
1123
    def test_absolute_file_url(self):
 
1124
        self.assertEquals("file:///bar", location_to_url("file:/bar"))