← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2011-05-04 12:10:51 UTC
  • mfrom: (5819.1.4 777007-developer-doc)
  • Revision ID: pqm@pqm.ubuntu.com-20110504121051-aovlsmqiivjmc4fc
(jelmer) Small fixes to developer documentation. (Jonathan Riddell)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
1
 
# Copyright (C) 2004, 2005, 2006 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
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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
 
 
17
 
 
18
 
import os
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
16
 
 
17
 
 
18
from cStringIO import StringIO
 
19
import subprocess
19
20
import sys
20
 
import stat
21
 
from cStringIO import StringIO
 
21
import threading
22
22
 
23
 
import bzrlib
24
 
from bzrlib import urlutils
25
 
from bzrlib.errors import (NoSuchFile, FileExists,
26
 
                           TransportNotPossible,
27
 
                           ConnectionError,
28
 
                           DependencyNotPresent,
29
 
                           UnsupportedProtocol,
30
 
                           PathNotChild,
31
 
                           )
32
 
from bzrlib.tests import TestCase, TestCaseInTempDir
33
 
from bzrlib.transport import (_CoalescedOffset,
34
 
                              _get_protocol_handlers,
35
 
                              _get_transport_modules,
36
 
                              get_transport,
37
 
                              register_lazy_transport,
38
 
                              _set_protocol_handlers,
39
 
                              Transport,
40
 
                              )
41
 
from bzrlib.transport.memory import MemoryTransport
42
 
from bzrlib.transport.local import LocalTransport
 
23
from bzrlib import (
 
24
    errors,
 
25
    osutils,
 
26
    tests,
 
27
    transport,
 
28
    urlutils,
 
29
    )
 
30
from bzrlib.transport import (
 
31
    chroot,
 
32
    fakenfs,
 
33
    http,
 
34
    local,
 
35
    memory,
 
36
    pathfilter,
 
37
    readonly,
 
38
    )
 
39
from bzrlib.tests import (
 
40
    features,
 
41
    test_server,
 
42
    )
43
43
 
44
44
 
45
45
# TODO: Should possibly split transport-specific tests into their own files.
46
46
 
47
47
 
48
 
class TestTransport(TestCase):
 
48
class TestTransport(tests.TestCase):
49
49
    """Test the non transport-concrete class functionality."""
50
50
 
 
51
    # FIXME: These tests should use addCleanup() and/or overrideAttr() instead
 
52
    # of try/finally -- vila 20100205
 
53
 
51
54
    def test__get_set_protocol_handlers(self):
52
 
        handlers = _get_protocol_handlers()
53
 
        self.assertNotEqual({}, handlers)
 
55
        handlers = transport._get_protocol_handlers()
 
56
        self.assertNotEqual([], handlers.keys( ))
54
57
        try:
55
 
            _set_protocol_handlers({})
56
 
            self.assertEqual({}, _get_protocol_handlers())
 
58
            transport._clear_protocol_handlers()
 
59
            self.assertEqual([], transport._get_protocol_handlers().keys())
57
60
        finally:
58
 
            _set_protocol_handlers(handlers)
 
61
            transport._set_protocol_handlers(handlers)
59
62
 
60
63
    def test_get_transport_modules(self):
61
 
        handlers = _get_protocol_handlers()
 
64
        handlers = transport._get_protocol_handlers()
 
65
        # don't pollute the current handlers
 
66
        transport._clear_protocol_handlers()
62
67
        class SampleHandler(object):
63
68
            """I exist, isnt that enough?"""
64
69
        try:
65
 
            my_handlers = {}
66
 
            _set_protocol_handlers(my_handlers)
67
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
68
 
            register_lazy_transport('bar', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
69
 
            self.assertEqual([SampleHandler.__module__],
70
 
                             _get_transport_modules())
 
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())
71
83
        finally:
72
 
            _set_protocol_handlers(handlers)
 
84
            transport._set_protocol_handlers(handlers)
73
85
 
74
86
    def test_transport_dependency(self):
75
87
        """Transport with missing dependency causes no error"""
76
 
        saved_handlers = _get_protocol_handlers()
 
88
        saved_handlers = transport._get_protocol_handlers()
 
89
        # don't pollute the current handlers
 
90
        transport._clear_protocol_handlers()
77
91
        try:
78
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
79
 
                    'BadTransportHandler')
 
92
            transport.register_transport_proto('foo')
 
93
            transport.register_lazy_transport(
 
94
                'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
80
95
            try:
81
 
                get_transport('foo://fooserver/foo')
82
 
            except UnsupportedProtocol, e:
 
96
                transport.get_transport('foo://fooserver/foo')
 
97
            except errors.UnsupportedProtocol, e:
83
98
                e_str = str(e)
84
99
                self.assertEquals('Unsupported protocol'
85
100
                                  ' for url "foo://fooserver/foo":'
89
104
                self.fail('Did not raise UnsupportedProtocol')
90
105
        finally:
91
106
            # restore original values
92
 
            _set_protocol_handlers(saved_handlers)
93
 
            
 
107
            transport._set_protocol_handlers(saved_handlers)
 
108
 
94
109
    def test_transport_fallback(self):
95
110
        """Transport with missing dependency causes no error"""
96
 
        saved_handlers = _get_protocol_handlers()
 
111
        saved_handlers = transport._get_protocol_handlers()
97
112
        try:
98
 
            _set_protocol_handlers({})
99
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
100
 
                    'BackupTransportHandler')
101
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
102
 
                    'BadTransportHandler')
103
 
            t = get_transport('foo://fooserver/foo')
 
113
            transport._clear_protocol_handlers()
 
114
            transport.register_transport_proto('foo')
 
115
            transport.register_lazy_transport(
 
116
                'foo', 'bzrlib.tests.test_transport', 'BackupTransportHandler')
 
117
            transport.register_lazy_transport(
 
118
                'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
 
119
            t = transport.get_transport('foo://fooserver/foo')
104
120
            self.assertTrue(isinstance(t, BackupTransportHandler))
105
121
        finally:
106
 
            _set_protocol_handlers(saved_handlers)
 
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()
107
147
 
108
148
    def test__combine_paths(self):
109
 
        t = Transport('/')
 
149
        t = transport.Transport('/')
110
150
        self.assertEqual('/home/sarah/project/foo',
111
151
                         t._combine_paths('/home/sarah', 'project/foo'))
112
152
        self.assertEqual('/etc',
116
156
        self.assertEqual('/etc',
117
157
                         t._combine_paths('/home/sarah', '/etc'))
118
158
 
119
 
 
120
 
class TestCoalesceOffsets(TestCase):
121
 
    
122
 
    def check(self, expected, offsets, limit=0, fudge=0):
123
 
        coalesce = Transport._coalesce_offsets
124
 
        exp = [_CoalescedOffset(*x) for x in expected]
125
 
        out = list(coalesce(offsets, limit=limit, fudge_factor=fudge))
 
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))
126
173
        self.assertEqual(exp, out)
127
174
 
128
175
    def test_coalesce_empty(self):
135
182
        self.check([(0, 10, [(0, 10)]),
136
183
                    (20, 10, [(0, 10)]),
137
184
                   ], [(0, 10), (20, 10)])
138
 
            
 
185
 
139
186
    def test_coalesce_unsorted(self):
140
187
        self.check([(20, 10, [(0, 10)]),
141
188
                    (0, 10, [(0, 10)]),
146
193
                   [(0, 10), (10, 10)])
147
194
 
148
195
    def test_coalesce_overlapped(self):
149
 
        self.check([(0, 15, [(0, 10), (5, 10)])],
150
 
                   [(0, 10), (5, 10)])
 
196
        self.assertRaises(ValueError,
 
197
            self.check, [(0, 15, [(0, 10), (5, 10)])],
 
198
                        [(0, 10), (5, 10)])
151
199
 
152
200
    def test_coalesce_limit(self):
153
201
        self.check([(10, 50, [(0, 10), (10, 10), (20, 10),
174
222
                   ], [(10, 10), (30, 10), (100, 10)],
175
223
                   fudge=10
176
224
                  )
177
 
 
178
 
 
179
 
class TestMemoryTransport(TestCase):
 
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
267
 
181
268
    def test_get_transport(self):
182
 
        MemoryTransport()
 
269
        memory.MemoryTransport()
183
270
 
184
271
    def test_clone(self):
185
 
        transport = MemoryTransport()
186
 
        self.assertTrue(isinstance(transport, MemoryTransport))
187
 
        self.assertEqual("memory:///", transport.clone("/").base)
 
272
        t = memory.MemoryTransport()
 
273
        self.assertTrue(isinstance(t, memory.MemoryTransport))
 
274
        self.assertEqual("memory:///", t.clone("/").base)
188
275
 
189
276
    def test_abspath(self):
190
 
        transport = MemoryTransport()
191
 
        self.assertEqual("memory:///relpath", transport.abspath('relpath'))
 
277
        t = memory.MemoryTransport()
 
278
        self.assertEqual("memory:///relpath", t.abspath('relpath'))
192
279
 
193
280
    def test_abspath_of_root(self):
194
 
        transport = MemoryTransport()
195
 
        self.assertEqual("memory:///", transport.base)
196
 
        self.assertEqual("memory:///", transport.abspath('/'))
 
281
        t = memory.MemoryTransport()
 
282
        self.assertEqual("memory:///", t.base)
 
283
        self.assertEqual("memory:///", t.abspath('/'))
197
284
 
198
285
    def test_abspath_of_relpath_starting_at_root(self):
199
 
        transport = MemoryTransport()
200
 
        self.assertEqual("memory:///foo", transport.abspath('/foo'))
 
286
        t = memory.MemoryTransport()
 
287
        self.assertEqual("memory:///foo", t.abspath('/foo'))
201
288
 
202
289
    def test_append_and_get(self):
203
 
        transport = MemoryTransport()
204
 
        transport.append_bytes('path', 'content')
205
 
        self.assertEqual(transport.get('path').read(), 'content')
206
 
        transport.append_file('path', StringIO('content'))
207
 
        self.assertEqual(transport.get('path').read(), 'contentcontent')
 
290
        t = memory.MemoryTransport()
 
291
        t.append_bytes('path', 'content')
 
292
        self.assertEqual(t.get('path').read(), 'content')
 
293
        t.append_file('path', StringIO('content'))
 
294
        self.assertEqual(t.get('path').read(), 'contentcontent')
208
295
 
209
296
    def test_put_and_get(self):
210
 
        transport = MemoryTransport()
211
 
        transport.put_file('path', StringIO('content'))
212
 
        self.assertEqual(transport.get('path').read(), 'content')
213
 
        transport.put_bytes('path', 'content')
214
 
        self.assertEqual(transport.get('path').read(), 'content')
 
297
        t = memory.MemoryTransport()
 
298
        t.put_file('path', StringIO('content'))
 
299
        self.assertEqual(t.get('path').read(), 'content')
 
300
        t.put_bytes('path', 'content')
 
301
        self.assertEqual(t.get('path').read(), 'content')
215
302
 
216
303
    def test_append_without_dir_fails(self):
217
 
        transport = MemoryTransport()
218
 
        self.assertRaises(NoSuchFile,
219
 
                          transport.append_bytes, 'dir/path', 'content')
 
304
        t = memory.MemoryTransport()
 
305
        self.assertRaises(errors.NoSuchFile,
 
306
                          t.append_bytes, 'dir/path', 'content')
220
307
 
221
308
    def test_put_without_dir_fails(self):
222
 
        transport = MemoryTransport()
223
 
        self.assertRaises(NoSuchFile,
224
 
                          transport.put_file, 'dir/path', StringIO('content'))
 
309
        t = memory.MemoryTransport()
 
310
        self.assertRaises(errors.NoSuchFile,
 
311
                          t.put_file, 'dir/path', StringIO('content'))
225
312
 
226
313
    def test_get_missing(self):
227
 
        transport = MemoryTransport()
228
 
        self.assertRaises(NoSuchFile, transport.get, 'foo')
 
314
        transport = memory.MemoryTransport()
 
315
        self.assertRaises(errors.NoSuchFile, transport.get, 'foo')
229
316
 
230
317
    def test_has_missing(self):
231
 
        transport = MemoryTransport()
232
 
        self.assertEquals(False, transport.has('foo'))
 
318
        t = memory.MemoryTransport()
 
319
        self.assertEquals(False, t.has('foo'))
233
320
 
234
321
    def test_has_present(self):
235
 
        transport = MemoryTransport()
236
 
        transport.append_bytes('foo', 'content')
237
 
        self.assertEquals(True, transport.has('foo'))
 
322
        t = memory.MemoryTransport()
 
323
        t.append_bytes('foo', 'content')
 
324
        self.assertEquals(True, t.has('foo'))
238
325
 
239
326
    def test_list_dir(self):
240
 
        transport = MemoryTransport()
241
 
        transport.put_bytes('foo', 'content')
242
 
        transport.mkdir('dir')
243
 
        transport.put_bytes('dir/subfoo', 'content')
244
 
        transport.put_bytes('dirlike', 'content')
 
327
        t = memory.MemoryTransport()
 
328
        t.put_bytes('foo', 'content')
 
329
        t.mkdir('dir')
 
330
        t.put_bytes('dir/subfoo', 'content')
 
331
        t.put_bytes('dirlike', 'content')
245
332
 
246
 
        self.assertEquals(['dir', 'dirlike', 'foo'], sorted(transport.list_dir('.')))
247
 
        self.assertEquals(['subfoo'], sorted(transport.list_dir('dir')))
 
333
        self.assertEquals(['dir', 'dirlike', 'foo'], sorted(t.list_dir('.')))
 
334
        self.assertEquals(['subfoo'], sorted(t.list_dir('dir')))
248
335
 
249
336
    def test_mkdir(self):
250
 
        transport = MemoryTransport()
251
 
        transport.mkdir('dir')
252
 
        transport.append_bytes('dir/path', 'content')
253
 
        self.assertEqual(transport.get('dir/path').read(), 'content')
 
337
        t = memory.MemoryTransport()
 
338
        t.mkdir('dir')
 
339
        t.append_bytes('dir/path', 'content')
 
340
        self.assertEqual(t.get('dir/path').read(), 'content')
254
341
 
255
342
    def test_mkdir_missing_parent(self):
256
 
        transport = MemoryTransport()
257
 
        self.assertRaises(NoSuchFile,
258
 
                          transport.mkdir, 'dir/dir')
 
343
        t = memory.MemoryTransport()
 
344
        self.assertRaises(errors.NoSuchFile, t.mkdir, 'dir/dir')
259
345
 
260
346
    def test_mkdir_twice(self):
261
 
        transport = MemoryTransport()
262
 
        transport.mkdir('dir')
263
 
        self.assertRaises(FileExists, transport.mkdir, 'dir')
 
347
        t = memory.MemoryTransport()
 
348
        t.mkdir('dir')
 
349
        self.assertRaises(errors.FileExists, t.mkdir, 'dir')
264
350
 
265
351
    def test_parameters(self):
266
 
        transport = MemoryTransport()
267
 
        self.assertEqual(True, transport.listable())
268
 
        self.assertEqual(False, transport.should_cache())
269
 
        self.assertEqual(False, transport.is_readonly())
 
352
        t = memory.MemoryTransport()
 
353
        self.assertEqual(True, t.listable())
 
354
        self.assertEqual(False, t.is_readonly())
270
355
 
271
356
    def test_iter_files_recursive(self):
272
 
        transport = MemoryTransport()
273
 
        transport.mkdir('dir')
274
 
        transport.put_bytes('dir/foo', 'content')
275
 
        transport.put_bytes('dir/bar', 'content')
276
 
        transport.put_bytes('bar', 'content')
277
 
        paths = set(transport.iter_files_recursive())
 
357
        t = memory.MemoryTransport()
 
358
        t.mkdir('dir')
 
359
        t.put_bytes('dir/foo', 'content')
 
360
        t.put_bytes('dir/bar', 'content')
 
361
        t.put_bytes('bar', 'content')
 
362
        paths = set(t.iter_files_recursive())
278
363
        self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
279
364
 
280
365
    def test_stat(self):
281
 
        transport = MemoryTransport()
282
 
        transport.put_bytes('foo', 'content')
283
 
        transport.put_bytes('bar', 'phowar')
284
 
        self.assertEqual(7, transport.stat('foo').st_size)
285
 
        self.assertEqual(6, transport.stat('bar').st_size)
286
 
 
287
 
 
288
 
class ChrootDecoratorTransportTest(TestCase):
 
366
        t = memory.MemoryTransport()
 
367
        t.put_bytes('foo', 'content')
 
368
        t.put_bytes('bar', 'phowar')
 
369
        self.assertEqual(7, t.stat('foo').st_size)
 
370
        self.assertEqual(6, t.stat('bar').st_size)
 
371
 
 
372
 
 
373
class ChrootDecoratorTransportTest(tests.TestCase):
289
374
    """Chroot decoration specific tests."""
290
375
 
 
376
    def test_abspath(self):
 
377
        # 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
 
291
434
    def test_construct(self):
292
 
        from bzrlib.transport import chroot
293
 
        transport = chroot.ChrootTransportDecorator('chroot+memory:///pathA/')
294
 
        self.assertEqual('memory:///pathA/', transport.chroot_url)
295
 
 
296
 
        transport = chroot.ChrootTransportDecorator(
297
 
            'chroot+memory:///path/B', chroot='memory:///path/')
298
 
        self.assertEqual('memory:///path/', transport.chroot_url)
299
 
 
300
 
    def test_append_file(self):
301
 
        transport = get_transport('chroot+memory:///foo/bar')
302
 
        self.assertRaises(PathNotChild, transport.append_file, '/foo', None)
303
 
 
304
 
    def test_append_bytes(self):
305
 
        transport = get_transport('chroot+memory:///foo/bar')
306
 
        self.assertRaises(PathNotChild, transport.append_bytes, '/foo', 'bytes')
 
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)
307
525
 
308
526
    def test_clone(self):
309
 
        transport = get_transport('chroot+memory:///foo/bar')
310
 
        self.assertRaises(PathNotChild, transport.clone, '/foo')
311
 
 
312
 
    def test_delete(self):
313
 
        transport = get_transport('chroot+memory:///foo/bar')
314
 
        self.assertRaises(PathNotChild, transport.delete, '/foo')
315
 
 
316
 
    def test_delete_tree(self):
317
 
        transport = get_transport('chroot+memory:///foo/bar')
318
 
        self.assertRaises(PathNotChild, transport.delete_tree, '/foo')
319
 
 
320
 
    def test_get(self):
321
 
        transport = get_transport('chroot+memory:///foo/bar')
322
 
        self.assertRaises(PathNotChild, transport.get, '/foo')
323
 
 
324
 
    def test_get_bytes(self):
325
 
        transport = get_transport('chroot+memory:///foo/bar')
326
 
        self.assertRaises(PathNotChild, transport.get_bytes, '/foo')
327
 
 
328
 
    def test_has(self):
329
 
        transport = get_transport('chroot+memory:///foo/bar')
330
 
        self.assertRaises(PathNotChild, transport.has, '/foo')
331
 
 
332
 
    def test_list_dir(self):
333
 
        transport = get_transport('chroot+memory:///foo/bar')
334
 
        self.assertRaises(PathNotChild, transport.list_dir, '/foo')
335
 
 
336
 
    def test_lock_read(self):
337
 
        transport = get_transport('chroot+memory:///foo/bar')
338
 
        self.assertRaises(PathNotChild, transport.lock_read, '/foo')
339
 
 
340
 
    def test_lock_write(self):
341
 
        transport = get_transport('chroot+memory:///foo/bar')
342
 
        self.assertRaises(PathNotChild, transport.lock_write, '/foo')
343
 
 
344
 
    def test_mkdir(self):
345
 
        transport = get_transport('chroot+memory:///foo/bar')
346
 
        self.assertRaises(PathNotChild, transport.mkdir, '/foo')
347
 
 
348
 
    def test_put_bytes(self):
349
 
        transport = get_transport('chroot+memory:///foo/bar')
350
 
        self.assertRaises(PathNotChild, transport.put_bytes, '/foo', 'bytes')
351
 
 
352
 
    def test_put_file(self):
353
 
        transport = get_transport('chroot+memory:///foo/bar')
354
 
        self.assertRaises(PathNotChild, transport.put_file, '/foo', None)
355
 
 
356
 
    def test_rename(self):
357
 
        transport = get_transport('chroot+memory:///foo/bar')
358
 
        self.assertRaises(PathNotChild, transport.rename, '/aaa', 'bbb')
359
 
        self.assertRaises(PathNotChild, transport.rename, 'ccc', '/d')
360
 
 
361
 
    def test_rmdir(self):
362
 
        transport = get_transport('chroot+memory:///foo/bar')
363
 
        self.assertRaises(PathNotChild, transport.rmdir, '/foo')
364
 
 
365
 
    def test_stat(self):
366
 
        transport = get_transport('chroot+memory:///foo/bar')
367
 
        self.assertRaises(PathNotChild, transport.stat, '/foo')
368
 
 
369
 
 
370
 
class ReadonlyDecoratorTransportTest(TestCase):
 
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):
371
552
    """Readonly decoration specific tests."""
372
553
 
373
554
    def test_local_parameters(self):
374
 
        import bzrlib.transport.readonly as readonly
375
555
        # connect to . in readonly mode
376
 
        transport = readonly.ReadonlyTransportDecorator('readonly+.')
377
 
        self.assertEqual(True, transport.listable())
378
 
        self.assertEqual(False, transport.should_cache())
379
 
        self.assertEqual(True, transport.is_readonly())
 
556
        t = readonly.ReadonlyTransportDecorator('readonly+.')
 
557
        self.assertEqual(True, t.listable())
 
558
        self.assertEqual(True, t.is_readonly())
380
559
 
381
560
    def test_http_parameters(self):
382
 
        from bzrlib.tests.HttpServer import HttpServer
383
 
        import bzrlib.transport.readonly as readonly
384
 
        # connect to . via http which is not listable
 
561
        from bzrlib.tests.http_server import HttpServer
 
562
        # connect to '.' via http which is not listable
385
563
        server = HttpServer()
386
 
        server.setUp()
387
 
        try:
388
 
            transport = get_transport('readonly+' + server.get_url())
389
 
            self.failUnless(isinstance(transport,
390
 
                                       readonly.ReadonlyTransportDecorator))
391
 
            self.assertEqual(False, transport.listable())
392
 
            self.assertEqual(True, transport.should_cache())
393
 
            self.assertEqual(True, transport.is_readonly())
394
 
        finally:
395
 
            server.tearDown()
396
 
 
397
 
 
398
 
class FakeNFSDecoratorTests(TestCaseInTempDir):
 
564
        self.start_server(server)
 
565
        t = transport.get_transport('readonly+' + server.get_url())
 
566
        self.assertIsInstance(t, readonly.ReadonlyTransportDecorator)
 
567
        self.assertEqual(False, t.listable())
 
568
        self.assertEqual(True, t.is_readonly())
 
569
 
 
570
 
 
571
class FakeNFSDecoratorTests(tests.TestCaseInTempDir):
399
572
    """NFS decorator specific tests."""
400
573
 
401
574
    def get_nfs_transport(self, url):
402
 
        import bzrlib.transport.fakenfs as fakenfs
403
575
        # connect to url with nfs decoration
404
576
        return fakenfs.FakeNFSTransportDecorator('fakenfs+' + url)
405
577
 
406
578
    def test_local_parameters(self):
407
 
        # the listable, should_cache and is_readonly parameters
 
579
        # the listable and is_readonly parameters
408
580
        # are not changed by the fakenfs decorator
409
 
        transport = self.get_nfs_transport('.')
410
 
        self.assertEqual(True, transport.listable())
411
 
        self.assertEqual(False, transport.should_cache())
412
 
        self.assertEqual(False, transport.is_readonly())
 
581
        t = self.get_nfs_transport('.')
 
582
        self.assertEqual(True, t.listable())
 
583
        self.assertEqual(False, t.is_readonly())
413
584
 
414
585
    def test_http_parameters(self):
415
 
        # the listable, should_cache and is_readonly parameters
 
586
        # the listable and is_readonly parameters
416
587
        # are not changed by the fakenfs decorator
417
 
        from bzrlib.tests.HttpServer import HttpServer
418
 
        # connect to . via http which is not listable
 
588
        from bzrlib.tests.http_server import HttpServer
 
589
        # connect to '.' via http which is not listable
419
590
        server = HttpServer()
420
 
        server.setUp()
421
 
        try:
422
 
            transport = self.get_nfs_transport(server.get_url())
423
 
            self.assertIsInstance(
424
 
                transport, bzrlib.transport.fakenfs.FakeNFSTransportDecorator)
425
 
            self.assertEqual(False, transport.listable())
426
 
            self.assertEqual(True, transport.should_cache())
427
 
            self.assertEqual(True, transport.is_readonly())
428
 
        finally:
429
 
            server.tearDown()
 
591
        self.start_server(server)
 
592
        t = self.get_nfs_transport(server.get_url())
 
593
        self.assertIsInstance(t, fakenfs.FakeNFSTransportDecorator)
 
594
        self.assertEqual(False, t.listable())
 
595
        self.assertEqual(True, t.is_readonly())
430
596
 
431
597
    def test_fakenfs_server_default(self):
432
598
        # a FakeNFSServer() should bring up a local relpath server for itself
433
 
        import bzrlib.transport.fakenfs as fakenfs
434
 
        server = fakenfs.FakeNFSServer()
435
 
        server.setUp()
436
 
        try:
437
 
            # the url should be decorated appropriately
438
 
            self.assertStartsWith(server.get_url(), 'fakenfs+')
439
 
            # and we should be able to get a transport for it
440
 
            transport = get_transport(server.get_url())
441
 
            # which must be a FakeNFSTransportDecorator instance.
442
 
            self.assertIsInstance(
443
 
                transport, fakenfs.FakeNFSTransportDecorator)
444
 
        finally:
445
 
            server.tearDown()
 
599
        server = test_server.FakeNFSServer()
 
600
        self.start_server(server)
 
601
        # the url should be decorated appropriately
 
602
        self.assertStartsWith(server.get_url(), 'fakenfs+')
 
603
        # and we should be able to get a transport for it
 
604
        t = transport.get_transport(server.get_url())
 
605
        # which must be a FakeNFSTransportDecorator instance.
 
606
        self.assertIsInstance(t, fakenfs.FakeNFSTransportDecorator)
446
607
 
447
608
    def test_fakenfs_rename_semantics(self):
448
609
        # a FakeNFS transport must mangle the way rename errors occur to
449
610
        # look like NFS problems.
450
 
        transport = self.get_nfs_transport('.')
 
611
        t = self.get_nfs_transport('.')
451
612
        self.build_tree(['from/', 'from/foo', 'to/', 'to/bar'],
452
 
                        transport=transport)
453
 
        self.assertRaises(bzrlib.errors.ResourceBusy,
454
 
                          transport.rename, 'from', 'to')
455
 
 
456
 
 
457
 
class FakeVFATDecoratorTests(TestCaseInTempDir):
 
613
                        transport=t)
 
614
        self.assertRaises(errors.ResourceBusy, t.rename, 'from', 'to')
 
615
 
 
616
 
 
617
class FakeVFATDecoratorTests(tests.TestCaseInTempDir):
458
618
    """Tests for simulation of VFAT restrictions"""
459
619
 
460
620
    def get_vfat_transport(self, url):
464
624
 
465
625
    def test_transport_creation(self):
466
626
        from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
467
 
        transport = self.get_vfat_transport('.')
468
 
        self.assertIsInstance(transport, FakeVFATTransportDecorator)
 
627
        t = self.get_vfat_transport('.')
 
628
        self.assertIsInstance(t, FakeVFATTransportDecorator)
469
629
 
470
630
    def test_transport_mkdir(self):
471
 
        transport = self.get_vfat_transport('.')
472
 
        transport.mkdir('HELLO')
473
 
        self.assertTrue(transport.has('hello'))
474
 
        self.assertTrue(transport.has('Hello'))
 
631
        t = self.get_vfat_transport('.')
 
632
        t.mkdir('HELLO')
 
633
        self.assertTrue(t.has('hello'))
 
634
        self.assertTrue(t.has('Hello'))
475
635
 
476
636
    def test_forbidden_chars(self):
477
 
        transport = self.get_vfat_transport('.')
478
 
        self.assertRaises(ValueError, transport.has, "<NU>")
479
 
 
480
 
 
481
 
class BadTransportHandler(Transport):
 
637
        t = self.get_vfat_transport('.')
 
638
        self.assertRaises(ValueError, t.has, "<NU>")
 
639
 
 
640
 
 
641
class BadTransportHandler(transport.Transport):
482
642
    def __init__(self, base_url):
483
 
        raise DependencyNotPresent('some_lib', 'testing missing dependency')
484
 
 
485
 
 
486
 
class BackupTransportHandler(Transport):
 
643
        raise errors.DependencyNotPresent('some_lib',
 
644
                                          'testing missing dependency')
 
645
 
 
646
 
 
647
class BackupTransportHandler(transport.Transport):
487
648
    """Test transport that works as a backup for the BadTransportHandler"""
488
649
    pass
489
650
 
490
651
 
491
 
class TestTransportImplementation(TestCaseInTempDir):
 
652
class TestTransportImplementation(tests.TestCaseInTempDir):
492
653
    """Implementation verification for transports.
493
 
    
 
654
 
494
655
    To verify a transport we need a server factory, which is a callable
495
656
    that accepts no parameters and returns an implementation of
496
657
    bzrlib.transport.Server.
497
 
    
 
658
 
498
659
    That Server is then used to construct transport instances and test
499
660
    the transport via loopback activity.
500
661
 
501
 
    Currently this assumes that the Transport object is connected to the 
502
 
    current working directory.  So that whatever is done 
503
 
    through the transport, should show up in the working 
 
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
504
665
    directory, and vice-versa. This is a bug, because its possible to have
505
 
    URL schemes which provide access to something that may not be 
506
 
    result in storage on the local disk, i.e. due to file system limits, or 
 
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
507
668
    due to it being a database or some other non-filesystem tool.
508
669
 
509
670
    This also tests to make sure that the functions work with both
510
671
    generators and lists (assuming iter(list) is effectively a generator)
511
672
    """
512
 
    
 
673
 
513
674
    def setUp(self):
514
675
        super(TestTransportImplementation, self).setUp()
515
676
        self._server = self.transport_server()
516
 
        self._server.setUp()
517
 
 
518
 
    def tearDown(self):
519
 
        super(TestTransportImplementation, self).tearDown()
520
 
        self._server.tearDown()
521
 
        
522
 
    def get_transport(self):
523
 
        """Return a connected transport to the local directory."""
 
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
        """
524
684
        base_url = self._server.get_url()
 
685
        url = self._adjust_url(base_url, relpath)
525
686
        # try getting the transport via the regular interface:
526
 
        t = get_transport(base_url)
 
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 ?
527
691
        if not isinstance(t, self.transport_class):
528
692
            # we did not get the correct transport class type. Override the
529
693
            # regular connection behaviour by direct construction.
530
 
            t = self.transport_class(base_url)
 
694
            t = self.transport_class(url)
531
695
        return t
532
696
 
533
697
 
534
 
class TestLocalTransports(TestCase):
 
698
class TestLocalTransports(tests.TestCase):
535
699
 
536
700
    def test_get_transport_from_abspath(self):
537
 
        here = os.path.abspath('.')
538
 
        t = get_transport(here)
539
 
        self.assertIsInstance(t, LocalTransport)
 
701
        here = osutils.abspath('.')
 
702
        t = transport.get_transport(here)
 
703
        self.assertIsInstance(t, local.LocalTransport)
540
704
        self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
541
705
 
542
706
    def test_get_transport_from_relpath(self):
543
 
        here = os.path.abspath('.')
544
 
        t = get_transport('.')
545
 
        self.assertIsInstance(t, LocalTransport)
 
707
        here = osutils.abspath('.')
 
708
        t = transport.get_transport('.')
 
709
        self.assertIsInstance(t, local.LocalTransport)
546
710
        self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
547
711
 
548
712
    def test_get_transport_from_local_url(self):
549
 
        here = os.path.abspath('.')
 
713
        here = osutils.abspath('.')
550
714
        here_url = urlutils.local_path_to_url(here) + '/'
551
 
        t = get_transport(here_url)
552
 
        self.assertIsInstance(t, LocalTransport)
 
715
        t = transport.get_transport(here_url)
 
716
        self.assertIsInstance(t, local.LocalTransport)
553
717
        self.assertEquals(t.base, here_url)
 
718
 
 
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):
 
726
 
 
727
    def test_unc_clone_to_root(self):
 
728
        # Win32 UNC path like \\HOST\path
 
729
        # clone to root should stop at least at \\HOST part
 
730
        # not on \\
 
731
        t = local.EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
 
732
        for i in xrange(4):
 
733
            t = t.clone('..')
 
734
        self.assertEquals(t.base, 'file://HOST/')
 
735
        # make sure we reach the root
 
736
        t = t.clone('..')
 
737
        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.assertTrue(t._user is None)
 
750
        self.assertTrue(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
 
 
959
        # FIXME: I don't understand the above comment, SFTPFullAbsoluteServer
 
960
        # inherits from SFTPServer which forces the SSH vendor to
 
961
        # ssh.ParamikoVendor(). So it's forced, not detected. --vila 20100623
 
962
        self.start_server(ssh_server)
 
963
        port = ssh_server.port
 
964
 
 
965
        if sys.platform == 'win32':
 
966
            bzr_remote_path = sys.executable + ' ' + self.get_bzr_path()
 
967
        else:
 
968
            bzr_remote_path = self.get_bzr_path()
 
969
        self.overrideEnv('BZR_REMOTE_PATH', bzr_remote_path)
 
970
 
 
971
        # Access the branch via a bzr+ssh URL.  The BZR_REMOTE_PATH environment
 
972
        # variable is used to tell bzr what command to run on the remote end.
 
973
        path_to_branch = osutils.abspath('.')
 
974
        if sys.platform == 'win32':
 
975
            # On Windows, we export all drives as '/C:/, etc. So we need to
 
976
            # prefix a '/' to get the right path.
 
977
            path_to_branch = '/' + path_to_branch
 
978
        url = 'bzr+ssh://fred:secret@localhost:%d%s' % (port, path_to_branch)
 
979
        t = transport.get_transport(url)
 
980
        self.permit_url(t.base)
 
981
        t.mkdir('foo')
 
982
 
 
983
        self.assertEqual(
 
984
            ['%s serve --inet --directory=/ --allow-writes' % bzr_remote_path],
 
985
            self.command_executed)
 
986
        # Make sure to disconnect, so that the remote process can stop, and we
 
987
        # can cleanup. Then pause the test until everything is shutdown
 
988
        t._client._medium.disconnect()
 
989
        if not started:
 
990
            return
 
991
        # First wait for the subprocess
 
992
        started[0].wait()
 
993
        # And the rest are threads
 
994
        for t in started[1:]:
 
995
            t.join()
 
996
 
 
997
 
 
998
class TestUnhtml(tests.TestCase):
 
999
 
 
1000
    """Tests for unhtml_roughly"""
 
1001
 
 
1002
    def test_truncation(self):
 
1003
        fake_html = "<p>something!\n" * 1000
 
1004
        result = http.unhtml_roughly(fake_html)
 
1005
        self.assertEquals(len(result), 1000)
 
1006
        self.assertStartsWith(result, " something!")