← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

Merge bzr.dev.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
1
 
# Copyright (C) 2004, 2005, 2006 by Canonical Ltd
2
 
 
 
1
# Copyright (C) 2004, 2005, 2006, 2007 Canonical Ltd
 
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
5
5
# the Free Software Foundation; either version 2 of the License, or
6
6
# (at your option) any later version.
7
 
 
 
7
#
8
8
# This program is distributed in the hope that it will be useful,
9
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
11
# GNU General Public License for more details.
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
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
 
18
 
import os
19
 
import sys
20
 
import stat
21
18
from cStringIO import StringIO
22
19
 
23
20
import bzrlib
24
 
from bzrlib.errors import (NoSuchFile, FileExists,
25
 
                           TransportNotPossible,
26
 
                           ConnectionError,
27
 
                           DependencyNotPresent,
 
21
from bzrlib import (
 
22
    errors,
 
23
    osutils,
 
24
    urlutils,
 
25
    )
 
26
from bzrlib.errors import (DependencyNotPresent,
 
27
                           FileExists,
 
28
                           InvalidURLJoin,
 
29
                           NoSuchFile,
 
30
                           PathNotChild,
 
31
                           ReadError,
 
32
                           UnsupportedProtocol,
28
33
                           )
29
34
from bzrlib.tests import TestCase, TestCaseInTempDir
30
 
from bzrlib.transport import (_get_protocol_handlers,
 
35
from bzrlib.transport import (_clear_protocol_handlers,
 
36
                              _CoalescedOffset,
 
37
                              ConnectedTransport,
 
38
                              _get_protocol_handlers,
 
39
                              _set_protocol_handlers,
31
40
                              _get_transport_modules,
32
41
                              get_transport,
 
42
                              LateReadError,
33
43
                              register_lazy_transport,
34
 
                              _set_protocol_handlers,
35
 
                              urlescape,
 
44
                              register_transport_proto,
36
45
                              Transport,
37
46
                              )
 
47
from bzrlib.transport.chroot import ChrootServer
38
48
from bzrlib.transport.memory import MemoryTransport
39
 
from bzrlib.transport.local import LocalTransport
 
49
from bzrlib.transport.local import (LocalTransport,
 
50
                                    EmulatedWin32LocalTransport)
 
51
 
 
52
 
 
53
# TODO: Should possibly split transport-specific tests into their own files.
40
54
 
41
55
 
42
56
class TestTransport(TestCase):
43
57
    """Test the non transport-concrete class functionality."""
44
58
 
45
 
    def test_urlescape(self):
46
 
        self.assertEqual('%25', urlescape('%'))
47
 
 
48
59
    def test__get_set_protocol_handlers(self):
49
60
        handlers = _get_protocol_handlers()
50
 
        self.assertNotEqual({}, handlers)
 
61
        self.assertNotEqual([], handlers.keys( ))
51
62
        try:
52
 
            _set_protocol_handlers({})
53
 
            self.assertEqual({}, _get_protocol_handlers())
 
63
            _clear_protocol_handlers()
 
64
            self.assertEqual([], _get_protocol_handlers().keys())
54
65
        finally:
55
66
            _set_protocol_handlers(handlers)
56
67
 
57
68
    def test_get_transport_modules(self):
58
69
        handlers = _get_protocol_handlers()
 
70
        # don't pollute the current handlers
 
71
        _clear_protocol_handlers()
59
72
        class SampleHandler(object):
60
73
            """I exist, isnt that enough?"""
61
74
        try:
62
 
            my_handlers = {}
63
 
            _set_protocol_handlers(my_handlers)
64
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
65
 
            register_lazy_transport('bar', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
66
 
            self.assertEqual([SampleHandler.__module__],
 
75
            _clear_protocol_handlers()
 
76
            register_transport_proto('foo')
 
77
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
 
78
                                    'TestTransport.SampleHandler')
 
79
            register_transport_proto('bar')
 
80
            register_lazy_transport('bar', 'bzrlib.tests.test_transport',
 
81
                                    'TestTransport.SampleHandler')
 
82
            self.assertEqual([SampleHandler.__module__,
 
83
                              'bzrlib.transport.chroot'],
67
84
                             _get_transport_modules())
68
85
        finally:
69
86
            _set_protocol_handlers(handlers)
71
88
    def test_transport_dependency(self):
72
89
        """Transport with missing dependency causes no error"""
73
90
        saved_handlers = _get_protocol_handlers()
 
91
        # don't pollute the current handlers
 
92
        _clear_protocol_handlers()
74
93
        try:
 
94
            register_transport_proto('foo')
75
95
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
76
96
                    'BadTransportHandler')
77
 
            t = get_transport('foo://fooserver/foo')
78
 
            # because we failed to load the transport
79
 
            self.assertTrue(isinstance(t, LocalTransport))
 
97
            try:
 
98
                get_transport('foo://fooserver/foo')
 
99
            except UnsupportedProtocol, e:
 
100
                e_str = str(e)
 
101
                self.assertEquals('Unsupported protocol'
 
102
                                  ' for url "foo://fooserver/foo":'
 
103
                                  ' Unable to import library "some_lib":'
 
104
                                  ' testing missing dependency', str(e))
 
105
            else:
 
106
                self.fail('Did not raise UnsupportedProtocol')
80
107
        finally:
81
108
            # restore original values
82
109
            _set_protocol_handlers(saved_handlers)
85
112
        """Transport with missing dependency causes no error"""
86
113
        saved_handlers = _get_protocol_handlers()
87
114
        try:
88
 
            _set_protocol_handlers({})
 
115
            _clear_protocol_handlers()
 
116
            register_transport_proto('foo')
89
117
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
90
118
                    'BackupTransportHandler')
91
119
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
94
122
            self.assertTrue(isinstance(t, BackupTransportHandler))
95
123
        finally:
96
124
            _set_protocol_handlers(saved_handlers)
97
 
            
 
125
 
 
126
    def test_LateReadError(self):
 
127
        """The LateReadError helper should raise on read()."""
 
128
        a_file = LateReadError('a path')
 
129
        try:
 
130
            a_file.read()
 
131
        except ReadError, error:
 
132
            self.assertEqual('a path', error.path)
 
133
        self.assertRaises(ReadError, a_file.read, 40)
 
134
        a_file.close()
 
135
 
 
136
    def test__combine_paths(self):
 
137
        t = Transport('/')
 
138
        self.assertEqual('/home/sarah/project/foo',
 
139
                         t._combine_paths('/home/sarah', 'project/foo'))
 
140
        self.assertEqual('/etc',
 
141
                         t._combine_paths('/home/sarah', '../../etc'))
 
142
        self.assertEqual('/etc',
 
143
                         t._combine_paths('/home/sarah', '../../../etc'))
 
144
        self.assertEqual('/etc',
 
145
                         t._combine_paths('/home/sarah', '/etc'))
 
146
 
 
147
    def test_local_abspath_non_local_transport(self):
 
148
        # the base implementation should throw
 
149
        t = MemoryTransport()
 
150
        e = self.assertRaises(errors.NotLocalUrl, t.local_abspath, 't')
 
151
        self.assertEqual('memory:///t is not a local path.', str(e))
 
152
 
 
153
 
 
154
class TestCoalesceOffsets(TestCase):
 
155
 
 
156
    def check(self, expected, offsets, limit=0, max_size=0, fudge=0):
 
157
        coalesce = Transport._coalesce_offsets
 
158
        exp = [_CoalescedOffset(*x) for x in expected]
 
159
        out = list(coalesce(offsets, limit=limit, fudge_factor=fudge,
 
160
                            max_size=max_size))
 
161
        self.assertEqual(exp, out)
 
162
 
 
163
    def test_coalesce_empty(self):
 
164
        self.check([], [])
 
165
 
 
166
    def test_coalesce_simple(self):
 
167
        self.check([(0, 10, [(0, 10)])], [(0, 10)])
 
168
 
 
169
    def test_coalesce_unrelated(self):
 
170
        self.check([(0, 10, [(0, 10)]),
 
171
                    (20, 10, [(0, 10)]),
 
172
                   ], [(0, 10), (20, 10)])
 
173
 
 
174
    def test_coalesce_unsorted(self):
 
175
        self.check([(20, 10, [(0, 10)]),
 
176
                    (0, 10, [(0, 10)]),
 
177
                   ], [(20, 10), (0, 10)])
 
178
 
 
179
    def test_coalesce_nearby(self):
 
180
        self.check([(0, 20, [(0, 10), (10, 10)])],
 
181
                   [(0, 10), (10, 10)])
 
182
 
 
183
    def test_coalesce_overlapped(self):
 
184
        self.assertRaises(ValueError,
 
185
            self.check, [(0, 15, [(0, 10), (5, 10)])],
 
186
                        [(0, 10), (5, 10)])
 
187
 
 
188
    def test_coalesce_limit(self):
 
189
        self.check([(10, 50, [(0, 10), (10, 10), (20, 10),
 
190
                              (30, 10), (40, 10)]),
 
191
                    (60, 50, [(0, 10), (10, 10), (20, 10),
 
192
                              (30, 10), (40, 10)]),
 
193
                   ], [(10, 10), (20, 10), (30, 10), (40, 10),
 
194
                       (50, 10), (60, 10), (70, 10), (80, 10),
 
195
                       (90, 10), (100, 10)],
 
196
                    limit=5)
 
197
 
 
198
    def test_coalesce_no_limit(self):
 
199
        self.check([(10, 100, [(0, 10), (10, 10), (20, 10),
 
200
                               (30, 10), (40, 10), (50, 10),
 
201
                               (60, 10), (70, 10), (80, 10),
 
202
                               (90, 10)]),
 
203
                   ], [(10, 10), (20, 10), (30, 10), (40, 10),
 
204
                       (50, 10), (60, 10), (70, 10), (80, 10),
 
205
                       (90, 10), (100, 10)])
 
206
 
 
207
    def test_coalesce_fudge(self):
 
208
        self.check([(10, 30, [(0, 10), (20, 10)]),
 
209
                    (100, 10, [(0, 10),]),
 
210
                   ], [(10, 10), (30, 10), (100, 10)],
 
211
                   fudge=10
 
212
                  )
 
213
    def test_coalesce_max_size(self):
 
214
        self.check([(10, 20, [(0, 10), (10, 10)]),
 
215
                    (30, 50, [(0, 50)]),
 
216
                    # If one range is above max_size, it gets its own coalesced
 
217
                    # offset
 
218
                    (100, 80, [(0, 80),]),],
 
219
                   [(10, 10), (20, 10), (30, 50), (100, 80)],
 
220
                   max_size=50
 
221
                  )
 
222
 
 
223
    def test_coalesce_no_max_size(self):
 
224
        self.check([(10, 170, [(0, 10), (10, 10), (20, 50), (70, 100)]),],
 
225
                   [(10, 10), (20, 10), (30, 50), (80, 100)],
 
226
                  )
 
227
 
98
228
 
99
229
class TestMemoryTransport(TestCase):
100
230
 
104
234
    def test_clone(self):
105
235
        transport = MemoryTransport()
106
236
        self.assertTrue(isinstance(transport, MemoryTransport))
 
237
        self.assertEqual("memory:///", transport.clone("/").base)
107
238
 
108
239
    def test_abspath(self):
109
240
        transport = MemoryTransport()
110
 
        self.assertEqual("memory:/relpath", transport.abspath('relpath'))
111
 
 
112
 
    def test_relpath(self):
113
 
        transport = MemoryTransport()
 
241
        self.assertEqual("memory:///relpath", transport.abspath('relpath'))
 
242
 
 
243
    def test_abspath_of_root(self):
 
244
        transport = MemoryTransport()
 
245
        self.assertEqual("memory:///", transport.base)
 
246
        self.assertEqual("memory:///", transport.abspath('/'))
 
247
 
 
248
    def test_abspath_of_relpath_starting_at_root(self):
 
249
        transport = MemoryTransport()
 
250
        self.assertEqual("memory:///foo", transport.abspath('/foo'))
114
251
 
115
252
    def test_append_and_get(self):
116
253
        transport = MemoryTransport()
117
 
        transport.append('path', StringIO('content'))
 
254
        transport.append_bytes('path', 'content')
118
255
        self.assertEqual(transport.get('path').read(), 'content')
119
 
        transport.append('path', StringIO('content'))
 
256
        transport.append_file('path', StringIO('content'))
120
257
        self.assertEqual(transport.get('path').read(), 'contentcontent')
121
258
 
122
259
    def test_put_and_get(self):
123
260
        transport = MemoryTransport()
124
 
        transport.put('path', StringIO('content'))
 
261
        transport.put_file('path', StringIO('content'))
125
262
        self.assertEqual(transport.get('path').read(), 'content')
126
 
        transport.put('path', StringIO('content'))
 
263
        transport.put_bytes('path', 'content')
127
264
        self.assertEqual(transport.get('path').read(), 'content')
128
265
 
129
266
    def test_append_without_dir_fails(self):
130
267
        transport = MemoryTransport()
131
268
        self.assertRaises(NoSuchFile,
132
 
                          transport.append, 'dir/path', StringIO('content'))
 
269
                          transport.append_bytes, 'dir/path', 'content')
133
270
 
134
271
    def test_put_without_dir_fails(self):
135
272
        transport = MemoryTransport()
136
273
        self.assertRaises(NoSuchFile,
137
 
                          transport.put, 'dir/path', StringIO('content'))
 
274
                          transport.put_file, 'dir/path', StringIO('content'))
138
275
 
139
276
    def test_get_missing(self):
140
277
        transport = MemoryTransport()
146
283
 
147
284
    def test_has_present(self):
148
285
        transport = MemoryTransport()
149
 
        transport.append('foo', StringIO('content'))
 
286
        transport.append_bytes('foo', 'content')
150
287
        self.assertEquals(True, transport.has('foo'))
151
288
 
 
289
    def test_list_dir(self):
 
290
        transport = MemoryTransport()
 
291
        transport.put_bytes('foo', 'content')
 
292
        transport.mkdir('dir')
 
293
        transport.put_bytes('dir/subfoo', 'content')
 
294
        transport.put_bytes('dirlike', 'content')
 
295
 
 
296
        self.assertEquals(['dir', 'dirlike', 'foo'], sorted(transport.list_dir('.')))
 
297
        self.assertEquals(['subfoo'], sorted(transport.list_dir('dir')))
 
298
 
152
299
    def test_mkdir(self):
153
300
        transport = MemoryTransport()
154
301
        transport.mkdir('dir')
155
 
        transport.append('dir/path', StringIO('content'))
 
302
        transport.append_bytes('dir/path', 'content')
156
303
        self.assertEqual(transport.get('dir/path').read(), 'content')
157
304
 
158
305
    def test_mkdir_missing_parent(self):
168
315
    def test_parameters(self):
169
316
        transport = MemoryTransport()
170
317
        self.assertEqual(True, transport.listable())
171
 
        self.assertEqual(False, transport.should_cache())
172
318
        self.assertEqual(False, transport.is_readonly())
173
319
 
174
320
    def test_iter_files_recursive(self):
175
321
        transport = MemoryTransport()
176
322
        transport.mkdir('dir')
177
 
        transport.put('dir/foo', StringIO('content'))
178
 
        transport.put('dir/bar', StringIO('content'))
179
 
        transport.put('bar', StringIO('content'))
 
323
        transport.put_bytes('dir/foo', 'content')
 
324
        transport.put_bytes('dir/bar', 'content')
 
325
        transport.put_bytes('bar', 'content')
180
326
        paths = set(transport.iter_files_recursive())
181
327
        self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
182
328
 
183
329
    def test_stat(self):
184
330
        transport = MemoryTransport()
185
 
        transport.put('foo', StringIO('content'))
186
 
        transport.put('bar', StringIO('phowar'))
 
331
        transport.put_bytes('foo', 'content')
 
332
        transport.put_bytes('bar', 'phowar')
187
333
        self.assertEqual(7, transport.stat('foo').st_size)
188
334
        self.assertEqual(6, transport.stat('bar').st_size)
189
335
 
 
336
 
 
337
class ChrootDecoratorTransportTest(TestCase):
 
338
    """Chroot decoration specific tests."""
 
339
 
 
340
    def test_abspath(self):
 
341
        # The abspath is always relative to the chroot_url.
 
342
        server = ChrootServer(get_transport('memory:///foo/bar/'))
 
343
        server.setUp()
 
344
        transport = get_transport(server.get_url())
 
345
        self.assertEqual(server.get_url(), transport.abspath('/'))
 
346
 
 
347
        subdir_transport = transport.clone('subdir')
 
348
        self.assertEqual(server.get_url(), subdir_transport.abspath('/'))
 
349
        server.tearDown()
 
350
 
 
351
    def test_clone(self):
 
352
        server = ChrootServer(get_transport('memory:///foo/bar/'))
 
353
        server.setUp()
 
354
        transport = get_transport(server.get_url())
 
355
        # relpath from root and root path are the same
 
356
        relpath_cloned = transport.clone('foo')
 
357
        abspath_cloned = transport.clone('/foo')
 
358
        self.assertEqual(server, relpath_cloned.server)
 
359
        self.assertEqual(server, abspath_cloned.server)
 
360
        server.tearDown()
 
361
    
 
362
    def test_chroot_url_preserves_chroot(self):
 
363
        """Calling get_transport on a chroot transport's base should produce a
 
364
        transport with exactly the same behaviour as the original chroot
 
365
        transport.
 
366
 
 
367
        This is so that it is not possible to escape a chroot by doing::
 
368
            url = chroot_transport.base
 
369
            parent_url = urlutils.join(url, '..')
 
370
            new_transport = get_transport(parent_url)
 
371
        """
 
372
        server = ChrootServer(get_transport('memory:///path/subpath'))
 
373
        server.setUp()
 
374
        transport = get_transport(server.get_url())
 
375
        new_transport = get_transport(transport.base)
 
376
        self.assertEqual(transport.server, new_transport.server)
 
377
        self.assertEqual(transport.base, new_transport.base)
 
378
        server.tearDown()
190
379
        
 
380
    def test_urljoin_preserves_chroot(self):
 
381
        """Using urlutils.join(url, '..') on a chroot URL should not produce a
 
382
        URL that escapes the intended chroot.
 
383
 
 
384
        This is so that it is not possible to escape a chroot by doing::
 
385
            url = chroot_transport.base
 
386
            parent_url = urlutils.join(url, '..')
 
387
            new_transport = get_transport(parent_url)
 
388
        """
 
389
        server = ChrootServer(get_transport('memory:///path/'))
 
390
        server.setUp()
 
391
        transport = get_transport(server.get_url())
 
392
        self.assertRaises(
 
393
            InvalidURLJoin, urlutils.join, transport.base, '..')
 
394
        server.tearDown()
 
395
 
 
396
 
 
397
class ChrootServerTest(TestCase):
 
398
 
 
399
    def test_construct(self):
 
400
        backing_transport = MemoryTransport()
 
401
        server = ChrootServer(backing_transport)
 
402
        self.assertEqual(backing_transport, server.backing_transport)
 
403
 
 
404
    def test_setUp(self):
 
405
        backing_transport = MemoryTransport()
 
406
        server = ChrootServer(backing_transport)
 
407
        server.setUp()
 
408
        self.assertTrue(server.scheme in _get_protocol_handlers().keys())
 
409
 
 
410
    def test_tearDown(self):
 
411
        backing_transport = MemoryTransport()
 
412
        server = ChrootServer(backing_transport)
 
413
        server.setUp()
 
414
        server.tearDown()
 
415
        self.assertFalse(server.scheme in _get_protocol_handlers().keys())
 
416
 
 
417
    def test_get_url(self):
 
418
        backing_transport = MemoryTransport()
 
419
        server = ChrootServer(backing_transport)
 
420
        server.setUp()
 
421
        self.assertEqual('chroot-%d:///' % id(server), server.get_url())
 
422
        server.tearDown()
 
423
 
 
424
 
191
425
class ReadonlyDecoratorTransportTest(TestCase):
192
426
    """Readonly decoration specific tests."""
193
427
 
196
430
        # connect to . in readonly mode
197
431
        transport = readonly.ReadonlyTransportDecorator('readonly+.')
198
432
        self.assertEqual(True, transport.listable())
199
 
        self.assertEqual(False, transport.should_cache())
200
433
        self.assertEqual(True, transport.is_readonly())
201
434
 
202
435
    def test_http_parameters(self):
 
436
        from bzrlib.tests.http_server import HttpServer
203
437
        import bzrlib.transport.readonly as readonly
204
 
        from bzrlib.transport.http import HttpServer
205
 
        # connect to . via http which is not listable
 
438
        # connect to '.' via http which is not listable
206
439
        server = HttpServer()
207
440
        server.setUp()
208
441
        try:
210
443
            self.failUnless(isinstance(transport,
211
444
                                       readonly.ReadonlyTransportDecorator))
212
445
            self.assertEqual(False, transport.listable())
213
 
            self.assertEqual(True, transport.should_cache())
214
446
            self.assertEqual(True, transport.is_readonly())
215
447
        finally:
216
448
            server.tearDown()
225
457
        return fakenfs.FakeNFSTransportDecorator('fakenfs+' + url)
226
458
 
227
459
    def test_local_parameters(self):
228
 
        # the listable, should_cache and is_readonly parameters
 
460
        # the listable and is_readonly parameters
229
461
        # are not changed by the fakenfs decorator
230
462
        transport = self.get_nfs_transport('.')
231
463
        self.assertEqual(True, transport.listable())
232
 
        self.assertEqual(False, transport.should_cache())
233
464
        self.assertEqual(False, transport.is_readonly())
234
465
 
235
466
    def test_http_parameters(self):
236
 
        # the listable, should_cache and is_readonly parameters
 
467
        # the listable and is_readonly parameters
237
468
        # are not changed by the fakenfs decorator
238
 
        from bzrlib.transport.http import HttpServer
239
 
        # connect to . via http which is not listable
 
469
        from bzrlib.tests.http_server import HttpServer
 
470
        # connect to '.' via http which is not listable
240
471
        server = HttpServer()
241
472
        server.setUp()
242
473
        try:
244
475
            self.assertIsInstance(
245
476
                transport, bzrlib.transport.fakenfs.FakeNFSTransportDecorator)
246
477
            self.assertEqual(False, transport.listable())
247
 
            self.assertEqual(True, transport.should_cache())
248
478
            self.assertEqual(True, transport.is_readonly())
249
479
        finally:
250
480
            server.tearDown()
255
485
        server = fakenfs.FakeNFSServer()
256
486
        server.setUp()
257
487
        try:
258
 
            # the server should be a relpath localhost server
259
 
            self.assertEqual(server.get_url(), 'fakenfs+.')
 
488
            # the url should be decorated appropriately
 
489
            self.assertStartsWith(server.get_url(), 'fakenfs+')
260
490
            # and we should be able to get a transport for it
261
491
            transport = get_transport(server.get_url())
262
492
            # which must be a FakeNFSTransportDecorator instance.
271
501
        transport = self.get_nfs_transport('.')
272
502
        self.build_tree(['from/', 'from/foo', 'to/', 'to/bar'],
273
503
                        transport=transport)
274
 
        self.assertRaises(bzrlib.errors.ResourceBusy,
 
504
        self.assertRaises(errors.ResourceBusy,
275
505
                          transport.rename, 'from', 'to')
276
506
 
277
507
 
307
537
class BackupTransportHandler(Transport):
308
538
    """Test transport that works as a backup for the BadTransportHandler"""
309
539
    pass
 
540
 
 
541
 
 
542
class TestTransportImplementation(TestCaseInTempDir):
 
543
    """Implementation verification for transports.
 
544
    
 
545
    To verify a transport we need a server factory, which is a callable
 
546
    that accepts no parameters and returns an implementation of
 
547
    bzrlib.transport.Server.
 
548
    
 
549
    That Server is then used to construct transport instances and test
 
550
    the transport via loopback activity.
 
551
 
 
552
    Currently this assumes that the Transport object is connected to the 
 
553
    current working directory.  So that whatever is done 
 
554
    through the transport, should show up in the working 
 
555
    directory, and vice-versa. This is a bug, because its possible to have
 
556
    URL schemes which provide access to something that may not be 
 
557
    result in storage on the local disk, i.e. due to file system limits, or 
 
558
    due to it being a database or some other non-filesystem tool.
 
559
 
 
560
    This also tests to make sure that the functions work with both
 
561
    generators and lists (assuming iter(list) is effectively a generator)
 
562
    """
 
563
    
 
564
    def setUp(self):
 
565
        super(TestTransportImplementation, self).setUp()
 
566
        self._server = self.transport_server()
 
567
        self._server.setUp()
 
568
        self.addCleanup(self._server.tearDown)
 
569
 
 
570
    def get_transport(self, relpath=None):
 
571
        """Return a connected transport to the local directory.
 
572
 
 
573
        :param relpath: a path relative to the base url.
 
574
        """
 
575
        base_url = self._server.get_url()
 
576
        url = self._adjust_url(base_url, relpath)
 
577
        # try getting the transport via the regular interface:
 
578
        t = get_transport(url)
 
579
        # vila--20070607 if the following are commented out the test suite
 
580
        # still pass. Is this really still needed or was it a forgotten
 
581
        # temporary fix ?
 
582
        if not isinstance(t, self.transport_class):
 
583
            # we did not get the correct transport class type. Override the
 
584
            # regular connection behaviour by direct construction.
 
585
            t = self.transport_class(url)
 
586
        return t
 
587
 
 
588
 
 
589
class TestLocalTransports(TestCase):
 
590
 
 
591
    def test_get_transport_from_abspath(self):
 
592
        here = osutils.abspath('.')
 
593
        t = get_transport(here)
 
594
        self.assertIsInstance(t, LocalTransport)
 
595
        self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
 
596
 
 
597
    def test_get_transport_from_relpath(self):
 
598
        here = osutils.abspath('.')
 
599
        t = get_transport('.')
 
600
        self.assertIsInstance(t, LocalTransport)
 
601
        self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
 
602
 
 
603
    def test_get_transport_from_local_url(self):
 
604
        here = osutils.abspath('.')
 
605
        here_url = urlutils.local_path_to_url(here) + '/'
 
606
        t = get_transport(here_url)
 
607
        self.assertIsInstance(t, LocalTransport)
 
608
        self.assertEquals(t.base, here_url)
 
609
 
 
610
    def test_local_abspath(self):
 
611
        here = osutils.abspath('.')
 
612
        t = get_transport(here)
 
613
        self.assertEquals(t.local_abspath(''), here)
 
614
 
 
615
 
 
616
class TestWin32LocalTransport(TestCase):
 
617
 
 
618
    def test_unc_clone_to_root(self):
 
619
        # Win32 UNC path like \\HOST\path
 
620
        # clone to root should stop at least at \\HOST part
 
621
        # not on \\
 
622
        t = EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
 
623
        for i in xrange(4):
 
624
            t = t.clone('..')
 
625
        self.assertEquals(t.base, 'file://HOST/')
 
626
        # make sure we reach the root
 
627
        t = t.clone('..')
 
628
        self.assertEquals(t.base, 'file://HOST/')
 
629
 
 
630
 
 
631
class TestConnectedTransport(TestCase):
 
632
    """Tests for connected to remote server transports"""
 
633
 
 
634
    def test_parse_url(self):
 
635
        t = ConnectedTransport('http://simple.example.com/home/source')
 
636
        self.assertEquals(t._host, 'simple.example.com')
 
637
        self.assertEquals(t._port, None)
 
638
        self.assertEquals(t._path, '/home/source/')
 
639
        self.failUnless(t._user is None)
 
640
        self.failUnless(t._password is None)
 
641
 
 
642
        self.assertEquals(t.base, 'http://simple.example.com/home/source/')
 
643
 
 
644
    def test_parse_url_with_at_in_user(self):
 
645
        # Bug 228058
 
646
        t = ConnectedTransport('ftp://user@host.com@www.host.com/')
 
647
        self.assertEquals(t._user, 'user@host.com')
 
648
 
 
649
    def test_parse_quoted_url(self):
 
650
        t = ConnectedTransport('http://ro%62ey:h%40t@ex%41mple.com:2222/path')
 
651
        self.assertEquals(t._host, 'exAmple.com')
 
652
        self.assertEquals(t._port, 2222)
 
653
        self.assertEquals(t._user, 'robey')
 
654
        self.assertEquals(t._password, 'h@t')
 
655
        self.assertEquals(t._path, '/path/')
 
656
 
 
657
        # Base should not keep track of the password
 
658
        self.assertEquals(t.base, 'http://robey@exAmple.com:2222/path/')
 
659
 
 
660
    def test_parse_invalid_url(self):
 
661
        self.assertRaises(errors.InvalidURL,
 
662
                          ConnectedTransport,
 
663
                          'sftp://lily.org:~janneke/public/bzr/gub')
 
664
 
 
665
    def test_relpath(self):
 
666
        t = ConnectedTransport('sftp://user@host.com/abs/path')
 
667
 
 
668
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
 
669
        self.assertRaises(errors.PathNotChild, t.relpath,
 
670
                          'http://user@host.com/abs/path/sub')
 
671
        self.assertRaises(errors.PathNotChild, t.relpath,
 
672
                          'sftp://user2@host.com/abs/path/sub')
 
673
        self.assertRaises(errors.PathNotChild, t.relpath,
 
674
                          'sftp://user@otherhost.com/abs/path/sub')
 
675
        self.assertRaises(errors.PathNotChild, t.relpath,
 
676
                          'sftp://user@host.com:33/abs/path/sub')
 
677
        # Make sure it works when we don't supply a username
 
678
        t = ConnectedTransport('sftp://host.com/abs/path')
 
679
        self.assertEquals(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
 
680
 
 
681
        # Make sure it works when parts of the path will be url encoded
 
682
        t = ConnectedTransport('sftp://host.com/dev/%path')
 
683
        self.assertEquals(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
 
684
 
 
685
    def test_connection_sharing_propagate_credentials(self):
 
686
        t = ConnectedTransport('ftp://user@host.com/abs/path')
 
687
        self.assertEquals('user', t._user)
 
688
        self.assertEquals('host.com', t._host)
 
689
        self.assertIs(None, t._get_connection())
 
690
        self.assertIs(None, t._password)
 
691
        c = t.clone('subdir')
 
692
        self.assertIs(None, c._get_connection())
 
693
        self.assertIs(None, t._password)
 
694
 
 
695
        # Simulate the user entering a password
 
696
        password = 'secret'
 
697
        connection = object()
 
698
        t._set_connection(connection, password)
 
699
        self.assertIs(connection, t._get_connection())
 
700
        self.assertIs(password, t._get_credentials())
 
701
        self.assertIs(connection, c._get_connection())
 
702
        self.assertIs(password, c._get_credentials())
 
703
 
 
704
        # credentials can be updated
 
705
        new_password = 'even more secret'
 
706
        c._update_credentials(new_password)
 
707
        self.assertIs(connection, t._get_connection())
 
708
        self.assertIs(new_password, t._get_credentials())
 
709
        self.assertIs(connection, c._get_connection())
 
710
        self.assertIs(new_password, c._get_credentials())
 
711
 
 
712
 
 
713
class TestReusedTransports(TestCase):
 
714
    """Tests for transport reuse"""
 
715
 
 
716
    def test_reuse_same_transport(self):
 
717
        possible_transports = []
 
718
        t1 = get_transport('http://foo/',
 
719
                           possible_transports=possible_transports)
 
720
        self.assertEqual([t1], possible_transports)
 
721
        t2 = get_transport('http://foo/', possible_transports=[t1])
 
722
        self.assertIs(t1, t2)
 
723
 
 
724
        # Also check that final '/' are handled correctly
 
725
        t3 = get_transport('http://foo/path/')
 
726
        t4 = get_transport('http://foo/path', possible_transports=[t3])
 
727
        self.assertIs(t3, t4)
 
728
 
 
729
        t5 = get_transport('http://foo/path')
 
730
        t6 = get_transport('http://foo/path/', possible_transports=[t5])
 
731
        self.assertIs(t5, t6)
 
732
 
 
733
    def test_don_t_reuse_different_transport(self):
 
734
        t1 = get_transport('http://foo/path')
 
735
        t2 = get_transport('http://bar/path', possible_transports=[t1])
 
736
        self.assertIsNot(t1, t2)
 
737
 
 
738
 
 
739
class TestTransportTrace(TestCase):
 
740
 
 
741
    def test_get(self):
 
742
        transport = get_transport('trace+memory://')
 
743
        self.assertIsInstance(
 
744
            transport, bzrlib.transport.trace.TransportTraceDecorator)
 
745
 
 
746
    def test_clone_preserves_activity(self):
 
747
        transport = get_transport('trace+memory://')
 
748
        transport2 = transport.clone('.')
 
749
        self.assertTrue(transport is not transport2)
 
750
        self.assertTrue(transport._activity is transport2._activity)
 
751
 
 
752
    # the following specific tests are for the operations that have made use of
 
753
    # logging in tests; we could test every single operation but doing that
 
754
    # still won't cause a test failure when the top level Transport API
 
755
    # changes; so there is little return doing that.
 
756
    def test_get(self):
 
757
        transport = get_transport('trace+memory:///')
 
758
        transport.put_bytes('foo', 'barish')
 
759
        transport.get('foo')
 
760
        expected_result = []
 
761
        # put_bytes records the bytes, not the content to avoid memory
 
762
        # pressure.
 
763
        expected_result.append(('put_bytes', 'foo', 6, None))
 
764
        # get records the file name only.
 
765
        expected_result.append(('get', 'foo'))
 
766
        self.assertEqual(expected_result, transport._activity)
 
767
 
 
768
    def test_readv(self):
 
769
        transport = get_transport('trace+memory:///')
 
770
        transport.put_bytes('foo', 'barish')
 
771
        list(transport.readv('foo', [(0, 1), (3, 2)], adjust_for_latency=True,
 
772
            upper_limit=6))
 
773
        expected_result = []
 
774
        # put_bytes records the bytes, not the content to avoid memory
 
775
        # pressure.
 
776
        expected_result.append(('put_bytes', 'foo', 6, None))
 
777
        # readv records the supplied offset request
 
778
        expected_result.append(('readv', 'foo', [(0, 1), (3, 2)], True, 6))
 
779
        self.assertEqual(expected_result, transport._activity)