← 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: 2008-07-16 23:23:13 UTC
  • mfrom: (1551.19.47 Aaron's mergeable stuff)
  • Revision ID: pqm@pqm.ubuntu.com-20080716232313-9pcoze89hualg5qn
Fetch copies all required data even with stacking (abentley, #248506)

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,
28
 
                           InvalidURL,
 
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,
29
33
                           )
30
34
from bzrlib.tests import TestCase, TestCaseInTempDir
31
 
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,
32
40
                              _get_transport_modules,
33
41
                              get_transport,
 
42
                              LateReadError,
34
43
                              register_lazy_transport,
35
 
                              _set_protocol_handlers,
 
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):
44
58
 
45
59
    def test__get_set_protocol_handlers(self):
46
60
        handlers = _get_protocol_handlers()
47
 
        self.assertNotEqual({}, handlers)
 
61
        self.assertNotEqual([], handlers.keys( ))
48
62
        try:
49
 
            _set_protocol_handlers({})
50
 
            self.assertEqual({}, _get_protocol_handlers())
 
63
            _clear_protocol_handlers()
 
64
            self.assertEqual([], _get_protocol_handlers().keys())
51
65
        finally:
52
66
            _set_protocol_handlers(handlers)
53
67
 
54
68
    def test_get_transport_modules(self):
55
69
        handlers = _get_protocol_handlers()
 
70
        # don't pollute the current handlers
 
71
        _clear_protocol_handlers()
56
72
        class SampleHandler(object):
57
73
            """I exist, isnt that enough?"""
58
74
        try:
59
 
            my_handlers = {}
60
 
            _set_protocol_handlers(my_handlers)
61
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
62
 
            register_lazy_transport('bar', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
63
 
            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'],
64
84
                             _get_transport_modules())
65
85
        finally:
66
86
            _set_protocol_handlers(handlers)
68
88
    def test_transport_dependency(self):
69
89
        """Transport with missing dependency causes no error"""
70
90
        saved_handlers = _get_protocol_handlers()
 
91
        # don't pollute the current handlers
 
92
        _clear_protocol_handlers()
71
93
        try:
 
94
            register_transport_proto('foo')
72
95
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
73
96
                    'BadTransportHandler')
74
 
            # TODO: jam 20060427 Now we get InvalidURL because it looks like 
75
 
            #       a URL but we have no support for it.
76
 
            #       Is it better to always fall back to LocalTransport?
77
 
            #       I think this is a better error than a future NoSuchFile
78
 
            self.assertRaises(InvalidURL, get_transport, 'foo://fooserver/foo')
 
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')
79
107
        finally:
80
108
            # restore original values
81
109
            _set_protocol_handlers(saved_handlers)
84
112
        """Transport with missing dependency causes no error"""
85
113
        saved_handlers = _get_protocol_handlers()
86
114
        try:
87
 
            _set_protocol_handlers({})
 
115
            _clear_protocol_handlers()
 
116
            register_transport_proto('foo')
88
117
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
89
118
                    'BackupTransportHandler')
90
119
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
93
122
            self.assertTrue(isinstance(t, BackupTransportHandler))
94
123
        finally:
95
124
            _set_protocol_handlers(saved_handlers)
96
 
            
 
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
    # XXX: scary, http.readv() can't handle that --vila20071209
 
184
    def test_coalesce_overlapped(self):
 
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
 
97
228
 
98
229
class TestMemoryTransport(TestCase):
99
230
 
103
234
    def test_clone(self):
104
235
        transport = MemoryTransport()
105
236
        self.assertTrue(isinstance(transport, MemoryTransport))
 
237
        self.assertEqual("memory:///", transport.clone("/").base)
106
238
 
107
239
    def test_abspath(self):
108
240
        transport = MemoryTransport()
109
241
        self.assertEqual("memory:///relpath", transport.abspath('relpath'))
110
242
 
111
 
    def test_relpath(self):
112
 
        transport = MemoryTransport()
 
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'))
113
251
 
114
252
    def test_append_and_get(self):
115
253
        transport = MemoryTransport()
116
 
        transport.append('path', StringIO('content'))
 
254
        transport.append_bytes('path', 'content')
117
255
        self.assertEqual(transport.get('path').read(), 'content')
118
 
        transport.append('path', StringIO('content'))
 
256
        transport.append_file('path', StringIO('content'))
119
257
        self.assertEqual(transport.get('path').read(), 'contentcontent')
120
258
 
121
259
    def test_put_and_get(self):
122
260
        transport = MemoryTransport()
123
 
        transport.put('path', StringIO('content'))
 
261
        transport.put_file('path', StringIO('content'))
124
262
        self.assertEqual(transport.get('path').read(), 'content')
125
 
        transport.put('path', StringIO('content'))
 
263
        transport.put_bytes('path', 'content')
126
264
        self.assertEqual(transport.get('path').read(), 'content')
127
265
 
128
266
    def test_append_without_dir_fails(self):
129
267
        transport = MemoryTransport()
130
268
        self.assertRaises(NoSuchFile,
131
 
                          transport.append, 'dir/path', StringIO('content'))
 
269
                          transport.append_bytes, 'dir/path', 'content')
132
270
 
133
271
    def test_put_without_dir_fails(self):
134
272
        transport = MemoryTransport()
135
273
        self.assertRaises(NoSuchFile,
136
 
                          transport.put, 'dir/path', StringIO('content'))
 
274
                          transport.put_file, 'dir/path', StringIO('content'))
137
275
 
138
276
    def test_get_missing(self):
139
277
        transport = MemoryTransport()
145
283
 
146
284
    def test_has_present(self):
147
285
        transport = MemoryTransport()
148
 
        transport.append('foo', StringIO('content'))
 
286
        transport.append_bytes('foo', 'content')
149
287
        self.assertEquals(True, transport.has('foo'))
150
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
 
151
299
    def test_mkdir(self):
152
300
        transport = MemoryTransport()
153
301
        transport.mkdir('dir')
154
 
        transport.append('dir/path', StringIO('content'))
 
302
        transport.append_bytes('dir/path', 'content')
155
303
        self.assertEqual(transport.get('dir/path').read(), 'content')
156
304
 
157
305
    def test_mkdir_missing_parent(self):
167
315
    def test_parameters(self):
168
316
        transport = MemoryTransport()
169
317
        self.assertEqual(True, transport.listable())
170
 
        self.assertEqual(False, transport.should_cache())
171
318
        self.assertEqual(False, transport.is_readonly())
172
319
 
173
320
    def test_iter_files_recursive(self):
174
321
        transport = MemoryTransport()
175
322
        transport.mkdir('dir')
176
 
        transport.put('dir/foo', StringIO('content'))
177
 
        transport.put('dir/bar', StringIO('content'))
178
 
        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')
179
326
        paths = set(transport.iter_files_recursive())
180
327
        self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
181
328
 
182
329
    def test_stat(self):
183
330
        transport = MemoryTransport()
184
 
        transport.put('foo', StringIO('content'))
185
 
        transport.put('bar', StringIO('phowar'))
 
331
        transport.put_bytes('foo', 'content')
 
332
        transport.put_bytes('bar', 'phowar')
186
333
        self.assertEqual(7, transport.stat('foo').st_size)
187
334
        self.assertEqual(6, transport.stat('bar').st_size)
188
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()
189
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
 
190
425
class ReadonlyDecoratorTransportTest(TestCase):
191
426
    """Readonly decoration specific tests."""
192
427
 
195
430
        # connect to . in readonly mode
196
431
        transport = readonly.ReadonlyTransportDecorator('readonly+.')
197
432
        self.assertEqual(True, transport.listable())
198
 
        self.assertEqual(False, transport.should_cache())
199
433
        self.assertEqual(True, transport.is_readonly())
200
434
 
201
435
    def test_http_parameters(self):
 
436
        from bzrlib.tests.http_server import HttpServer
202
437
        import bzrlib.transport.readonly as readonly
203
 
        from bzrlib.transport.http import HttpServer
204
 
        # connect to . via http which is not listable
 
438
        # connect to '.' via http which is not listable
205
439
        server = HttpServer()
206
440
        server.setUp()
207
441
        try:
209
443
            self.failUnless(isinstance(transport,
210
444
                                       readonly.ReadonlyTransportDecorator))
211
445
            self.assertEqual(False, transport.listable())
212
 
            self.assertEqual(True, transport.should_cache())
213
446
            self.assertEqual(True, transport.is_readonly())
214
447
        finally:
215
448
            server.tearDown()
224
457
        return fakenfs.FakeNFSTransportDecorator('fakenfs+' + url)
225
458
 
226
459
    def test_local_parameters(self):
227
 
        # the listable, should_cache and is_readonly parameters
 
460
        # the listable and is_readonly parameters
228
461
        # are not changed by the fakenfs decorator
229
462
        transport = self.get_nfs_transport('.')
230
463
        self.assertEqual(True, transport.listable())
231
 
        self.assertEqual(False, transport.should_cache())
232
464
        self.assertEqual(False, transport.is_readonly())
233
465
 
234
466
    def test_http_parameters(self):
235
 
        # the listable, should_cache and is_readonly parameters
 
467
        # the listable and is_readonly parameters
236
468
        # are not changed by the fakenfs decorator
237
 
        from bzrlib.transport.http import HttpServer
238
 
        # 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
239
471
        server = HttpServer()
240
472
        server.setUp()
241
473
        try:
243
475
            self.assertIsInstance(
244
476
                transport, bzrlib.transport.fakenfs.FakeNFSTransportDecorator)
245
477
            self.assertEqual(False, transport.listable())
246
 
            self.assertEqual(True, transport.should_cache())
247
478
            self.assertEqual(True, transport.is_readonly())
248
479
        finally:
249
480
            server.tearDown()
254
485
        server = fakenfs.FakeNFSServer()
255
486
        server.setUp()
256
487
        try:
257
 
            # the server should be a relpath localhost server
258
 
            self.assertEqual(server.get_url(), 'fakenfs+.')
 
488
            # the url should be decorated appropriately
 
489
            self.assertStartsWith(server.get_url(), 'fakenfs+')
259
490
            # and we should be able to get a transport for it
260
491
            transport = get_transport(server.get_url())
261
492
            # which must be a FakeNFSTransportDecorator instance.
270
501
        transport = self.get_nfs_transport('.')
271
502
        self.build_tree(['from/', 'from/foo', 'to/', 'to/bar'],
272
503
                        transport=transport)
273
 
        self.assertRaises(bzrlib.errors.ResourceBusy,
 
504
        self.assertRaises(errors.ResourceBusy,
274
505
                          transport.rename, 'from', 'to')
275
506
 
276
507
 
306
537
class BackupTransportHandler(Transport):
307
538
    """Test transport that works as a backup for the BadTransportHandler"""
308
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)