← 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: Martin Pool
  • Date: 2006-03-21 12:26:54 UTC
  • mto: This revision was merged to the branch mainline in revision 1621.
  • Revision ID: mbp@sourcefrog.net-20060321122654-514047ed65795a17
New developer commands 'weave-list' and 'weave-join'.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
1
 
# Copyright (C) 2004, 2005, 2006, 2007 Canonical Ltd
2
 
#
 
1
# Copyright (C) 2004, 2005, 2006 by 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
20
20
import stat
21
21
from cStringIO import StringIO
22
22
 
23
 
import bzrlib
24
 
from bzrlib import (
25
 
    errors,
26
 
    urlutils,
27
 
    )
28
 
from bzrlib.errors import (ConnectionError,
29
 
                           DependencyNotPresent,
30
 
                           FileExists,
31
 
                           InvalidURLJoin,
32
 
                           NoSuchFile,
33
 
                           PathNotChild,
 
23
from bzrlib.errors import (NoSuchFile, FileExists,
34
24
                           TransportNotPossible,
35
25
                           ConnectionError,
36
26
                           DependencyNotPresent,
37
 
                           ReadError,
38
 
                           UnsupportedProtocol,
39
27
                           )
40
 
from bzrlib.tests import TestCase, TestCaseInTempDir
41
 
from bzrlib.transport import (_CoalescedOffset,
42
 
                              ConnectedTransport,
43
 
                              _get_protocol_handlers,
44
 
                              _set_protocol_handlers,
 
28
from bzrlib.tests import TestCase
 
29
from bzrlib.transport import (_get_protocol_handlers,
45
30
                              _get_transport_modules,
46
31
                              get_transport,
47
 
                              LateReadError,
48
32
                              register_lazy_transport,
49
 
                              register_transport_proto,
50
 
                              _clear_protocol_handlers,
 
33
                              _set_protocol_handlers,
 
34
                              urlescape,
51
35
                              Transport,
52
36
                              )
53
 
from bzrlib.transport.chroot import ChrootServer
54
37
from bzrlib.transport.memory import MemoryTransport
55
 
from bzrlib.transport.local import (LocalTransport,
56
 
                                    EmulatedWin32LocalTransport)
57
 
 
58
 
 
59
 
# TODO: Should possibly split transport-specific tests into their own files.
 
38
from bzrlib.transport.local import LocalTransport
60
39
 
61
40
 
62
41
class TestTransport(TestCase):
63
42
    """Test the non transport-concrete class functionality."""
64
43
 
 
44
    def test_urlescape(self):
 
45
        self.assertEqual('%25', urlescape('%'))
 
46
 
65
47
    def test__get_set_protocol_handlers(self):
66
48
        handlers = _get_protocol_handlers()
67
 
        self.assertNotEqual([], handlers.keys( ))
 
49
        self.assertNotEqual({}, handlers)
68
50
        try:
69
 
            _clear_protocol_handlers()
70
 
            self.assertEqual([], _get_protocol_handlers().keys())
 
51
            _set_protocol_handlers({})
 
52
            self.assertEqual({}, _get_protocol_handlers())
71
53
        finally:
72
54
            _set_protocol_handlers(handlers)
73
55
 
76
58
        class SampleHandler(object):
77
59
            """I exist, isnt that enough?"""
78
60
        try:
79
 
            _clear_protocol_handlers()
80
 
            register_transport_proto('foo')
 
61
            my_handlers = {}
 
62
            _set_protocol_handlers(my_handlers)
81
63
            register_lazy_transport('foo', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
82
 
            register_transport_proto('bar')
83
64
            register_lazy_transport('bar', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
84
 
            self.assertEqual([SampleHandler.__module__, 'bzrlib.transport.chroot'],
 
65
            self.assertEqual([SampleHandler.__module__],
85
66
                             _get_transport_modules())
86
67
        finally:
87
68
            _set_protocol_handlers(handlers)
90
71
        """Transport with missing dependency causes no error"""
91
72
        saved_handlers = _get_protocol_handlers()
92
73
        try:
93
 
            register_transport_proto('foo')
94
74
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
95
75
                    'BadTransportHandler')
96
 
            try:
97
 
                get_transport('foo://fooserver/foo')
98
 
            except UnsupportedProtocol, e:
99
 
                e_str = str(e)
100
 
                self.assertEquals('Unsupported protocol'
101
 
                                  ' for url "foo://fooserver/foo":'
102
 
                                  ' Unable to import library "some_lib":'
103
 
                                  ' testing missing dependency', str(e))
104
 
            else:
105
 
                self.fail('Did not raise UnsupportedProtocol')
 
76
            t = get_transport('foo://fooserver/foo')
 
77
            # because we failed to load the transport
 
78
            self.assertTrue(isinstance(t, LocalTransport))
106
79
        finally:
107
80
            # restore original values
108
81
            _set_protocol_handlers(saved_handlers)
111
84
        """Transport with missing dependency causes no error"""
112
85
        saved_handlers = _get_protocol_handlers()
113
86
        try:
114
 
            _clear_protocol_handlers()
115
 
            register_transport_proto('foo')
 
87
            _set_protocol_handlers({})
116
88
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
117
89
                    'BackupTransportHandler')
118
90
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
121
93
            self.assertTrue(isinstance(t, BackupTransportHandler))
122
94
        finally:
123
95
            _set_protocol_handlers(saved_handlers)
124
 
 
125
 
    def test_LateReadError(self):
126
 
        """The LateReadError helper should raise on read()."""
127
 
        a_file = LateReadError('a path')
128
 
        try:
129
 
            a_file.read()
130
 
        except ReadError, error:
131
 
            self.assertEqual('a path', error.path)
132
 
        self.assertRaises(ReadError, a_file.read, 40)
133
 
        a_file.close()
134
 
 
135
 
    def test__combine_paths(self):
136
 
        t = Transport('/')
137
 
        self.assertEqual('/home/sarah/project/foo',
138
 
                         t._combine_paths('/home/sarah', 'project/foo'))
139
 
        self.assertEqual('/etc',
140
 
                         t._combine_paths('/home/sarah', '../../etc'))
141
 
        self.assertEqual('/etc',
142
 
                         t._combine_paths('/home/sarah', '../../../etc'))
143
 
        self.assertEqual('/etc',
144
 
                         t._combine_paths('/home/sarah', '/etc'))
145
 
 
146
 
    def test_local_abspath_non_local_transport(self):
147
 
        # the base implementation should throw
148
 
        t = MemoryTransport()
149
 
        e = self.assertRaises(errors.NotLocalUrl, t.local_abspath, 't')
150
 
        self.assertEqual('memory:///t is not a local path.', str(e))
151
 
 
152
 
 
153
 
class TestCoalesceOffsets(TestCase):
154
 
    
155
 
    def check(self, expected, offsets, limit=0, fudge=0):
156
 
        coalesce = Transport._coalesce_offsets
157
 
        exp = [_CoalescedOffset(*x) for x in expected]
158
 
        out = list(coalesce(offsets, limit=limit, fudge_factor=fudge))
159
 
        self.assertEqual(exp, out)
160
 
 
161
 
    def test_coalesce_empty(self):
162
 
        self.check([], [])
163
 
 
164
 
    def test_coalesce_simple(self):
165
 
        self.check([(0, 10, [(0, 10)])], [(0, 10)])
166
 
 
167
 
    def test_coalesce_unrelated(self):
168
 
        self.check([(0, 10, [(0, 10)]),
169
 
                    (20, 10, [(0, 10)]),
170
 
                   ], [(0, 10), (20, 10)])
171
96
            
172
 
    def test_coalesce_unsorted(self):
173
 
        self.check([(20, 10, [(0, 10)]),
174
 
                    (0, 10, [(0, 10)]),
175
 
                   ], [(20, 10), (0, 10)])
176
 
 
177
 
    def test_coalesce_nearby(self):
178
 
        self.check([(0, 20, [(0, 10), (10, 10)])],
179
 
                   [(0, 10), (10, 10)])
180
 
 
181
 
    def test_coalesce_overlapped(self):
182
 
        self.check([(0, 15, [(0, 10), (5, 10)])],
183
 
                   [(0, 10), (5, 10)])
184
 
 
185
 
    def test_coalesce_limit(self):
186
 
        self.check([(10, 50, [(0, 10), (10, 10), (20, 10),
187
 
                              (30, 10), (40, 10)]),
188
 
                    (60, 50, [(0, 10), (10, 10), (20, 10),
189
 
                              (30, 10), (40, 10)]),
190
 
                   ], [(10, 10), (20, 10), (30, 10), (40, 10),
191
 
                       (50, 10), (60, 10), (70, 10), (80, 10),
192
 
                       (90, 10), (100, 10)],
193
 
                    limit=5)
194
 
 
195
 
    def test_coalesce_no_limit(self):
196
 
        self.check([(10, 100, [(0, 10), (10, 10), (20, 10),
197
 
                               (30, 10), (40, 10), (50, 10),
198
 
                               (60, 10), (70, 10), (80, 10),
199
 
                               (90, 10)]),
200
 
                   ], [(10, 10), (20, 10), (30, 10), (40, 10),
201
 
                       (50, 10), (60, 10), (70, 10), (80, 10),
202
 
                       (90, 10), (100, 10)])
203
 
 
204
 
    def test_coalesce_fudge(self):
205
 
        self.check([(10, 30, [(0, 10), (20, 10)]),
206
 
                    (100, 10, [(0, 10),]),
207
 
                   ], [(10, 10), (30, 10), (100, 10)],
208
 
                   fudge=10
209
 
                  )
210
 
 
211
97
 
212
98
class TestMemoryTransport(TestCase):
213
99
 
217
103
    def test_clone(self):
218
104
        transport = MemoryTransport()
219
105
        self.assertTrue(isinstance(transport, MemoryTransport))
220
 
        self.assertEqual("memory:///", transport.clone("/").base)
221
106
 
222
107
    def test_abspath(self):
223
108
        transport = MemoryTransport()
224
 
        self.assertEqual("memory:///relpath", transport.abspath('relpath'))
225
 
 
226
 
    def test_abspath_of_root(self):
227
 
        transport = MemoryTransport()
228
 
        self.assertEqual("memory:///", transport.base)
229
 
        self.assertEqual("memory:///", transport.abspath('/'))
230
 
 
231
 
    def test_abspath_of_relpath_starting_at_root(self):
232
 
        transport = MemoryTransport()
233
 
        self.assertEqual("memory:///foo", transport.abspath('/foo'))
 
109
        self.assertEqual("memory:/relpath", transport.abspath('relpath'))
 
110
 
 
111
    def test_relpath(self):
 
112
        transport = MemoryTransport()
234
113
 
235
114
    def test_append_and_get(self):
236
115
        transport = MemoryTransport()
237
 
        transport.append_bytes('path', 'content')
 
116
        transport.append('path', StringIO('content'))
238
117
        self.assertEqual(transport.get('path').read(), 'content')
239
 
        transport.append_file('path', StringIO('content'))
 
118
        transport.append('path', StringIO('content'))
240
119
        self.assertEqual(transport.get('path').read(), 'contentcontent')
241
120
 
242
121
    def test_put_and_get(self):
243
122
        transport = MemoryTransport()
244
 
        transport.put_file('path', StringIO('content'))
 
123
        transport.put('path', StringIO('content'))
245
124
        self.assertEqual(transport.get('path').read(), 'content')
246
 
        transport.put_bytes('path', 'content')
 
125
        transport.put('path', StringIO('content'))
247
126
        self.assertEqual(transport.get('path').read(), 'content')
248
127
 
249
128
    def test_append_without_dir_fails(self):
250
129
        transport = MemoryTransport()
251
130
        self.assertRaises(NoSuchFile,
252
 
                          transport.append_bytes, 'dir/path', 'content')
 
131
                          transport.append, 'dir/path', StringIO('content'))
253
132
 
254
133
    def test_put_without_dir_fails(self):
255
134
        transport = MemoryTransport()
256
135
        self.assertRaises(NoSuchFile,
257
 
                          transport.put_file, 'dir/path', StringIO('content'))
 
136
                          transport.put, 'dir/path', StringIO('content'))
258
137
 
259
138
    def test_get_missing(self):
260
139
        transport = MemoryTransport()
266
145
 
267
146
    def test_has_present(self):
268
147
        transport = MemoryTransport()
269
 
        transport.append_bytes('foo', 'content')
 
148
        transport.append('foo', StringIO('content'))
270
149
        self.assertEquals(True, transport.has('foo'))
271
150
 
272
 
    def test_list_dir(self):
273
 
        transport = MemoryTransport()
274
 
        transport.put_bytes('foo', 'content')
275
 
        transport.mkdir('dir')
276
 
        transport.put_bytes('dir/subfoo', 'content')
277
 
        transport.put_bytes('dirlike', 'content')
278
 
 
279
 
        self.assertEquals(['dir', 'dirlike', 'foo'], sorted(transport.list_dir('.')))
280
 
        self.assertEquals(['subfoo'], sorted(transport.list_dir('dir')))
281
 
 
282
151
    def test_mkdir(self):
283
152
        transport = MemoryTransport()
284
153
        transport.mkdir('dir')
285
 
        transport.append_bytes('dir/path', 'content')
 
154
        transport.append('dir/path', StringIO('content'))
286
155
        self.assertEqual(transport.get('dir/path').read(), 'content')
287
156
 
288
157
    def test_mkdir_missing_parent(self):
304
173
    def test_iter_files_recursive(self):
305
174
        transport = MemoryTransport()
306
175
        transport.mkdir('dir')
307
 
        transport.put_bytes('dir/foo', 'content')
308
 
        transport.put_bytes('dir/bar', 'content')
309
 
        transport.put_bytes('bar', 'content')
 
176
        transport.put('dir/foo', StringIO('content'))
 
177
        transport.put('dir/bar', StringIO('content'))
 
178
        transport.put('bar', StringIO('content'))
310
179
        paths = set(transport.iter_files_recursive())
311
180
        self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
312
181
 
313
182
    def test_stat(self):
314
183
        transport = MemoryTransport()
315
 
        transport.put_bytes('foo', 'content')
316
 
        transport.put_bytes('bar', 'phowar')
 
184
        transport.put('foo', StringIO('content'))
 
185
        transport.put('bar', StringIO('phowar'))
317
186
        self.assertEqual(7, transport.stat('foo').st_size)
318
187
        self.assertEqual(6, transport.stat('bar').st_size)
319
188
 
320
 
 
321
 
class ChrootDecoratorTransportTest(TestCase):
322
 
    """Chroot decoration specific tests."""
323
 
 
324
 
    def test_abspath(self):
325
 
        # The abspath is always relative to the chroot_url.
326
 
        server = ChrootServer(get_transport('memory:///foo/bar/'))
327
 
        server.setUp()
328
 
        transport = get_transport(server.get_url())
329
 
        self.assertEqual(server.get_url(), transport.abspath('/'))
330
 
 
331
 
        subdir_transport = transport.clone('subdir')
332
 
        self.assertEqual(server.get_url(), subdir_transport.abspath('/'))
333
 
        server.tearDown()
334
 
 
335
 
    def test_clone(self):
336
 
        server = ChrootServer(get_transport('memory:///foo/bar/'))
337
 
        server.setUp()
338
 
        transport = get_transport(server.get_url())
339
 
        # relpath from root and root path are the same
340
 
        relpath_cloned = transport.clone('foo')
341
 
        abspath_cloned = transport.clone('/foo')
342
 
        self.assertEqual(server, relpath_cloned.server)
343
 
        self.assertEqual(server, abspath_cloned.server)
344
 
        server.tearDown()
345
 
    
346
 
    def test_chroot_url_preserves_chroot(self):
347
 
        """Calling get_transport on a chroot transport's base should produce a
348
 
        transport with exactly the same behaviour as the original chroot
349
 
        transport.
350
 
 
351
 
        This is so that it is not possible to escape a chroot by doing::
352
 
            url = chroot_transport.base
353
 
            parent_url = urlutils.join(url, '..')
354
 
            new_transport = get_transport(parent_url)
355
 
        """
356
 
        server = ChrootServer(get_transport('memory:///path/subpath'))
357
 
        server.setUp()
358
 
        transport = get_transport(server.get_url())
359
 
        new_transport = get_transport(transport.base)
360
 
        self.assertEqual(transport.server, new_transport.server)
361
 
        self.assertEqual(transport.base, new_transport.base)
362
 
        server.tearDown()
363
189
        
364
 
    def test_urljoin_preserves_chroot(self):
365
 
        """Using urlutils.join(url, '..') on a chroot URL should not produce a
366
 
        URL that escapes the intended chroot.
367
 
 
368
 
        This is so that it is not possible to escape a chroot by doing::
369
 
            url = chroot_transport.base
370
 
            parent_url = urlutils.join(url, '..')
371
 
            new_transport = get_transport(parent_url)
372
 
        """
373
 
        server = ChrootServer(get_transport('memory:///path/'))
374
 
        server.setUp()
375
 
        transport = get_transport(server.get_url())
376
 
        self.assertRaises(
377
 
            InvalidURLJoin, urlutils.join, transport.base, '..')
378
 
        server.tearDown()
379
 
 
380
 
 
381
 
class ChrootServerTest(TestCase):
382
 
 
383
 
    def test_construct(self):
384
 
        backing_transport = MemoryTransport()
385
 
        server = ChrootServer(backing_transport)
386
 
        self.assertEqual(backing_transport, server.backing_transport)
387
 
 
388
 
    def test_setUp(self):
389
 
        backing_transport = MemoryTransport()
390
 
        server = ChrootServer(backing_transport)
391
 
        server.setUp()
392
 
        self.assertTrue(server.scheme in _get_protocol_handlers().keys())
393
 
 
394
 
    def test_tearDown(self):
395
 
        backing_transport = MemoryTransport()
396
 
        server = ChrootServer(backing_transport)
397
 
        server.setUp()
398
 
        server.tearDown()
399
 
        self.assertFalse(server.scheme in _get_protocol_handlers().keys())
400
 
 
401
 
    def test_get_url(self):
402
 
        backing_transport = MemoryTransport()
403
 
        server = ChrootServer(backing_transport)
404
 
        server.setUp()
405
 
        self.assertEqual('chroot-%d:///' % id(server), server.get_url())
406
 
        server.tearDown()
407
 
 
408
 
 
409
190
class ReadonlyDecoratorTransportTest(TestCase):
410
191
    """Readonly decoration specific tests."""
411
192
 
418
199
        self.assertEqual(True, transport.is_readonly())
419
200
 
420
201
    def test_http_parameters(self):
421
 
        from bzrlib.tests.HttpServer import HttpServer
422
202
        import bzrlib.transport.readonly as readonly
 
203
        from bzrlib.transport.http import HttpServer
423
204
        # connect to . via http which is not listable
424
205
        server = HttpServer()
425
206
        server.setUp()
434
215
            server.tearDown()
435
216
 
436
217
 
437
 
class FakeNFSDecoratorTests(TestCaseInTempDir):
438
 
    """NFS decorator specific tests."""
439
 
 
440
 
    def get_nfs_transport(self, url):
441
 
        import bzrlib.transport.fakenfs as fakenfs
442
 
        # connect to url with nfs decoration
443
 
        return fakenfs.FakeNFSTransportDecorator('fakenfs+' + url)
444
 
 
445
 
    def test_local_parameters(self):
446
 
        # the listable, should_cache and is_readonly parameters
447
 
        # are not changed by the fakenfs decorator
448
 
        transport = self.get_nfs_transport('.')
449
 
        self.assertEqual(True, transport.listable())
450
 
        self.assertEqual(False, transport.should_cache())
451
 
        self.assertEqual(False, transport.is_readonly())
452
 
 
453
 
    def test_http_parameters(self):
454
 
        # the listable, should_cache and is_readonly parameters
455
 
        # are not changed by the fakenfs decorator
456
 
        from bzrlib.tests.HttpServer import HttpServer
457
 
        # connect to . via http which is not listable
458
 
        server = HttpServer()
459
 
        server.setUp()
460
 
        try:
461
 
            transport = self.get_nfs_transport(server.get_url())
462
 
            self.assertIsInstance(
463
 
                transport, bzrlib.transport.fakenfs.FakeNFSTransportDecorator)
464
 
            self.assertEqual(False, transport.listable())
465
 
            self.assertEqual(True, transport.should_cache())
466
 
            self.assertEqual(True, transport.is_readonly())
467
 
        finally:
468
 
            server.tearDown()
469
 
 
470
 
    def test_fakenfs_server_default(self):
471
 
        # a FakeNFSServer() should bring up a local relpath server for itself
472
 
        import bzrlib.transport.fakenfs as fakenfs
473
 
        server = fakenfs.FakeNFSServer()
474
 
        server.setUp()
475
 
        try:
476
 
            # the url should be decorated appropriately
477
 
            self.assertStartsWith(server.get_url(), 'fakenfs+')
478
 
            # and we should be able to get a transport for it
479
 
            transport = get_transport(server.get_url())
480
 
            # which must be a FakeNFSTransportDecorator instance.
481
 
            self.assertIsInstance(
482
 
                transport, fakenfs.FakeNFSTransportDecorator)
483
 
        finally:
484
 
            server.tearDown()
485
 
 
486
 
    def test_fakenfs_rename_semantics(self):
487
 
        # a FakeNFS transport must mangle the way rename errors occur to
488
 
        # look like NFS problems.
489
 
        transport = self.get_nfs_transport('.')
490
 
        self.build_tree(['from/', 'from/foo', 'to/', 'to/bar'],
491
 
                        transport=transport)
492
 
        self.assertRaises(errors.ResourceBusy,
493
 
                          transport.rename, 'from', 'to')
494
 
 
495
 
 
496
 
class FakeVFATDecoratorTests(TestCaseInTempDir):
497
 
    """Tests for simulation of VFAT restrictions"""
498
 
 
499
 
    def get_vfat_transport(self, url):
500
 
        """Return vfat-backed transport for test directory"""
501
 
        from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
502
 
        return FakeVFATTransportDecorator('vfat+' + url)
503
 
 
504
 
    def test_transport_creation(self):
505
 
        from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
506
 
        transport = self.get_vfat_transport('.')
507
 
        self.assertIsInstance(transport, FakeVFATTransportDecorator)
508
 
 
509
 
    def test_transport_mkdir(self):
510
 
        transport = self.get_vfat_transport('.')
511
 
        transport.mkdir('HELLO')
512
 
        self.assertTrue(transport.has('hello'))
513
 
        self.assertTrue(transport.has('Hello'))
514
 
 
515
 
    def test_forbidden_chars(self):
516
 
        transport = self.get_vfat_transport('.')
517
 
        self.assertRaises(ValueError, transport.has, "<NU>")
518
 
 
519
 
 
520
218
class BadTransportHandler(Transport):
521
219
    def __init__(self, base_url):
522
220
        raise DependencyNotPresent('some_lib', 'testing missing dependency')
525
223
class BackupTransportHandler(Transport):
526
224
    """Test transport that works as a backup for the BadTransportHandler"""
527
225
    pass
528
 
 
529
 
 
530
 
class TestTransportImplementation(TestCaseInTempDir):
531
 
    """Implementation verification for transports.
532
 
    
533
 
    To verify a transport we need a server factory, which is a callable
534
 
    that accepts no parameters and returns an implementation of
535
 
    bzrlib.transport.Server.
536
 
    
537
 
    That Server is then used to construct transport instances and test
538
 
    the transport via loopback activity.
539
 
 
540
 
    Currently this assumes that the Transport object is connected to the 
541
 
    current working directory.  So that whatever is done 
542
 
    through the transport, should show up in the working 
543
 
    directory, and vice-versa. This is a bug, because its possible to have
544
 
    URL schemes which provide access to something that may not be 
545
 
    result in storage on the local disk, i.e. due to file system limits, or 
546
 
    due to it being a database or some other non-filesystem tool.
547
 
 
548
 
    This also tests to make sure that the functions work with both
549
 
    generators and lists (assuming iter(list) is effectively a generator)
550
 
    """
551
 
    
552
 
    def setUp(self):
553
 
        super(TestTransportImplementation, self).setUp()
554
 
        self._server = self.transport_server()
555
 
        self._server.setUp()
556
 
        self.addCleanup(self._server.tearDown)
557
 
 
558
 
    def get_transport(self, relpath=None):
559
 
        """Return a connected transport to the local directory.
560
 
 
561
 
        :param relpath: a path relative to the base url.
562
 
        """
563
 
        base_url = self._server.get_url()
564
 
        url = self._adjust_url(base_url, relpath)
565
 
        # try getting the transport via the regular interface:
566
 
        t = get_transport(url)
567
 
        # vila--20070607 if the following are commented out the test suite
568
 
        # still pass. Is this really still needed or was it a forgotten
569
 
        # temporary fix ?
570
 
        if not isinstance(t, self.transport_class):
571
 
            # we did not get the correct transport class type. Override the
572
 
            # regular connection behaviour by direct construction.
573
 
            t = self.transport_class(url)
574
 
        return t
575
 
 
576
 
 
577
 
class TestLocalTransports(TestCase):
578
 
 
579
 
    def test_get_transport_from_abspath(self):
580
 
        here = os.path.abspath('.')
581
 
        t = get_transport(here)
582
 
        self.assertIsInstance(t, LocalTransport)
583
 
        self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
584
 
 
585
 
    def test_get_transport_from_relpath(self):
586
 
        here = os.path.abspath('.')
587
 
        t = get_transport('.')
588
 
        self.assertIsInstance(t, LocalTransport)
589
 
        self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
590
 
 
591
 
    def test_get_transport_from_local_url(self):
592
 
        here = os.path.abspath('.')
593
 
        here_url = urlutils.local_path_to_url(here) + '/'
594
 
        t = get_transport(here_url)
595
 
        self.assertIsInstance(t, LocalTransport)
596
 
        self.assertEquals(t.base, here_url)
597
 
 
598
 
    def test_local_abspath(self):
599
 
        here = os.path.abspath('.')
600
 
        t = get_transport(here)
601
 
        self.assertEquals(t.local_abspath(''), here)
602
 
 
603
 
 
604
 
class TestWin32LocalTransport(TestCase):
605
 
 
606
 
    def test_unc_clone_to_root(self):
607
 
        # Win32 UNC path like \\HOST\path
608
 
        # clone to root should stop at least at \\HOST part
609
 
        # not on \\
610
 
        t = EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
611
 
        for i in xrange(4):
612
 
            t = t.clone('..')
613
 
        self.assertEquals(t.base, 'file://HOST/')
614
 
        # make sure we reach the root
615
 
        t = t.clone('..')
616
 
        self.assertEquals(t.base, 'file://HOST/')
617
 
 
618
 
class TestConnectedTransport(TestCase):
619
 
    """Tests for connected to remote server transports"""
620
 
 
621
 
    def test_parse_url(self):
622
 
        t = ConnectedTransport('sftp://simple.example.com/home/source')
623
 
        self.assertEquals(t._host, 'simple.example.com')
624
 
        self.assertEquals(t._port, None)
625
 
        self.assertEquals(t._path, '/home/source/')
626
 
        self.failUnless(t._user is None)
627
 
        self.failUnless(t._password is None)
628
 
 
629
 
        self.assertEquals(t.base, 'sftp://simple.example.com/home/source/')
630
 
 
631
 
    def test_parse_quoted_url(self):
632
 
        t = ConnectedTransport('http://ro%62ey:h%40t@ex%41mple.com:2222/path')
633
 
        self.assertEquals(t._host, 'exAmple.com')
634
 
        self.assertEquals(t._port, 2222)
635
 
        self.assertEquals(t._user, 'robey')
636
 
        self.assertEquals(t._password, 'h@t')
637
 
        self.assertEquals(t._path, '/path/')
638
 
 
639
 
        # Base should not keep track of the password
640
 
        self.assertEquals(t.base, 'http://robey@exAmple.com:2222/path/')
641
 
 
642
 
    def test_parse_invalid_url(self):
643
 
        self.assertRaises(errors.InvalidURL,
644
 
                          ConnectedTransport,
645
 
                          'sftp://lily.org:~janneke/public/bzr/gub')
646
 
 
647
 
    def test_relpath(self):
648
 
        t = ConnectedTransport('sftp://user@host.com/abs/path')
649
 
 
650
 
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
651
 
        self.assertRaises(errors.PathNotChild, t.relpath,
652
 
                          'http://user@host.com/abs/path/sub')
653
 
        self.assertRaises(errors.PathNotChild, t.relpath,
654
 
                          'sftp://user2@host.com/abs/path/sub')
655
 
        self.assertRaises(errors.PathNotChild, t.relpath,
656
 
                          'sftp://user@otherhost.com/abs/path/sub')
657
 
        self.assertRaises(errors.PathNotChild, t.relpath,
658
 
                          'sftp://user@host.com:33/abs/path/sub')
659
 
        # Make sure it works when we don't supply a username
660
 
        t = ConnectedTransport('sftp://host.com/abs/path')
661
 
        self.assertEquals(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
662
 
 
663
 
        # Make sure it works when parts of the path will be url encoded
664
 
        t = ConnectedTransport('sftp://host.com/dev/%path')
665
 
        self.assertEquals(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
666
 
 
667
 
    def test_connection_sharing_propagate_credentials(self):
668
 
        t = ConnectedTransport('foo://user@host.com/abs/path')
669
 
        self.assertIs(None, t._get_connection())
670
 
        self.assertIs(None, t._password)
671
 
        c = t.clone('subdir')
672
 
        self.assertEquals(None, c._get_connection())
673
 
        self.assertIs(None, t._password)
674
 
 
675
 
        # Simulate the user entering a password
676
 
        password = 'secret'
677
 
        connection = object()
678
 
        t._set_connection(connection, password)
679
 
        self.assertIs(connection, t._get_connection())
680
 
        self.assertIs(password, t._get_credentials())
681
 
        self.assertIs(connection, c._get_connection())
682
 
        self.assertIs(password, c._get_credentials())
683
 
 
684
 
        # credentials can be updated
685
 
        new_password = 'even more secret'
686
 
        c._update_credentials(new_password)
687
 
        self.assertIs(connection, t._get_connection())
688
 
        self.assertIs(new_password, t._get_credentials())
689
 
        self.assertIs(connection, c._get_connection())
690
 
        self.assertIs(new_password, c._get_credentials())
691
 
 
692
 
 
693
 
class TestReusedTransports(TestCase):
694
 
    """Tests for transport reuse"""
695
 
 
696
 
    def test_reuse_same_transport(self):
697
 
        t1 = get_transport('http://foo/')
698
 
        t2 = get_transport('http://foo/', possible_transports=[t1])
699
 
        self.assertIs(t1, t2)
700
 
 
701
 
        # Also check that final '/' are handled correctly
702
 
        t3 = get_transport('http://foo/path/')
703
 
        t4 = get_transport('http://foo/path', possible_transports=[t3])
704
 
        self.assertIs(t3, t4)
705
 
 
706
 
        t5 = get_transport('http://foo/path')
707
 
        t6 = get_transport('http://foo/path/', possible_transports=[t5])
708
 
        self.assertIs(t5, t6)
709
 
 
710
 
    def test_don_t_reuse_different_transport(self):
711
 
        t1 = get_transport('http://foo/path')
712
 
        t2 = get_transport('http://bar/path', possible_transports=[t1])
713
 
        self.assertIsNot(t1, t2)
714
 
 
715
 
 
716
 
def get_test_permutations():
717
 
    """Return transport permutations to be used in testing.
718
 
 
719
 
    This module registers some transports, but they're only for testing
720
 
    registration.  We don't really want to run all the transport tests against
721
 
    them.
722
 
    """
723
 
    return []