/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

First attempt to merge .dev and resolve the conflicts (but tests are 
failing)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
1
 
# Copyright (C) 2004, 2005, 2006 by Canonical Ltd
 
1
# Copyright (C) 2004, 2005, 2006, 2007 Canonical Ltd
2
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
15
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
 
18
 
import os
19
 
import sys
20
 
import stat
21
18
from cStringIO import StringIO
22
19
 
23
20
import bzrlib
24
 
from bzrlib.errors import (NoSuchFile, FileExists,
25
 
                           TransportNotPossible,
26
 
                           ConnectionError,
27
 
                           DependencyNotPresent,
 
21
from bzrlib import (
 
22
    errors,
 
23
    osutils,
 
24
    urlutils,
 
25
    )
 
26
from bzrlib.errors import (DependencyNotPresent,
 
27
                           FileExists,
 
28
                           InvalidURLJoin,
 
29
                           NoSuchFile,
 
30
                           PathNotChild,
 
31
                           ReadError,
28
32
                           UnsupportedProtocol,
29
33
                           )
30
34
from bzrlib.tests import TestCase, TestCaseInTempDir
31
 
from bzrlib.transport import (_CoalescedOffset,
 
35
from bzrlib.transport import (_clear_protocol_handlers,
 
36
                              _CoalescedOffset,
 
37
                              ConnectedTransport,
32
38
                              _get_protocol_handlers,
 
39
                              _set_protocol_handlers,
33
40
                              _get_transport_modules,
34
41
                              get_transport,
 
42
                              LateReadError,
35
43
                              register_lazy_transport,
36
 
                              _set_protocol_handlers,
 
44
                              register_transport_proto,
37
45
                              Transport,
38
46
                              )
 
47
from bzrlib.transport.chroot import ChrootServer
39
48
from bzrlib.transport.memory import MemoryTransport
40
 
from bzrlib.transport.local import LocalTransport
 
49
from bzrlib.transport.local import (LocalTransport,
 
50
                                    EmulatedWin32LocalTransport)
 
51
 
 
52
 
 
53
# TODO: Should possibly split transport-specific tests into their own files.
41
54
 
42
55
 
43
56
class TestTransport(TestCase):
45
58
 
46
59
    def test__get_set_protocol_handlers(self):
47
60
        handlers = _get_protocol_handlers()
48
 
        self.assertNotEqual({}, handlers)
 
61
        self.assertNotEqual([], handlers.keys( ))
49
62
        try:
50
 
            _set_protocol_handlers({})
51
 
            self.assertEqual({}, _get_protocol_handlers())
 
63
            _clear_protocol_handlers()
 
64
            self.assertEqual([], _get_protocol_handlers().keys())
52
65
        finally:
53
66
            _set_protocol_handlers(handlers)
54
67
 
55
68
    def test_get_transport_modules(self):
56
69
        handlers = _get_protocol_handlers()
 
70
        # don't pollute the current handlers
 
71
        _clear_protocol_handlers()
57
72
        class SampleHandler(object):
58
73
            """I exist, isnt that enough?"""
59
74
        try:
60
 
            my_handlers = {}
61
 
            _set_protocol_handlers(my_handlers)
62
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
63
 
            register_lazy_transport('bar', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
64
 
            self.assertEqual([SampleHandler.__module__],
 
75
            _clear_protocol_handlers()
 
76
            register_transport_proto('foo')
 
77
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
 
78
                                    'TestTransport.SampleHandler')
 
79
            register_transport_proto('bar')
 
80
            register_lazy_transport('bar', 'bzrlib.tests.test_transport',
 
81
                                    'TestTransport.SampleHandler')
 
82
            self.assertEqual([SampleHandler.__module__,
 
83
                              'bzrlib.transport.chroot'],
65
84
                             _get_transport_modules())
66
85
        finally:
67
86
            _set_protocol_handlers(handlers)
69
88
    def test_transport_dependency(self):
70
89
        """Transport with missing dependency causes no error"""
71
90
        saved_handlers = _get_protocol_handlers()
 
91
        # don't pollute the current handlers
 
92
        _clear_protocol_handlers()
72
93
        try:
 
94
            register_transport_proto('foo')
73
95
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
74
96
                    'BadTransportHandler')
75
97
            try:
90
112
        """Transport with missing dependency causes no error"""
91
113
        saved_handlers = _get_protocol_handlers()
92
114
        try:
93
 
            _set_protocol_handlers({})
 
115
            _clear_protocol_handlers()
 
116
            register_transport_proto('foo')
94
117
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
95
118
                    'BackupTransportHandler')
96
119
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
100
123
        finally:
101
124
            _set_protocol_handlers(saved_handlers)
102
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
 
103
153
 
104
154
class TestCoalesceOffsets(TestCase):
105
 
    
106
 
    def check(self, expected, offsets, limit=0, fudge=0):
 
155
 
 
156
    def check(self, expected, offsets, limit=0, max_size=0, fudge=0):
107
157
        coalesce = Transport._coalesce_offsets
108
158
        exp = [_CoalescedOffset(*x) for x in expected]
109
 
        out = list(coalesce(offsets, limit=limit, fudge_factor=fudge))
 
159
        out = list(coalesce(offsets, limit=limit, fudge_factor=fudge,
 
160
                            max_size=max_size))
110
161
        self.assertEqual(exp, out)
111
162
 
112
163
    def test_coalesce_empty(self):
119
170
        self.check([(0, 10, [(0, 10)]),
120
171
                    (20, 10, [(0, 10)]),
121
172
                   ], [(0, 10), (20, 10)])
122
 
            
 
173
 
123
174
    def test_coalesce_unsorted(self):
124
175
        self.check([(20, 10, [(0, 10)]),
125
176
                    (0, 10, [(0, 10)]),
129
180
        self.check([(0, 20, [(0, 10), (10, 10)])],
130
181
                   [(0, 10), (10, 10)])
131
182
 
 
183
    # XXX: scary, http.readv() can't handle that --vila20071209
132
184
    def test_coalesce_overlapped(self):
133
185
        self.check([(0, 15, [(0, 10), (5, 10)])],
134
186
                   [(0, 10), (5, 10)])
158
210
                   ], [(10, 10), (30, 10), (100, 10)],
159
211
                   fudge=10
160
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
                  )
161
227
 
162
228
 
163
229
class TestMemoryTransport(TestCase):
168
234
    def test_clone(self):
169
235
        transport = MemoryTransport()
170
236
        self.assertTrue(isinstance(transport, MemoryTransport))
 
237
        self.assertEqual("memory:///", transport.clone("/").base)
171
238
 
172
239
    def test_abspath(self):
173
240
        transport = MemoryTransport()
174
241
        self.assertEqual("memory:///relpath", transport.abspath('relpath'))
175
242
 
176
 
    def test_relpath(self):
177
 
        transport = MemoryTransport()
 
243
    def test_abspath_of_root(self):
 
244
        transport = MemoryTransport()
 
245
        self.assertEqual("memory:///", transport.base)
 
246
        self.assertEqual("memory:///", transport.abspath('/'))
 
247
 
 
248
    def test_abspath_of_relpath_starting_at_root(self):
 
249
        transport = MemoryTransport()
 
250
        self.assertEqual("memory:///foo", transport.abspath('/foo'))
178
251
 
179
252
    def test_append_and_get(self):
180
253
        transport = MemoryTransport()
213
286
        transport.append_bytes('foo', 'content')
214
287
        self.assertEquals(True, transport.has('foo'))
215
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
 
216
299
    def test_mkdir(self):
217
300
        transport = MemoryTransport()
218
301
        transport.mkdir('dir')
232
315
    def test_parameters(self):
233
316
        transport = MemoryTransport()
234
317
        self.assertEqual(True, transport.listable())
235
 
        self.assertEqual(False, transport.should_cache())
236
318
        self.assertEqual(False, transport.is_readonly())
237
319
 
238
320
    def test_iter_files_recursive(self):
251
333
        self.assertEqual(7, transport.stat('foo').st_size)
252
334
        self.assertEqual(6, transport.stat('bar').st_size)
253
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()
254
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
 
255
425
class ReadonlyDecoratorTransportTest(TestCase):
256
426
    """Readonly decoration specific tests."""
257
427
 
260
430
        # connect to . in readonly mode
261
431
        transport = readonly.ReadonlyTransportDecorator('readonly+.')
262
432
        self.assertEqual(True, transport.listable())
263
 
        self.assertEqual(False, transport.should_cache())
264
433
        self.assertEqual(True, transport.is_readonly())
265
434
 
266
435
    def test_http_parameters(self):
 
436
        from bzrlib.tests.http_server import HttpServer
267
437
        import bzrlib.transport.readonly as readonly
268
 
        from bzrlib.transport.http import HttpServer
269
 
        # connect to . via http which is not listable
 
438
        # connect to '.' via http which is not listable
270
439
        server = HttpServer()
271
440
        server.setUp()
272
441
        try:
274
443
            self.failUnless(isinstance(transport,
275
444
                                       readonly.ReadonlyTransportDecorator))
276
445
            self.assertEqual(False, transport.listable())
277
 
            self.assertEqual(True, transport.should_cache())
278
446
            self.assertEqual(True, transport.is_readonly())
279
447
        finally:
280
448
            server.tearDown()
289
457
        return fakenfs.FakeNFSTransportDecorator('fakenfs+' + url)
290
458
 
291
459
    def test_local_parameters(self):
292
 
        # the listable, should_cache and is_readonly parameters
 
460
        # the listable and is_readonly parameters
293
461
        # are not changed by the fakenfs decorator
294
462
        transport = self.get_nfs_transport('.')
295
463
        self.assertEqual(True, transport.listable())
296
 
        self.assertEqual(False, transport.should_cache())
297
464
        self.assertEqual(False, transport.is_readonly())
298
465
 
299
466
    def test_http_parameters(self):
300
 
        # the listable, should_cache and is_readonly parameters
 
467
        # the listable and is_readonly parameters
301
468
        # are not changed by the fakenfs decorator
302
 
        from bzrlib.transport.http import HttpServer
303
 
        # connect to . via http which is not listable
 
469
        from bzrlib.tests.http_server import HttpServer
 
470
        # connect to '.' via http which is not listable
304
471
        server = HttpServer()
305
472
        server.setUp()
306
473
        try:
308
475
            self.assertIsInstance(
309
476
                transport, bzrlib.transport.fakenfs.FakeNFSTransportDecorator)
310
477
            self.assertEqual(False, transport.listable())
311
 
            self.assertEqual(True, transport.should_cache())
312
478
            self.assertEqual(True, transport.is_readonly())
313
479
        finally:
314
480
            server.tearDown()
319
485
        server = fakenfs.FakeNFSServer()
320
486
        server.setUp()
321
487
        try:
322
 
            # the server should be a relpath localhost server
323
 
            self.assertEqual(server.get_url(), 'fakenfs+.')
 
488
            # the url should be decorated appropriately
 
489
            self.assertStartsWith(server.get_url(), 'fakenfs+')
324
490
            # and we should be able to get a transport for it
325
491
            transport = get_transport(server.get_url())
326
492
            # which must be a FakeNFSTransportDecorator instance.
335
501
        transport = self.get_nfs_transport('.')
336
502
        self.build_tree(['from/', 'from/foo', 'to/', 'to/bar'],
337
503
                        transport=transport)
338
 
        self.assertRaises(bzrlib.errors.ResourceBusy,
 
504
        self.assertRaises(errors.ResourceBusy,
339
505
                          transport.rename, 'from', 'to')
340
506
 
341
507
 
399
565
        super(TestTransportImplementation, self).setUp()
400
566
        self._server = self.transport_server()
401
567
        self._server.setUp()
402
 
 
403
 
    def tearDown(self):
404
 
        super(TestTransportImplementation, self).tearDown()
405
 
        self._server.tearDown()
406
 
        
407
 
    def get_transport(self):
408
 
        """Return a connected transport to the local directory."""
 
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
        """
409
575
        base_url = self._server.get_url()
 
576
        url = self._adjust_url(base_url, relpath)
410
577
        # try getting the transport via the regular interface:
411
 
        t = get_transport(base_url)
412
 
        if not isinstance(t, self.transport_class): 
 
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):
413
583
            # we did not get the correct transport class type. Override the
414
584
            # regular connection behaviour by direct construction.
415
 
            t = self.transport_class(base_url)
 
585
            t = self.transport_class(url)
416
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)