~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

  • Committer: Martin Pool
  • Date: 2005-05-06 02:34:54 UTC
  • Revision ID: mbp@sourcefrog.net-20050506023454-7118a1b22e8515bc
- ignore any diff files lying around in tree

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2004, 2005, 2006, 2007 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
 
from cStringIO import StringIO
19
 
 
20
 
import bzrlib
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,
33
 
                           )
34
 
from bzrlib.tests import TestCase, TestCaseInTempDir
35
 
from bzrlib.transport import (_clear_protocol_handlers,
36
 
                              _CoalescedOffset,
37
 
                              ConnectedTransport,
38
 
                              _get_protocol_handlers,
39
 
                              _set_protocol_handlers,
40
 
                              _get_transport_modules,
41
 
                              get_transport,
42
 
                              LateReadError,
43
 
                              register_lazy_transport,
44
 
                              register_transport_proto,
45
 
                              Transport,
46
 
                              )
47
 
from bzrlib.transport.chroot import ChrootServer
48
 
from bzrlib.transport.memory import MemoryTransport
49
 
from bzrlib.transport.local import (LocalTransport,
50
 
                                    EmulatedWin32LocalTransport)
51
 
 
52
 
 
53
 
# TODO: Should possibly split transport-specific tests into their own files.
54
 
 
55
 
 
56
 
class TestTransport(TestCase):
57
 
    """Test the non transport-concrete class functionality."""
58
 
 
59
 
    def test__get_set_protocol_handlers(self):
60
 
        handlers = _get_protocol_handlers()
61
 
        self.assertNotEqual([], handlers.keys( ))
62
 
        try:
63
 
            _clear_protocol_handlers()
64
 
            self.assertEqual([], _get_protocol_handlers().keys())
65
 
        finally:
66
 
            _set_protocol_handlers(handlers)
67
 
 
68
 
    def test_get_transport_modules(self):
69
 
        handlers = _get_protocol_handlers()
70
 
        # don't pollute the current handlers
71
 
        _clear_protocol_handlers()
72
 
        class SampleHandler(object):
73
 
            """I exist, isnt that enough?"""
74
 
        try:
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'],
84
 
                             _get_transport_modules())
85
 
        finally:
86
 
            _set_protocol_handlers(handlers)
87
 
 
88
 
    def test_transport_dependency(self):
89
 
        """Transport with missing dependency causes no error"""
90
 
        saved_handlers = _get_protocol_handlers()
91
 
        # don't pollute the current handlers
92
 
        _clear_protocol_handlers()
93
 
        try:
94
 
            register_transport_proto('foo')
95
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
96
 
                    'BadTransportHandler')
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')
107
 
        finally:
108
 
            # restore original values
109
 
            _set_protocol_handlers(saved_handlers)
110
 
            
111
 
    def test_transport_fallback(self):
112
 
        """Transport with missing dependency causes no error"""
113
 
        saved_handlers = _get_protocol_handlers()
114
 
        try:
115
 
            _clear_protocol_handlers()
116
 
            register_transport_proto('foo')
117
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
118
 
                    'BackupTransportHandler')
119
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
120
 
                    'BadTransportHandler')
121
 
            t = get_transport('foo://fooserver/foo')
122
 
            self.assertTrue(isinstance(t, BackupTransportHandler))
123
 
        finally:
124
 
            _set_protocol_handlers(saved_handlers)
125
 
 
126
 
    def test_LateReadError(self):
127
 
        """The LateReadError helper should raise on read()."""
128
 
        a_file = LateReadError('a path')
129
 
        try:
130
 
            a_file.read()
131
 
        except ReadError, error:
132
 
            self.assertEqual('a path', error.path)
133
 
        self.assertRaises(ReadError, a_file.read, 40)
134
 
        a_file.close()
135
 
 
136
 
    def test__combine_paths(self):
137
 
        t = Transport('/')
138
 
        self.assertEqual('/home/sarah/project/foo',
139
 
                         t._combine_paths('/home/sarah', 'project/foo'))
140
 
        self.assertEqual('/etc',
141
 
                         t._combine_paths('/home/sarah', '../../etc'))
142
 
        self.assertEqual('/etc',
143
 
                         t._combine_paths('/home/sarah', '../../../etc'))
144
 
        self.assertEqual('/etc',
145
 
                         t._combine_paths('/home/sarah', '/etc'))
146
 
 
147
 
    def test_local_abspath_non_local_transport(self):
148
 
        # the base implementation should throw
149
 
        t = MemoryTransport()
150
 
        e = self.assertRaises(errors.NotLocalUrl, t.local_abspath, 't')
151
 
        self.assertEqual('memory:///t is not a local path.', str(e))
152
 
 
153
 
 
154
 
class TestCoalesceOffsets(TestCase):
155
 
 
156
 
    def check(self, expected, offsets, limit=0, max_size=0, fudge=0):
157
 
        coalesce = Transport._coalesce_offsets
158
 
        exp = [_CoalescedOffset(*x) for x in expected]
159
 
        out = list(coalesce(offsets, limit=limit, fudge_factor=fudge,
160
 
                            max_size=max_size))
161
 
        self.assertEqual(exp, out)
162
 
 
163
 
    def test_coalesce_empty(self):
164
 
        self.check([], [])
165
 
 
166
 
    def test_coalesce_simple(self):
167
 
        self.check([(0, 10, [(0, 10)])], [(0, 10)])
168
 
 
169
 
    def test_coalesce_unrelated(self):
170
 
        self.check([(0, 10, [(0, 10)]),
171
 
                    (20, 10, [(0, 10)]),
172
 
                   ], [(0, 10), (20, 10)])
173
 
 
174
 
    def test_coalesce_unsorted(self):
175
 
        self.check([(20, 10, [(0, 10)]),
176
 
                    (0, 10, [(0, 10)]),
177
 
                   ], [(20, 10), (0, 10)])
178
 
 
179
 
    def test_coalesce_nearby(self):
180
 
        self.check([(0, 20, [(0, 10), (10, 10)])],
181
 
                   [(0, 10), (10, 10)])
182
 
 
183
 
    def test_coalesce_overlapped(self):
184
 
        self.assertRaises(ValueError,
185
 
            self.check, [(0, 15, [(0, 10), (5, 10)])],
186
 
                        [(0, 10), (5, 10)])
187
 
 
188
 
    def test_coalesce_limit(self):
189
 
        self.check([(10, 50, [(0, 10), (10, 10), (20, 10),
190
 
                              (30, 10), (40, 10)]),
191
 
                    (60, 50, [(0, 10), (10, 10), (20, 10),
192
 
                              (30, 10), (40, 10)]),
193
 
                   ], [(10, 10), (20, 10), (30, 10), (40, 10),
194
 
                       (50, 10), (60, 10), (70, 10), (80, 10),
195
 
                       (90, 10), (100, 10)],
196
 
                    limit=5)
197
 
 
198
 
    def test_coalesce_no_limit(self):
199
 
        self.check([(10, 100, [(0, 10), (10, 10), (20, 10),
200
 
                               (30, 10), (40, 10), (50, 10),
201
 
                               (60, 10), (70, 10), (80, 10),
202
 
                               (90, 10)]),
203
 
                   ], [(10, 10), (20, 10), (30, 10), (40, 10),
204
 
                       (50, 10), (60, 10), (70, 10), (80, 10),
205
 
                       (90, 10), (100, 10)])
206
 
 
207
 
    def test_coalesce_fudge(self):
208
 
        self.check([(10, 30, [(0, 10), (20, 10)]),
209
 
                    (100, 10, [(0, 10),]),
210
 
                   ], [(10, 10), (30, 10), (100, 10)],
211
 
                   fudge=10
212
 
                  )
213
 
    def test_coalesce_max_size(self):
214
 
        self.check([(10, 20, [(0, 10), (10, 10)]),
215
 
                    (30, 50, [(0, 50)]),
216
 
                    # If one range is above max_size, it gets its own coalesced
217
 
                    # offset
218
 
                    (100, 80, [(0, 80),]),],
219
 
                   [(10, 10), (20, 10), (30, 50), (100, 80)],
220
 
                   max_size=50
221
 
                  )
222
 
 
223
 
    def test_coalesce_no_max_size(self):
224
 
        self.check([(10, 170, [(0, 10), (10, 10), (20, 50), (70, 100)]),],
225
 
                   [(10, 10), (20, 10), (30, 50), (80, 100)],
226
 
                  )
227
 
 
228
 
 
229
 
class TestMemoryTransport(TestCase):
230
 
 
231
 
    def test_get_transport(self):
232
 
        MemoryTransport()
233
 
 
234
 
    def test_clone(self):
235
 
        transport = MemoryTransport()
236
 
        self.assertTrue(isinstance(transport, MemoryTransport))
237
 
        self.assertEqual("memory:///", transport.clone("/").base)
238
 
 
239
 
    def test_abspath(self):
240
 
        transport = MemoryTransport()
241
 
        self.assertEqual("memory:///relpath", transport.abspath('relpath'))
242
 
 
243
 
    def test_abspath_of_root(self):
244
 
        transport = MemoryTransport()
245
 
        self.assertEqual("memory:///", transport.base)
246
 
        self.assertEqual("memory:///", transport.abspath('/'))
247
 
 
248
 
    def test_abspath_of_relpath_starting_at_root(self):
249
 
        transport = MemoryTransport()
250
 
        self.assertEqual("memory:///foo", transport.abspath('/foo'))
251
 
 
252
 
    def test_append_and_get(self):
253
 
        transport = MemoryTransport()
254
 
        transport.append_bytes('path', 'content')
255
 
        self.assertEqual(transport.get('path').read(), 'content')
256
 
        transport.append_file('path', StringIO('content'))
257
 
        self.assertEqual(transport.get('path').read(), 'contentcontent')
258
 
 
259
 
    def test_put_and_get(self):
260
 
        transport = MemoryTransport()
261
 
        transport.put_file('path', StringIO('content'))
262
 
        self.assertEqual(transport.get('path').read(), 'content')
263
 
        transport.put_bytes('path', 'content')
264
 
        self.assertEqual(transport.get('path').read(), 'content')
265
 
 
266
 
    def test_append_without_dir_fails(self):
267
 
        transport = MemoryTransport()
268
 
        self.assertRaises(NoSuchFile,
269
 
                          transport.append_bytes, 'dir/path', 'content')
270
 
 
271
 
    def test_put_without_dir_fails(self):
272
 
        transport = MemoryTransport()
273
 
        self.assertRaises(NoSuchFile,
274
 
                          transport.put_file, 'dir/path', StringIO('content'))
275
 
 
276
 
    def test_get_missing(self):
277
 
        transport = MemoryTransport()
278
 
        self.assertRaises(NoSuchFile, transport.get, 'foo')
279
 
 
280
 
    def test_has_missing(self):
281
 
        transport = MemoryTransport()
282
 
        self.assertEquals(False, transport.has('foo'))
283
 
 
284
 
    def test_has_present(self):
285
 
        transport = MemoryTransport()
286
 
        transport.append_bytes('foo', 'content')
287
 
        self.assertEquals(True, transport.has('foo'))
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
 
 
299
 
    def test_mkdir(self):
300
 
        transport = MemoryTransport()
301
 
        transport.mkdir('dir')
302
 
        transport.append_bytes('dir/path', 'content')
303
 
        self.assertEqual(transport.get('dir/path').read(), 'content')
304
 
 
305
 
    def test_mkdir_missing_parent(self):
306
 
        transport = MemoryTransport()
307
 
        self.assertRaises(NoSuchFile,
308
 
                          transport.mkdir, 'dir/dir')
309
 
 
310
 
    def test_mkdir_twice(self):
311
 
        transport = MemoryTransport()
312
 
        transport.mkdir('dir')
313
 
        self.assertRaises(FileExists, transport.mkdir, 'dir')
314
 
 
315
 
    def test_parameters(self):
316
 
        transport = MemoryTransport()
317
 
        self.assertEqual(True, transport.listable())
318
 
        self.assertEqual(False, transport.is_readonly())
319
 
 
320
 
    def test_iter_files_recursive(self):
321
 
        transport = MemoryTransport()
322
 
        transport.mkdir('dir')
323
 
        transport.put_bytes('dir/foo', 'content')
324
 
        transport.put_bytes('dir/bar', 'content')
325
 
        transport.put_bytes('bar', 'content')
326
 
        paths = set(transport.iter_files_recursive())
327
 
        self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
328
 
 
329
 
    def test_stat(self):
330
 
        transport = MemoryTransport()
331
 
        transport.put_bytes('foo', 'content')
332
 
        transport.put_bytes('bar', 'phowar')
333
 
        self.assertEqual(7, transport.stat('foo').st_size)
334
 
        self.assertEqual(6, transport.stat('bar').st_size)
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()
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
 
 
425
 
class ReadonlyDecoratorTransportTest(TestCase):
426
 
    """Readonly decoration specific tests."""
427
 
 
428
 
    def test_local_parameters(self):
429
 
        import bzrlib.transport.readonly as readonly
430
 
        # connect to . in readonly mode
431
 
        transport = readonly.ReadonlyTransportDecorator('readonly+.')
432
 
        self.assertEqual(True, transport.listable())
433
 
        self.assertEqual(True, transport.is_readonly())
434
 
 
435
 
    def test_http_parameters(self):
436
 
        from bzrlib.tests.http_server import HttpServer
437
 
        import bzrlib.transport.readonly as readonly
438
 
        # connect to '.' via http which is not listable
439
 
        server = HttpServer()
440
 
        server.setUp()
441
 
        try:
442
 
            transport = get_transport('readonly+' + server.get_url())
443
 
            self.failUnless(isinstance(transport,
444
 
                                       readonly.ReadonlyTransportDecorator))
445
 
            self.assertEqual(False, transport.listable())
446
 
            self.assertEqual(True, transport.is_readonly())
447
 
        finally:
448
 
            server.tearDown()
449
 
 
450
 
 
451
 
class FakeNFSDecoratorTests(TestCaseInTempDir):
452
 
    """NFS decorator specific tests."""
453
 
 
454
 
    def get_nfs_transport(self, url):
455
 
        import bzrlib.transport.fakenfs as fakenfs
456
 
        # connect to url with nfs decoration
457
 
        return fakenfs.FakeNFSTransportDecorator('fakenfs+' + url)
458
 
 
459
 
    def test_local_parameters(self):
460
 
        # the listable and is_readonly parameters
461
 
        # are not changed by the fakenfs decorator
462
 
        transport = self.get_nfs_transport('.')
463
 
        self.assertEqual(True, transport.listable())
464
 
        self.assertEqual(False, transport.is_readonly())
465
 
 
466
 
    def test_http_parameters(self):
467
 
        # the listable and is_readonly parameters
468
 
        # are not changed by the fakenfs decorator
469
 
        from bzrlib.tests.http_server import HttpServer
470
 
        # connect to '.' via http which is not listable
471
 
        server = HttpServer()
472
 
        server.setUp()
473
 
        try:
474
 
            transport = self.get_nfs_transport(server.get_url())
475
 
            self.assertIsInstance(
476
 
                transport, bzrlib.transport.fakenfs.FakeNFSTransportDecorator)
477
 
            self.assertEqual(False, transport.listable())
478
 
            self.assertEqual(True, transport.is_readonly())
479
 
        finally:
480
 
            server.tearDown()
481
 
 
482
 
    def test_fakenfs_server_default(self):
483
 
        # a FakeNFSServer() should bring up a local relpath server for itself
484
 
        import bzrlib.transport.fakenfs as fakenfs
485
 
        server = fakenfs.FakeNFSServer()
486
 
        server.setUp()
487
 
        try:
488
 
            # the url should be decorated appropriately
489
 
            self.assertStartsWith(server.get_url(), 'fakenfs+')
490
 
            # and we should be able to get a transport for it
491
 
            transport = get_transport(server.get_url())
492
 
            # which must be a FakeNFSTransportDecorator instance.
493
 
            self.assertIsInstance(
494
 
                transport, fakenfs.FakeNFSTransportDecorator)
495
 
        finally:
496
 
            server.tearDown()
497
 
 
498
 
    def test_fakenfs_rename_semantics(self):
499
 
        # a FakeNFS transport must mangle the way rename errors occur to
500
 
        # look like NFS problems.
501
 
        transport = self.get_nfs_transport('.')
502
 
        self.build_tree(['from/', 'from/foo', 'to/', 'to/bar'],
503
 
                        transport=transport)
504
 
        self.assertRaises(errors.ResourceBusy,
505
 
                          transport.rename, 'from', 'to')
506
 
 
507
 
 
508
 
class FakeVFATDecoratorTests(TestCaseInTempDir):
509
 
    """Tests for simulation of VFAT restrictions"""
510
 
 
511
 
    def get_vfat_transport(self, url):
512
 
        """Return vfat-backed transport for test directory"""
513
 
        from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
514
 
        return FakeVFATTransportDecorator('vfat+' + url)
515
 
 
516
 
    def test_transport_creation(self):
517
 
        from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
518
 
        transport = self.get_vfat_transport('.')
519
 
        self.assertIsInstance(transport, FakeVFATTransportDecorator)
520
 
 
521
 
    def test_transport_mkdir(self):
522
 
        transport = self.get_vfat_transport('.')
523
 
        transport.mkdir('HELLO')
524
 
        self.assertTrue(transport.has('hello'))
525
 
        self.assertTrue(transport.has('Hello'))
526
 
 
527
 
    def test_forbidden_chars(self):
528
 
        transport = self.get_vfat_transport('.')
529
 
        self.assertRaises(ValueError, transport.has, "<NU>")
530
 
 
531
 
 
532
 
class BadTransportHandler(Transport):
533
 
    def __init__(self, base_url):
534
 
        raise DependencyNotPresent('some_lib', 'testing missing dependency')
535
 
 
536
 
 
537
 
class BackupTransportHandler(Transport):
538
 
    """Test transport that works as a backup for the BadTransportHandler"""
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)