/brz/remove-bazaar

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

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

Allow Repository tests to be backed onto a specific VFS as needed.

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Copyright (C) 2004, 2005, 2006 Canonical Ltd
 
2
#
 
3
# This program is free software; you can redistribute it and/or modify
 
4
# it under the terms of the GNU General Public License as published by
 
5
# the Free Software Foundation; either version 2 of the License, or
 
6
# (at your option) any later version.
 
7
#
 
8
# This program is distributed in the hope that it will be useful,
 
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
11
# GNU General Public License for more details.
 
12
#
 
13
# You should have received a copy of the GNU General Public License
 
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
 
19
import sys
 
20
import stat
 
21
from cStringIO import StringIO
 
22
 
 
23
import bzrlib
 
24
from bzrlib import urlutils
 
25
from bzrlib.errors import (
 
26
    ConnectionError,
 
27
    DependencyNotPresent,
 
28
    FileExists,
 
29
    NoSuchFile,
 
30
    PathNotChild,
 
31
    TransportNotPossible,
 
32
    UnsupportedProtocol,
 
33
    )
 
34
from bzrlib.tests import TestCase, TestCaseInTempDir
 
35
from bzrlib.transport import (_CoalescedOffset,
 
36
                              _get_protocol_handlers,
 
37
                              _get_transport_modules,
 
38
                              get_transport,
 
39
                              register_lazy_transport,
 
40
                              _set_protocol_handlers,
 
41
                              Transport,
 
42
                              )
 
43
from bzrlib.transport.memory import MemoryTransport
 
44
from bzrlib.transport.local import LocalTransport
 
45
 
 
46
 
 
47
# TODO: Should possibly split transport-specific tests into their own files.
 
48
 
 
49
 
 
50
class TestTransport(TestCase):
 
51
    """Test the non transport-concrete class functionality."""
 
52
 
 
53
    def test__get_set_protocol_handlers(self):
 
54
        handlers = _get_protocol_handlers()
 
55
        self.assertNotEqual({}, handlers)
 
56
        try:
 
57
            _set_protocol_handlers({})
 
58
            self.assertEqual({}, _get_protocol_handlers())
 
59
        finally:
 
60
            _set_protocol_handlers(handlers)
 
61
 
 
62
    def test_get_transport_modules(self):
 
63
        handlers = _get_protocol_handlers()
 
64
        class SampleHandler(object):
 
65
            """I exist, isnt that enough?"""
 
66
        try:
 
67
            my_handlers = {}
 
68
            _set_protocol_handlers(my_handlers)
 
69
            register_lazy_transport('foo', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
 
70
            register_lazy_transport('bar', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
 
71
            self.assertEqual([SampleHandler.__module__],
 
72
                             _get_transport_modules())
 
73
        finally:
 
74
            _set_protocol_handlers(handlers)
 
75
 
 
76
    def test_transport_dependency(self):
 
77
        """Transport with missing dependency causes no error"""
 
78
        saved_handlers = _get_protocol_handlers()
 
79
        try:
 
80
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
 
81
                    'BadTransportHandler')
 
82
            try:
 
83
                get_transport('foo://fooserver/foo')
 
84
            except UnsupportedProtocol, e:
 
85
                e_str = str(e)
 
86
                self.assertEquals('Unsupported protocol'
 
87
                                  ' for url "foo://fooserver/foo":'
 
88
                                  ' Unable to import library "some_lib":'
 
89
                                  ' testing missing dependency', str(e))
 
90
            else:
 
91
                self.fail('Did not raise UnsupportedProtocol')
 
92
        finally:
 
93
            # restore original values
 
94
            _set_protocol_handlers(saved_handlers)
 
95
            
 
96
    def test_transport_fallback(self):
 
97
        """Transport with missing dependency causes no error"""
 
98
        saved_handlers = _get_protocol_handlers()
 
99
        try:
 
100
            _set_protocol_handlers({})
 
101
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
 
102
                    'BackupTransportHandler')
 
103
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
 
104
                    'BadTransportHandler')
 
105
            t = get_transport('foo://fooserver/foo')
 
106
            self.assertTrue(isinstance(t, BackupTransportHandler))
 
107
        finally:
 
108
            _set_protocol_handlers(saved_handlers)
 
109
 
 
110
    def test__combine_paths(self):
 
111
        t = Transport('/')
 
112
        self.assertEqual('/home/sarah/project/foo',
 
113
                         t._combine_paths('/home/sarah', 'project/foo'))
 
114
        self.assertEqual('/etc',
 
115
                         t._combine_paths('/home/sarah', '../../etc'))
 
116
        self.assertEqual('/etc',
 
117
                         t._combine_paths('/home/sarah', '../../../etc'))
 
118
        self.assertEqual('/etc',
 
119
                         t._combine_paths('/home/sarah', '/etc'))
 
120
 
 
121
 
 
122
class TestCoalesceOffsets(TestCase):
 
123
    
 
124
    def check(self, expected, offsets, limit=0, fudge=0):
 
125
        coalesce = Transport._coalesce_offsets
 
126
        exp = [_CoalescedOffset(*x) for x in expected]
 
127
        out = list(coalesce(offsets, limit=limit, fudge_factor=fudge))
 
128
        self.assertEqual(exp, out)
 
129
 
 
130
    def test_coalesce_empty(self):
 
131
        self.check([], [])
 
132
 
 
133
    def test_coalesce_simple(self):
 
134
        self.check([(0, 10, [(0, 10)])], [(0, 10)])
 
135
 
 
136
    def test_coalesce_unrelated(self):
 
137
        self.check([(0, 10, [(0, 10)]),
 
138
                    (20, 10, [(0, 10)]),
 
139
                   ], [(0, 10), (20, 10)])
 
140
            
 
141
    def test_coalesce_unsorted(self):
 
142
        self.check([(20, 10, [(0, 10)]),
 
143
                    (0, 10, [(0, 10)]),
 
144
                   ], [(20, 10), (0, 10)])
 
145
 
 
146
    def test_coalesce_nearby(self):
 
147
        self.check([(0, 20, [(0, 10), (10, 10)])],
 
148
                   [(0, 10), (10, 10)])
 
149
 
 
150
    def test_coalesce_overlapped(self):
 
151
        self.check([(0, 15, [(0, 10), (5, 10)])],
 
152
                   [(0, 10), (5, 10)])
 
153
 
 
154
    def test_coalesce_limit(self):
 
155
        self.check([(10, 50, [(0, 10), (10, 10), (20, 10),
 
156
                              (30, 10), (40, 10)]),
 
157
                    (60, 50, [(0, 10), (10, 10), (20, 10),
 
158
                              (30, 10), (40, 10)]),
 
159
                   ], [(10, 10), (20, 10), (30, 10), (40, 10),
 
160
                       (50, 10), (60, 10), (70, 10), (80, 10),
 
161
                       (90, 10), (100, 10)],
 
162
                    limit=5)
 
163
 
 
164
    def test_coalesce_no_limit(self):
 
165
        self.check([(10, 100, [(0, 10), (10, 10), (20, 10),
 
166
                               (30, 10), (40, 10), (50, 10),
 
167
                               (60, 10), (70, 10), (80, 10),
 
168
                               (90, 10)]),
 
169
                   ], [(10, 10), (20, 10), (30, 10), (40, 10),
 
170
                       (50, 10), (60, 10), (70, 10), (80, 10),
 
171
                       (90, 10), (100, 10)])
 
172
 
 
173
    def test_coalesce_fudge(self):
 
174
        self.check([(10, 30, [(0, 10), (20, 10)]),
 
175
                    (100, 10, [(0, 10),]),
 
176
                   ], [(10, 10), (30, 10), (100, 10)],
 
177
                   fudge=10
 
178
                  )
 
179
 
 
180
 
 
181
class TestMemoryTransport(TestCase):
 
182
 
 
183
    def test_get_transport(self):
 
184
        MemoryTransport()
 
185
 
 
186
    def test_clone(self):
 
187
        transport = MemoryTransport()
 
188
        self.assertTrue(isinstance(transport, MemoryTransport))
 
189
        self.assertEqual("memory:///", transport.clone("/").base)
 
190
 
 
191
    def test_abspath(self):
 
192
        transport = MemoryTransport()
 
193
        self.assertEqual("memory:///relpath", transport.abspath('relpath'))
 
194
 
 
195
    def test_abspath_of_root(self):
 
196
        transport = MemoryTransport()
 
197
        self.assertEqual("memory:///", transport.base)
 
198
        self.assertEqual("memory:///", transport.abspath('/'))
 
199
 
 
200
    def test_abspath_of_relpath_starting_at_root(self):
 
201
        transport = MemoryTransport()
 
202
        self.assertEqual("memory:///foo", transport.abspath('/foo'))
 
203
 
 
204
    def test_append_and_get(self):
 
205
        transport = MemoryTransport()
 
206
        transport.append_bytes('path', 'content')
 
207
        self.assertEqual(transport.get('path').read(), 'content')
 
208
        transport.append_file('path', StringIO('content'))
 
209
        self.assertEqual(transport.get('path').read(), 'contentcontent')
 
210
 
 
211
    def test_put_and_get(self):
 
212
        transport = MemoryTransport()
 
213
        transport.put_file('path', StringIO('content'))
 
214
        self.assertEqual(transport.get('path').read(), 'content')
 
215
        transport.put_bytes('path', 'content')
 
216
        self.assertEqual(transport.get('path').read(), 'content')
 
217
 
 
218
    def test_append_without_dir_fails(self):
 
219
        transport = MemoryTransport()
 
220
        self.assertRaises(NoSuchFile,
 
221
                          transport.append_bytes, 'dir/path', 'content')
 
222
 
 
223
    def test_put_without_dir_fails(self):
 
224
        transport = MemoryTransport()
 
225
        self.assertRaises(NoSuchFile,
 
226
                          transport.put_file, 'dir/path', StringIO('content'))
 
227
 
 
228
    def test_get_missing(self):
 
229
        transport = MemoryTransport()
 
230
        self.assertRaises(NoSuchFile, transport.get, 'foo')
 
231
 
 
232
    def test_has_missing(self):
 
233
        transport = MemoryTransport()
 
234
        self.assertEquals(False, transport.has('foo'))
 
235
 
 
236
    def test_has_present(self):
 
237
        transport = MemoryTransport()
 
238
        transport.append_bytes('foo', 'content')
 
239
        self.assertEquals(True, transport.has('foo'))
 
240
 
 
241
    def test_list_dir(self):
 
242
        transport = MemoryTransport()
 
243
        transport.put_bytes('foo', 'content')
 
244
        transport.mkdir('dir')
 
245
        transport.put_bytes('dir/subfoo', 'content')
 
246
        transport.put_bytes('dirlike', 'content')
 
247
 
 
248
        self.assertEquals(['dir', 'dirlike', 'foo'], sorted(transport.list_dir('.')))
 
249
        self.assertEquals(['subfoo'], sorted(transport.list_dir('dir')))
 
250
 
 
251
    def test_mkdir(self):
 
252
        transport = MemoryTransport()
 
253
        transport.mkdir('dir')
 
254
        transport.append_bytes('dir/path', 'content')
 
255
        self.assertEqual(transport.get('dir/path').read(), 'content')
 
256
 
 
257
    def test_mkdir_missing_parent(self):
 
258
        transport = MemoryTransport()
 
259
        self.assertRaises(NoSuchFile,
 
260
                          transport.mkdir, 'dir/dir')
 
261
 
 
262
    def test_mkdir_twice(self):
 
263
        transport = MemoryTransport()
 
264
        transport.mkdir('dir')
 
265
        self.assertRaises(FileExists, transport.mkdir, 'dir')
 
266
 
 
267
    def test_parameters(self):
 
268
        transport = MemoryTransport()
 
269
        self.assertEqual(True, transport.listable())
 
270
        self.assertEqual(False, transport.should_cache())
 
271
        self.assertEqual(False, transport.is_readonly())
 
272
 
 
273
    def test_iter_files_recursive(self):
 
274
        transport = MemoryTransport()
 
275
        transport.mkdir('dir')
 
276
        transport.put_bytes('dir/foo', 'content')
 
277
        transport.put_bytes('dir/bar', 'content')
 
278
        transport.put_bytes('bar', 'content')
 
279
        paths = set(transport.iter_files_recursive())
 
280
        self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
 
281
 
 
282
    def test_stat(self):
 
283
        transport = MemoryTransport()
 
284
        transport.put_bytes('foo', 'content')
 
285
        transport.put_bytes('bar', 'phowar')
 
286
        self.assertEqual(7, transport.stat('foo').st_size)
 
287
        self.assertEqual(6, transport.stat('bar').st_size)
 
288
 
 
289
 
 
290
class ChrootDecoratorTransportTest(TestCase):
 
291
    """Chroot decoration specific tests."""
 
292
 
 
293
    def test_construct(self):
 
294
        from bzrlib.transport import chroot
 
295
        transport = chroot.ChrootTransportDecorator('chroot+memory:///pathA/')
 
296
        self.assertEqual('memory:///pathA/', transport.chroot_url)
 
297
        self.assertEqual('/', transport.chroot_relative)
 
298
        transport = chroot.ChrootTransportDecorator('chroot+memory:///pathA')
 
299
        self.assertEqual('memory:///pathA/', transport.chroot_url)
 
300
        self.assertEqual('/', transport.chroot_relative)
 
301
        transport = chroot.ChrootTransportDecorator(
 
302
            'chroot+memory:///path/B', chroot='memory:///path/')
 
303
        self.assertEqual('memory:///path/', transport.chroot_url)
 
304
        self.assertEqual('/B/', transport.chroot_relative)
 
305
 
 
306
    def test_abspath(self):
 
307
        # The abspath is always relative to the chroot_url.
 
308
        transport = get_transport('chroot+memory:///foo/bar/')
 
309
        self.assertEqual('chroot+memory:///foo/bar/', transport.abspath('/'))
 
310
 
 
311
        subdir_transport = transport.clone('subdir')
 
312
        self.assertEqual(
 
313
            'chroot+memory:///foo/bar/', subdir_transport.abspath('/'))
 
314
 
 
315
    def test_abspath_invalid(self):
 
316
        # You cannot have a url like "scheme:///chroot_root/..", which tries to
 
317
        # reference a location above chroot_url.
 
318
        transport = get_transport('chroot+memory:///foo/bar/')
 
319
 
 
320
        self.assertRaises(PathNotChild, transport.abspath, '..')
 
321
        self.assertRaises(PathNotChild, transport.abspath, '../..')
 
322
        self.assertRaises(PathNotChild, transport.abspath, 'foo/../..')
 
323
        self.assertRaises(PathNotChild, transport.abspath, '/..')
 
324
        self.assertRaises(PathNotChild, transport.abspath, '/foo/../..')
 
325
 
 
326
    def test_clone(self):
 
327
        transport = get_transport('chroot+memory:///foo/bar')
 
328
        # relpath from root and root path are the same
 
329
        relpath_cloned = transport.clone('foo')
 
330
        abspath_cloned = transport.clone('/foo')
 
331
        self.assertEqual(relpath_cloned.base, abspath_cloned.base)
 
332
        self.assertEqual(relpath_cloned.chroot_url, abspath_cloned.chroot_url)
 
333
        self.assertEqual(relpath_cloned.chroot_relative,
 
334
            abspath_cloned.chroot_relative)
 
335
        transport = transport.clone('subdir')
 
336
        # clone preserves chroot_url and adjusts chroot_relative
 
337
        self.assertEqual('memory:///foo/bar/', transport.chroot_url)
 
338
        self.assertEqual('/subdir/', transport.chroot_relative)
 
339
        transport = transport.clone('/otherdir')
 
340
        # clone preserves chroot_url and adjusts chroot_relative
 
341
        self.assertEqual('memory:///foo/bar/', transport.chroot_url)
 
342
        self.assertEqual('/otherdir/', transport.chroot_relative)
 
343
    
 
344
    def test_clone_to_root(self):
 
345
        # cloning to "/" (and similarly any offset beginning with "/") goes to
 
346
        # the chroot_url, not to root of the decorated transport.
 
347
        transport = get_transport('chroot+memory:///foo/bar/baz/')
 
348
        transport.clone('subdir')
 
349
        # now clone to "/" will take us back to the initial location, not to
 
350
        # "chroot_memory:///".
 
351
        transport.clone('/')
 
352
        self.assertEqual('chroot+memory:///foo/bar/baz/', transport.base)
 
353
 
 
354
    def test_clone_offset(self):
 
355
        # transport.clone('some offset') should call clone('some offset') on the
 
356
        # decorated transport, not some surprising variation like
 
357
        # ('/some offset').
 
358
        from bzrlib.transport import chroot
 
359
        decorated_transport = FakeTransport()
 
360
        transport = chroot.ChrootTransportDecorator(
 
361
            'chroot+fake:///', _decorated=decorated_transport)
 
362
        transport.clone('foo/bar')
 
363
        self.assertEqual([('clone', 'foo/bar')] , decorated_transport.calls)
 
364
 
 
365
    def test_clone_multiple_levels(self):
 
366
        url='chroot+memory:///hello/'
 
367
        transport = get_transport(url)
 
368
        transport = transport.clone("subdir1")
 
369
        transport = transport.clone("subdir2")
 
370
        self.assertEqual(
 
371
            'chroot+memory:///hello/subdir1/subdir2/', transport.base)
 
372
 
 
373
 
 
374
class FakeTransport(object):
 
375
    # XXX: FakeTransport copied from test_wsgi.py
 
376
 
 
377
    def __init__(self):
 
378
        self.calls = []
 
379
        self.base = 'fake:///'
 
380
 
 
381
    def abspath(self, relpath):
 
382
        return 'fake:///' + relpath
 
383
 
 
384
    def clone(self, relpath):
 
385
        self.calls.append(('clone', relpath))
 
386
        return self
 
387
 
 
388
 
 
389
class ReadonlyDecoratorTransportTest(TestCase):
 
390
    """Readonly decoration specific tests."""
 
391
 
 
392
    def test_local_parameters(self):
 
393
        import bzrlib.transport.readonly as readonly
 
394
        # connect to . in readonly mode
 
395
        transport = readonly.ReadonlyTransportDecorator('readonly+.')
 
396
        self.assertEqual(True, transport.listable())
 
397
        self.assertEqual(False, transport.should_cache())
 
398
        self.assertEqual(True, transport.is_readonly())
 
399
 
 
400
    def test_http_parameters(self):
 
401
        from bzrlib.tests.HttpServer import HttpServer
 
402
        import bzrlib.transport.readonly as readonly
 
403
        # connect to . via http which is not listable
 
404
        server = HttpServer()
 
405
        server.setUp()
 
406
        try:
 
407
            transport = get_transport('readonly+' + server.get_url())
 
408
            self.failUnless(isinstance(transport,
 
409
                                       readonly.ReadonlyTransportDecorator))
 
410
            self.assertEqual(False, transport.listable())
 
411
            self.assertEqual(True, transport.should_cache())
 
412
            self.assertEqual(True, transport.is_readonly())
 
413
        finally:
 
414
            server.tearDown()
 
415
 
 
416
 
 
417
class FakeNFSDecoratorTests(TestCaseInTempDir):
 
418
    """NFS decorator specific tests."""
 
419
 
 
420
    def get_nfs_transport(self, url):
 
421
        import bzrlib.transport.fakenfs as fakenfs
 
422
        # connect to url with nfs decoration
 
423
        return fakenfs.FakeNFSTransportDecorator('fakenfs+' + url)
 
424
 
 
425
    def test_local_parameters(self):
 
426
        # the listable, should_cache and is_readonly parameters
 
427
        # are not changed by the fakenfs decorator
 
428
        transport = self.get_nfs_transport('.')
 
429
        self.assertEqual(True, transport.listable())
 
430
        self.assertEqual(False, transport.should_cache())
 
431
        self.assertEqual(False, transport.is_readonly())
 
432
 
 
433
    def test_http_parameters(self):
 
434
        # the listable, should_cache and is_readonly parameters
 
435
        # are not changed by the fakenfs decorator
 
436
        from bzrlib.tests.HttpServer import HttpServer
 
437
        # connect to . via http which is not listable
 
438
        server = HttpServer()
 
439
        server.setUp()
 
440
        try:
 
441
            transport = self.get_nfs_transport(server.get_url())
 
442
            self.assertIsInstance(
 
443
                transport, bzrlib.transport.fakenfs.FakeNFSTransportDecorator)
 
444
            self.assertEqual(False, transport.listable())
 
445
            self.assertEqual(True, transport.should_cache())
 
446
            self.assertEqual(True, transport.is_readonly())
 
447
        finally:
 
448
            server.tearDown()
 
449
 
 
450
    def test_fakenfs_server_default(self):
 
451
        # a FakeNFSServer() should bring up a local relpath server for itself
 
452
        import bzrlib.transport.fakenfs as fakenfs
 
453
        server = fakenfs.FakeNFSServer()
 
454
        server.setUp()
 
455
        try:
 
456
            # the url should be decorated appropriately
 
457
            self.assertStartsWith(server.get_url(), 'fakenfs+')
 
458
            # and we should be able to get a transport for it
 
459
            transport = get_transport(server.get_url())
 
460
            # which must be a FakeNFSTransportDecorator instance.
 
461
            self.assertIsInstance(
 
462
                transport, fakenfs.FakeNFSTransportDecorator)
 
463
        finally:
 
464
            server.tearDown()
 
465
 
 
466
    def test_fakenfs_rename_semantics(self):
 
467
        # a FakeNFS transport must mangle the way rename errors occur to
 
468
        # look like NFS problems.
 
469
        transport = self.get_nfs_transport('.')
 
470
        self.build_tree(['from/', 'from/foo', 'to/', 'to/bar'],
 
471
                        transport=transport)
 
472
        self.assertRaises(bzrlib.errors.ResourceBusy,
 
473
                          transport.rename, 'from', 'to')
 
474
 
 
475
 
 
476
class FakeVFATDecoratorTests(TestCaseInTempDir):
 
477
    """Tests for simulation of VFAT restrictions"""
 
478
 
 
479
    def get_vfat_transport(self, url):
 
480
        """Return vfat-backed transport for test directory"""
 
481
        from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
 
482
        return FakeVFATTransportDecorator('vfat+' + url)
 
483
 
 
484
    def test_transport_creation(self):
 
485
        from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
 
486
        transport = self.get_vfat_transport('.')
 
487
        self.assertIsInstance(transport, FakeVFATTransportDecorator)
 
488
 
 
489
    def test_transport_mkdir(self):
 
490
        transport = self.get_vfat_transport('.')
 
491
        transport.mkdir('HELLO')
 
492
        self.assertTrue(transport.has('hello'))
 
493
        self.assertTrue(transport.has('Hello'))
 
494
 
 
495
    def test_forbidden_chars(self):
 
496
        transport = self.get_vfat_transport('.')
 
497
        self.assertRaises(ValueError, transport.has, "<NU>")
 
498
 
 
499
 
 
500
class BadTransportHandler(Transport):
 
501
    def __init__(self, base_url):
 
502
        raise DependencyNotPresent('some_lib', 'testing missing dependency')
 
503
 
 
504
 
 
505
class BackupTransportHandler(Transport):
 
506
    """Test transport that works as a backup for the BadTransportHandler"""
 
507
    pass
 
508
 
 
509
 
 
510
class TestTransportImplementation(TestCaseInTempDir):
 
511
    """Implementation verification for transports.
 
512
    
 
513
    To verify a transport we need a server factory, which is a callable
 
514
    that accepts no parameters and returns an implementation of
 
515
    bzrlib.transport.Server.
 
516
    
 
517
    That Server is then used to construct transport instances and test
 
518
    the transport via loopback activity.
 
519
 
 
520
    Currently this assumes that the Transport object is connected to the 
 
521
    current working directory.  So that whatever is done 
 
522
    through the transport, should show up in the working 
 
523
    directory, and vice-versa. This is a bug, because its possible to have
 
524
    URL schemes which provide access to something that may not be 
 
525
    result in storage on the local disk, i.e. due to file system limits, or 
 
526
    due to it being a database or some other non-filesystem tool.
 
527
 
 
528
    This also tests to make sure that the functions work with both
 
529
    generators and lists (assuming iter(list) is effectively a generator)
 
530
    """
 
531
    
 
532
    def setUp(self):
 
533
        super(TestTransportImplementation, self).setUp()
 
534
        self._server = self.transport_server()
 
535
        self._server.setUp()
 
536
 
 
537
    def tearDown(self):
 
538
        super(TestTransportImplementation, self).tearDown()
 
539
        self._server.tearDown()
 
540
        
 
541
    def get_transport(self):
 
542
        """Return a connected transport to the local directory."""
 
543
        base_url = self._server.get_url()
 
544
        # try getting the transport via the regular interface:
 
545
        t = get_transport(base_url)
 
546
        if not isinstance(t, self.transport_class):
 
547
            # we did not get the correct transport class type. Override the
 
548
            # regular connection behaviour by direct construction.
 
549
            t = self.transport_class(base_url)
 
550
        return t
 
551
 
 
552
 
 
553
class TestLocalTransports(TestCase):
 
554
 
 
555
    def test_get_transport_from_abspath(self):
 
556
        here = os.path.abspath('.')
 
557
        t = get_transport(here)
 
558
        self.assertIsInstance(t, LocalTransport)
 
559
        self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
 
560
 
 
561
    def test_get_transport_from_relpath(self):
 
562
        here = os.path.abspath('.')
 
563
        t = get_transport('.')
 
564
        self.assertIsInstance(t, LocalTransport)
 
565
        self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
 
566
 
 
567
    def test_get_transport_from_local_url(self):
 
568
        here = os.path.abspath('.')
 
569
        here_url = urlutils.local_path_to_url(here) + '/'
 
570
        t = get_transport(here_url)
 
571
        self.assertIsInstance(t, LocalTransport)
 
572
        self.assertEquals(t.base, here_url)