← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2007-10-24 12:49:17 UTC
  • mfrom: (2935.1.1 ianc-integration)
  • Revision ID: pqm@pqm.ubuntu.com-20071024124917-xb75eckyxx6vkrlg
Makefile fixes - hooks.html generation & allow python to be overridden (Ian Clatworthy)

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
from bzrlib.transport.remote import (
 
52
    BZR_DEFAULT_PORT,
 
53
    RemoteTCPTransport
 
54
    )
 
55
 
 
56
 
 
57
# TODO: Should possibly split transport-specific tests into their own files.
41
58
 
42
59
 
43
60
class TestTransport(TestCase):
45
62
 
46
63
    def test__get_set_protocol_handlers(self):
47
64
        handlers = _get_protocol_handlers()
48
 
        self.assertNotEqual({}, handlers)
 
65
        self.assertNotEqual([], handlers.keys( ))
49
66
        try:
50
 
            _set_protocol_handlers({})
51
 
            self.assertEqual({}, _get_protocol_handlers())
 
67
            _clear_protocol_handlers()
 
68
            self.assertEqual([], _get_protocol_handlers().keys())
52
69
        finally:
53
70
            _set_protocol_handlers(handlers)
54
71
 
55
72
    def test_get_transport_modules(self):
56
73
        handlers = _get_protocol_handlers()
 
74
        # don't pollute the current handlers
 
75
        _clear_protocol_handlers()
57
76
        class SampleHandler(object):
58
77
            """I exist, isnt that enough?"""
59
78
        try:
60
 
            my_handlers = {}
61
 
            _set_protocol_handlers(my_handlers)
 
79
            _clear_protocol_handlers()
 
80
            register_transport_proto('foo')
62
81
            register_lazy_transport('foo', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
 
82
            register_transport_proto('bar')
63
83
            register_lazy_transport('bar', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
64
 
            self.assertEqual([SampleHandler.__module__],
 
84
            self.assertEqual([SampleHandler.__module__, 'bzrlib.transport.chroot'],
65
85
                             _get_transport_modules())
66
86
        finally:
67
87
            _set_protocol_handlers(handlers)
69
89
    def test_transport_dependency(self):
70
90
        """Transport with missing dependency causes no error"""
71
91
        saved_handlers = _get_protocol_handlers()
 
92
        # don't pollute the current handlers
 
93
        _clear_protocol_handlers()
72
94
        try:
 
95
            register_transport_proto('foo')
73
96
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
74
97
                    'BadTransportHandler')
75
98
            try:
90
113
        """Transport with missing dependency causes no error"""
91
114
        saved_handlers = _get_protocol_handlers()
92
115
        try:
93
 
            _set_protocol_handlers({})
 
116
            _clear_protocol_handlers()
 
117
            register_transport_proto('foo')
94
118
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
95
119
                    'BackupTransportHandler')
96
120
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
100
124
        finally:
101
125
            _set_protocol_handlers(saved_handlers)
102
126
 
 
127
    def test_LateReadError(self):
 
128
        """The LateReadError helper should raise on read()."""
 
129
        a_file = LateReadError('a path')
 
130
        try:
 
131
            a_file.read()
 
132
        except ReadError, error:
 
133
            self.assertEqual('a path', error.path)
 
134
        self.assertRaises(ReadError, a_file.read, 40)
 
135
        a_file.close()
 
136
 
 
137
    def test__combine_paths(self):
 
138
        t = Transport('/')
 
139
        self.assertEqual('/home/sarah/project/foo',
 
140
                         t._combine_paths('/home/sarah', 'project/foo'))
 
141
        self.assertEqual('/etc',
 
142
                         t._combine_paths('/home/sarah', '../../etc'))
 
143
        self.assertEqual('/etc',
 
144
                         t._combine_paths('/home/sarah', '../../../etc'))
 
145
        self.assertEqual('/etc',
 
146
                         t._combine_paths('/home/sarah', '/etc'))
 
147
 
 
148
    def test_local_abspath_non_local_transport(self):
 
149
        # the base implementation should throw
 
150
        t = MemoryTransport()
 
151
        e = self.assertRaises(errors.NotLocalUrl, t.local_abspath, 't')
 
152
        self.assertEqual('memory:///t is not a local path.', str(e))
 
153
 
103
154
 
104
155
class TestCoalesceOffsets(TestCase):
105
156
    
168
219
    def test_clone(self):
169
220
        transport = MemoryTransport()
170
221
        self.assertTrue(isinstance(transport, MemoryTransport))
 
222
        self.assertEqual("memory:///", transport.clone("/").base)
171
223
 
172
224
    def test_abspath(self):
173
225
        transport = MemoryTransport()
174
226
        self.assertEqual("memory:///relpath", transport.abspath('relpath'))
175
227
 
176
 
    def test_relpath(self):
177
 
        transport = MemoryTransport()
 
228
    def test_abspath_of_root(self):
 
229
        transport = MemoryTransport()
 
230
        self.assertEqual("memory:///", transport.base)
 
231
        self.assertEqual("memory:///", transport.abspath('/'))
 
232
 
 
233
    def test_abspath_of_relpath_starting_at_root(self):
 
234
        transport = MemoryTransport()
 
235
        self.assertEqual("memory:///foo", transport.abspath('/foo'))
178
236
 
179
237
    def test_append_and_get(self):
180
238
        transport = MemoryTransport()
181
 
        transport.append('path', StringIO('content'))
 
239
        transport.append_bytes('path', 'content')
182
240
        self.assertEqual(transport.get('path').read(), 'content')
183
 
        transport.append('path', StringIO('content'))
 
241
        transport.append_file('path', StringIO('content'))
184
242
        self.assertEqual(transport.get('path').read(), 'contentcontent')
185
243
 
186
244
    def test_put_and_get(self):
187
245
        transport = MemoryTransport()
188
 
        transport.put('path', StringIO('content'))
 
246
        transport.put_file('path', StringIO('content'))
189
247
        self.assertEqual(transport.get('path').read(), 'content')
190
 
        transport.put('path', StringIO('content'))
 
248
        transport.put_bytes('path', 'content')
191
249
        self.assertEqual(transport.get('path').read(), 'content')
192
250
 
193
251
    def test_append_without_dir_fails(self):
194
252
        transport = MemoryTransport()
195
253
        self.assertRaises(NoSuchFile,
196
 
                          transport.append, 'dir/path', StringIO('content'))
 
254
                          transport.append_bytes, 'dir/path', 'content')
197
255
 
198
256
    def test_put_without_dir_fails(self):
199
257
        transport = MemoryTransport()
200
258
        self.assertRaises(NoSuchFile,
201
 
                          transport.put, 'dir/path', StringIO('content'))
 
259
                          transport.put_file, 'dir/path', StringIO('content'))
202
260
 
203
261
    def test_get_missing(self):
204
262
        transport = MemoryTransport()
210
268
 
211
269
    def test_has_present(self):
212
270
        transport = MemoryTransport()
213
 
        transport.append('foo', StringIO('content'))
 
271
        transport.append_bytes('foo', 'content')
214
272
        self.assertEquals(True, transport.has('foo'))
215
273
 
 
274
    def test_list_dir(self):
 
275
        transport = MemoryTransport()
 
276
        transport.put_bytes('foo', 'content')
 
277
        transport.mkdir('dir')
 
278
        transport.put_bytes('dir/subfoo', 'content')
 
279
        transport.put_bytes('dirlike', 'content')
 
280
 
 
281
        self.assertEquals(['dir', 'dirlike', 'foo'], sorted(transport.list_dir('.')))
 
282
        self.assertEquals(['subfoo'], sorted(transport.list_dir('dir')))
 
283
 
216
284
    def test_mkdir(self):
217
285
        transport = MemoryTransport()
218
286
        transport.mkdir('dir')
219
 
        transport.append('dir/path', StringIO('content'))
 
287
        transport.append_bytes('dir/path', 'content')
220
288
        self.assertEqual(transport.get('dir/path').read(), 'content')
221
289
 
222
290
    def test_mkdir_missing_parent(self):
232
300
    def test_parameters(self):
233
301
        transport = MemoryTransport()
234
302
        self.assertEqual(True, transport.listable())
235
 
        self.assertEqual(False, transport.should_cache())
236
303
        self.assertEqual(False, transport.is_readonly())
237
304
 
238
305
    def test_iter_files_recursive(self):
239
306
        transport = MemoryTransport()
240
307
        transport.mkdir('dir')
241
 
        transport.put('dir/foo', StringIO('content'))
242
 
        transport.put('dir/bar', StringIO('content'))
243
 
        transport.put('bar', StringIO('content'))
 
308
        transport.put_bytes('dir/foo', 'content')
 
309
        transport.put_bytes('dir/bar', 'content')
 
310
        transport.put_bytes('bar', 'content')
244
311
        paths = set(transport.iter_files_recursive())
245
312
        self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
246
313
 
247
314
    def test_stat(self):
248
315
        transport = MemoryTransport()
249
 
        transport.put('foo', StringIO('content'))
250
 
        transport.put('bar', StringIO('phowar'))
 
316
        transport.put_bytes('foo', 'content')
 
317
        transport.put_bytes('bar', 'phowar')
251
318
        self.assertEqual(7, transport.stat('foo').st_size)
252
319
        self.assertEqual(6, transport.stat('bar').st_size)
253
320
 
 
321
 
 
322
class ChrootDecoratorTransportTest(TestCase):
 
323
    """Chroot decoration specific tests."""
 
324
 
 
325
    def test_abspath(self):
 
326
        # The abspath is always relative to the chroot_url.
 
327
        server = ChrootServer(get_transport('memory:///foo/bar/'))
 
328
        server.setUp()
 
329
        transport = get_transport(server.get_url())
 
330
        self.assertEqual(server.get_url(), transport.abspath('/'))
 
331
 
 
332
        subdir_transport = transport.clone('subdir')
 
333
        self.assertEqual(server.get_url(), subdir_transport.abspath('/'))
 
334
        server.tearDown()
 
335
 
 
336
    def test_clone(self):
 
337
        server = ChrootServer(get_transport('memory:///foo/bar/'))
 
338
        server.setUp()
 
339
        transport = get_transport(server.get_url())
 
340
        # relpath from root and root path are the same
 
341
        relpath_cloned = transport.clone('foo')
 
342
        abspath_cloned = transport.clone('/foo')
 
343
        self.assertEqual(server, relpath_cloned.server)
 
344
        self.assertEqual(server, abspath_cloned.server)
 
345
        server.tearDown()
 
346
    
 
347
    def test_chroot_url_preserves_chroot(self):
 
348
        """Calling get_transport on a chroot transport's base should produce a
 
349
        transport with exactly the same behaviour as the original chroot
 
350
        transport.
 
351
 
 
352
        This is so that it is not possible to escape a chroot by doing::
 
353
            url = chroot_transport.base
 
354
            parent_url = urlutils.join(url, '..')
 
355
            new_transport = get_transport(parent_url)
 
356
        """
 
357
        server = ChrootServer(get_transport('memory:///path/subpath'))
 
358
        server.setUp()
 
359
        transport = get_transport(server.get_url())
 
360
        new_transport = get_transport(transport.base)
 
361
        self.assertEqual(transport.server, new_transport.server)
 
362
        self.assertEqual(transport.base, new_transport.base)
 
363
        server.tearDown()
254
364
        
 
365
    def test_urljoin_preserves_chroot(self):
 
366
        """Using urlutils.join(url, '..') on a chroot URL should not produce a
 
367
        URL that escapes the intended chroot.
 
368
 
 
369
        This is so that it is not possible to escape a chroot by doing::
 
370
            url = chroot_transport.base
 
371
            parent_url = urlutils.join(url, '..')
 
372
            new_transport = get_transport(parent_url)
 
373
        """
 
374
        server = ChrootServer(get_transport('memory:///path/'))
 
375
        server.setUp()
 
376
        transport = get_transport(server.get_url())
 
377
        self.assertRaises(
 
378
            InvalidURLJoin, urlutils.join, transport.base, '..')
 
379
        server.tearDown()
 
380
 
 
381
 
 
382
class ChrootServerTest(TestCase):
 
383
 
 
384
    def test_construct(self):
 
385
        backing_transport = MemoryTransport()
 
386
        server = ChrootServer(backing_transport)
 
387
        self.assertEqual(backing_transport, server.backing_transport)
 
388
 
 
389
    def test_setUp(self):
 
390
        backing_transport = MemoryTransport()
 
391
        server = ChrootServer(backing_transport)
 
392
        server.setUp()
 
393
        self.assertTrue(server.scheme in _get_protocol_handlers().keys())
 
394
 
 
395
    def test_tearDown(self):
 
396
        backing_transport = MemoryTransport()
 
397
        server = ChrootServer(backing_transport)
 
398
        server.setUp()
 
399
        server.tearDown()
 
400
        self.assertFalse(server.scheme in _get_protocol_handlers().keys())
 
401
 
 
402
    def test_get_url(self):
 
403
        backing_transport = MemoryTransport()
 
404
        server = ChrootServer(backing_transport)
 
405
        server.setUp()
 
406
        self.assertEqual('chroot-%d:///' % id(server), server.get_url())
 
407
        server.tearDown()
 
408
 
 
409
 
255
410
class ReadonlyDecoratorTransportTest(TestCase):
256
411
    """Readonly decoration specific tests."""
257
412
 
260
415
        # connect to . in readonly mode
261
416
        transport = readonly.ReadonlyTransportDecorator('readonly+.')
262
417
        self.assertEqual(True, transport.listable())
263
 
        self.assertEqual(False, transport.should_cache())
264
418
        self.assertEqual(True, transport.is_readonly())
265
419
 
266
420
    def test_http_parameters(self):
 
421
        from bzrlib.tests.HttpServer import HttpServer
267
422
        import bzrlib.transport.readonly as readonly
268
 
        from bzrlib.transport.http import HttpServer
269
423
        # connect to . via http which is not listable
270
424
        server = HttpServer()
271
425
        server.setUp()
274
428
            self.failUnless(isinstance(transport,
275
429
                                       readonly.ReadonlyTransportDecorator))
276
430
            self.assertEqual(False, transport.listable())
277
 
            self.assertEqual(True, transport.should_cache())
278
431
            self.assertEqual(True, transport.is_readonly())
279
432
        finally:
280
433
            server.tearDown()
289
442
        return fakenfs.FakeNFSTransportDecorator('fakenfs+' + url)
290
443
 
291
444
    def test_local_parameters(self):
292
 
        # the listable, should_cache and is_readonly parameters
 
445
        # the listable and is_readonly parameters
293
446
        # are not changed by the fakenfs decorator
294
447
        transport = self.get_nfs_transport('.')
295
448
        self.assertEqual(True, transport.listable())
296
 
        self.assertEqual(False, transport.should_cache())
297
449
        self.assertEqual(False, transport.is_readonly())
298
450
 
299
451
    def test_http_parameters(self):
300
 
        # the listable, should_cache and is_readonly parameters
 
452
        # the listable and is_readonly parameters
301
453
        # are not changed by the fakenfs decorator
302
 
        from bzrlib.transport.http import HttpServer
 
454
        from bzrlib.tests.HttpServer import HttpServer
303
455
        # connect to . via http which is not listable
304
456
        server = HttpServer()
305
457
        server.setUp()
308
460
            self.assertIsInstance(
309
461
                transport, bzrlib.transport.fakenfs.FakeNFSTransportDecorator)
310
462
            self.assertEqual(False, transport.listable())
311
 
            self.assertEqual(True, transport.should_cache())
312
463
            self.assertEqual(True, transport.is_readonly())
313
464
        finally:
314
465
            server.tearDown()
319
470
        server = fakenfs.FakeNFSServer()
320
471
        server.setUp()
321
472
        try:
322
 
            # the server should be a relpath localhost server
323
 
            self.assertEqual(server.get_url(), 'fakenfs+.')
 
473
            # the url should be decorated appropriately
 
474
            self.assertStartsWith(server.get_url(), 'fakenfs+')
324
475
            # and we should be able to get a transport for it
325
476
            transport = get_transport(server.get_url())
326
477
            # which must be a FakeNFSTransportDecorator instance.
335
486
        transport = self.get_nfs_transport('.')
336
487
        self.build_tree(['from/', 'from/foo', 'to/', 'to/bar'],
337
488
                        transport=transport)
338
 
        self.assertRaises(bzrlib.errors.ResourceBusy,
 
489
        self.assertRaises(errors.ResourceBusy,
339
490
                          transport.rename, 'from', 'to')
340
491
 
341
492
 
399
550
        super(TestTransportImplementation, self).setUp()
400
551
        self._server = self.transport_server()
401
552
        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."""
 
553
        self.addCleanup(self._server.tearDown)
 
554
 
 
555
    def get_transport(self, relpath=None):
 
556
        """Return a connected transport to the local directory.
 
557
 
 
558
        :param relpath: a path relative to the base url.
 
559
        """
409
560
        base_url = self._server.get_url()
 
561
        url = self._adjust_url(base_url, relpath)
410
562
        # try getting the transport via the regular interface:
411
 
        t = get_transport(base_url)
412
 
        if not isinstance(t, self.transport_class): 
 
563
        t = get_transport(url)
 
564
        # vila--20070607 if the following are commented out the test suite
 
565
        # still pass. Is this really still needed or was it a forgotten
 
566
        # temporary fix ?
 
567
        if not isinstance(t, self.transport_class):
413
568
            # we did not get the correct transport class type. Override the
414
569
            # regular connection behaviour by direct construction.
415
 
            t = self.transport_class(base_url)
 
570
            t = self.transport_class(url)
416
571
        return t
 
572
 
 
573
 
 
574
class TestLocalTransports(TestCase):
 
575
 
 
576
    def test_get_transport_from_abspath(self):
 
577
        here = osutils.abspath('.')
 
578
        t = get_transport(here)
 
579
        self.assertIsInstance(t, LocalTransport)
 
580
        self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
 
581
 
 
582
    def test_get_transport_from_relpath(self):
 
583
        here = osutils.abspath('.')
 
584
        t = get_transport('.')
 
585
        self.assertIsInstance(t, LocalTransport)
 
586
        self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
 
587
 
 
588
    def test_get_transport_from_local_url(self):
 
589
        here = osutils.abspath('.')
 
590
        here_url = urlutils.local_path_to_url(here) + '/'
 
591
        t = get_transport(here_url)
 
592
        self.assertIsInstance(t, LocalTransport)
 
593
        self.assertEquals(t.base, here_url)
 
594
 
 
595
    def test_local_abspath(self):
 
596
        here = osutils.abspath('.')
 
597
        t = get_transport(here)
 
598
        self.assertEquals(t.local_abspath(''), here)
 
599
 
 
600
 
 
601
class TestWin32LocalTransport(TestCase):
 
602
 
 
603
    def test_unc_clone_to_root(self):
 
604
        # Win32 UNC path like \\HOST\path
 
605
        # clone to root should stop at least at \\HOST part
 
606
        # not on \\
 
607
        t = EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
 
608
        for i in xrange(4):
 
609
            t = t.clone('..')
 
610
        self.assertEquals(t.base, 'file://HOST/')
 
611
        # make sure we reach the root
 
612
        t = t.clone('..')
 
613
        self.assertEquals(t.base, 'file://HOST/')
 
614
 
 
615
 
 
616
class TestConnectedTransport(TestCase):
 
617
    """Tests for connected to remote server transports"""
 
618
 
 
619
    def test_parse_url(self):
 
620
        t = ConnectedTransport('http://simple.example.com/home/source')
 
621
        self.assertEquals(t._host, 'simple.example.com')
 
622
        self.assertEquals(t._port, 80)
 
623
        self.assertEquals(t._path, '/home/source/')
 
624
        self.failUnless(t._user is None)
 
625
        self.failUnless(t._password is None)
 
626
 
 
627
        self.assertEquals(t.base, 'http://simple.example.com/home/source/')
 
628
 
 
629
    def test_parse_quoted_url(self):
 
630
        t = ConnectedTransport('http://ro%62ey:h%40t@ex%41mple.com:2222/path')
 
631
        self.assertEquals(t._host, 'exAmple.com')
 
632
        self.assertEquals(t._port, 2222)
 
633
        self.assertEquals(t._user, 'robey')
 
634
        self.assertEquals(t._password, 'h@t')
 
635
        self.assertEquals(t._path, '/path/')
 
636
 
 
637
        # Base should not keep track of the password
 
638
        self.assertEquals(t.base, 'http://robey@exAmple.com:2222/path/')
 
639
 
 
640
    def test_parse_invalid_url(self):
 
641
        self.assertRaises(errors.InvalidURL,
 
642
                          ConnectedTransport,
 
643
                          'sftp://lily.org:~janneke/public/bzr/gub')
 
644
 
 
645
    def test_relpath(self):
 
646
        t = ConnectedTransport('sftp://user@host.com/abs/path')
 
647
 
 
648
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
 
649
        self.assertRaises(errors.PathNotChild, t.relpath,
 
650
                          'http://user@host.com/abs/path/sub')
 
651
        self.assertRaises(errors.PathNotChild, t.relpath,
 
652
                          'sftp://user2@host.com/abs/path/sub')
 
653
        self.assertRaises(errors.PathNotChild, t.relpath,
 
654
                          'sftp://user@otherhost.com/abs/path/sub')
 
655
        self.assertRaises(errors.PathNotChild, t.relpath,
 
656
                          'sftp://user@host.com:33/abs/path/sub')
 
657
        # Make sure it works when we don't supply a username
 
658
        t = ConnectedTransport('sftp://host.com/abs/path')
 
659
        self.assertEquals(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
 
660
 
 
661
        # Make sure it works when parts of the path will be url encoded
 
662
        t = ConnectedTransport('sftp://host.com/dev/%path')
 
663
        self.assertEquals(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
 
664
 
 
665
    def test_connection_sharing_propagate_credentials(self):
 
666
        t = ConnectedTransport('foo://user@host.com/abs/path')
 
667
        self.assertIs(None, t._get_connection())
 
668
        self.assertIs(None, t._password)
 
669
        c = t.clone('subdir')
 
670
        self.assertEquals(None, c._get_connection())
 
671
        self.assertIs(None, t._password)
 
672
 
 
673
        # Simulate the user entering a password
 
674
        password = 'secret'
 
675
        connection = object()
 
676
        t._set_connection(connection, password)
 
677
        self.assertIs(connection, t._get_connection())
 
678
        self.assertIs(password, t._get_credentials())
 
679
        self.assertIs(connection, c._get_connection())
 
680
        self.assertIs(password, c._get_credentials())
 
681
 
 
682
        # credentials can be updated
 
683
        new_password = 'even more secret'
 
684
        c._update_credentials(new_password)
 
685
        self.assertIs(connection, t._get_connection())
 
686
        self.assertIs(new_password, t._get_credentials())
 
687
        self.assertIs(connection, c._get_connection())
 
688
        self.assertIs(new_password, c._get_credentials())
 
689
 
 
690
 
 
691
class TestReusedTransports(TestCase):
 
692
    """Tests for transport reuse"""
 
693
 
 
694
    def test_reuse_same_transport(self):
 
695
        possible_transports = []
 
696
        t1 = get_transport('http://foo/',
 
697
                           possible_transports=possible_transports)
 
698
        self.assertEqual([t1], possible_transports)
 
699
        t2 = get_transport('http://foo/', possible_transports=[t1])
 
700
        self.assertIs(t1, t2)
 
701
 
 
702
        # Also check that final '/' are handled correctly
 
703
        t3 = get_transport('http://foo/path/')
 
704
        t4 = get_transport('http://foo/path', possible_transports=[t3])
 
705
        self.assertIs(t3, t4)
 
706
 
 
707
        t5 = get_transport('http://foo/path')
 
708
        t6 = get_transport('http://foo/path/', possible_transports=[t5])
 
709
        self.assertIs(t5, t6)
 
710
 
 
711
    def test_don_t_reuse_different_transport(self):
 
712
        t1 = get_transport('http://foo/path')
 
713
        t2 = get_transport('http://bar/path', possible_transports=[t1])
 
714
        self.assertIsNot(t1, t2)
 
715
 
 
716
 
 
717
class TestRemoteTCPTransport(TestCase):
 
718
    """Tests for bzr:// transport (RemoteTCPTransport)."""
 
719
 
 
720
    def test_relpath_with_implicit_port(self):
 
721
        """Connected transports with the same URL are the same, even if the
 
722
        port is implicit.
 
723
 
 
724
        So t.relpath(url) should always be '' if t.base is the same as url, or
 
725
        if the only difference is that one explicitly specifies the default
 
726
        port and the other doesn't specify a port.
 
727
        """
 
728
        t_implicit_port = RemoteTCPTransport('bzr://host.com/')
 
729
        self.assertEquals('', t_implicit_port.relpath('bzr://host.com/'))
 
730
        self.assertEquals('', t_implicit_port.relpath('bzr://host.com:4155/'))
 
731
        t_explicit_port = RemoteTCPTransport('bzr://host.com:4155/')
 
732
        self.assertEquals('', t_explicit_port.relpath('bzr://host.com/'))
 
733
        self.assertEquals('', t_explicit_port.relpath('bzr://host.com:4155/'))
 
734
 
 
735
    def test_construct_uses_default_port(self):
 
736
        """If no port is specified, then RemoteTCPTransport uses
 
737
        BZR_DEFAULT_PORT.
 
738
        """
 
739
        t = get_transport('bzr://host.com/')
 
740
        self.assertEquals(BZR_DEFAULT_PORT, t._port)
 
741
 
 
742
    def test_url_omits_default_port(self):
 
743
        """If a RemoteTCPTransport uses the default port, then its base URL
 
744
        will omit the port.
 
745
 
 
746
        This is like how ":80" is omitted from "http://example.com/".
 
747
        """
 
748
        t = get_transport('bzr://host.com:4155/')
 
749
        self.assertEquals('bzr://host.com/', t.base)
 
750
 
 
751
    def test_url_includes_non_default_port(self):
 
752
        """Non-default ports are included in the transport's URL.
 
753
 
 
754
        Contrast this to `test_url_omits_default_port`.
 
755
        """
 
756
        t = get_transport('bzr://host.com:666/')
 
757
        self.assertEquals('bzr://host.com:666/', t.base)
 
758
 
 
759
 
 
760
class SSHPortTestMixin(object):
 
761
    """Mixin class for testing SSH-based transports' use of ports in URLs.
 
762
    
 
763
    Unlike other connected transports, SSH-based transports (sftp, bzr+ssh)
 
764
    don't have a default port, because the user may have OpenSSH configured to
 
765
    use a non-standard port.
 
766
    """
 
767
 
 
768
    def make_url(self, netloc):
 
769
        """Make a url for the given netloc, using the scheme defined on the
 
770
        TestCase.
 
771
        """
 
772
        return '%s://%s/' % (self.scheme, netloc)
 
773
 
 
774
    def test_relpath_with_implicit_port(self):
 
775
        """SSH-based transports with the same URL are the same.
 
776
        
 
777
        Note than an unspecified port number is different to port 22 (because
 
778
        OpenSSH may be configured to use a non-standard port).
 
779
 
 
780
        So t.relpath(url) should always be '' if t.base is the same as url, but
 
781
        raise PathNotChild if the ports in t and url are not both specified (or
 
782
        both unspecified).
 
783
        """
 
784
        url_implicit_port = self.make_url('host.com')
 
785
        url_explicit_port = self.make_url('host.com:22')
 
786
 
 
787
        t_implicit_port = get_transport(url_implicit_port)
 
788
        self.assertEquals('', t_implicit_port.relpath(url_implicit_port))
 
789
        self.assertRaises(
 
790
            PathNotChild, t_implicit_port.relpath, url_explicit_port)
 
791
        
 
792
        t_explicit_port = get_transport(url_explicit_port)
 
793
        self.assertRaises(
 
794
            PathNotChild, t_explicit_port.relpath, url_implicit_port)
 
795
        self.assertEquals('', t_explicit_port.relpath(url_explicit_port))
 
796
 
 
797
    def test_construct_with_no_port(self):
 
798
        """If no port is specified, then the SSH-based transport's _port will
 
799
        be None.
 
800
        """
 
801
        t = get_transport(self.make_url('host.com'))
 
802
        self.assertEquals(None, t._port)
 
803
 
 
804
    def test_url_with_no_port(self):
 
805
        """If no port was specified, none is shown in the base URL."""
 
806
        t = get_transport(self.make_url('host.com'))
 
807
        self.assertEquals(self.make_url('host.com'), t.base)
 
808
 
 
809
    def test_url_includes_port(self):
 
810
        """An SSH-based transport's base will show the port if one was
 
811
        specified, even if that port is 22, because we do not assume 22 is the
 
812
        default port.
 
813
        """
 
814
        # 22 is the "standard" port for SFTP.
 
815
        t = get_transport(self.make_url('host.com:22'))
 
816
        self.assertEquals(self.make_url('host.com:22'), t.base)
 
817
        # 666 is not a standard port.
 
818
        t = get_transport(self.make_url('host.com:666'))
 
819
        self.assertEquals(self.make_url('host.com:666'), t.base)
 
820
 
 
821
 
 
822
class SFTPTransportPortTest(TestCase, SSHPortTestMixin):
 
823
    """Tests for sftp:// transport (SFTPTransport)."""
 
824
 
 
825
    scheme = 'sftp'
 
826
 
 
827
 
 
828
class BzrSSHTransportPortTest(TestCase, SSHPortTestMixin):
 
829
    """Tests for bzr+ssh:// transport (RemoteSSHTransport)."""
 
830
 
 
831
    scheme = 'bzr+ssh'
 
832
 
 
833
 
 
834
class TestTransportTrace(TestCase):
 
835
 
 
836
    def test_get(self):
 
837
        transport = get_transport('trace+memory://')
 
838
        self.assertIsInstance(
 
839
            transport, bzrlib.transport.trace.TransportTraceDecorator)
 
840
 
 
841
    def test_clone_preserves_activity(self):
 
842
        transport = get_transport('trace+memory://')
 
843
        transport2 = transport.clone('.')
 
844
        self.assertTrue(transport is not transport2)
 
845
        self.assertTrue(transport._activity is transport2._activity)
 
846
 
 
847
    # the following specific tests are for the operations that have made use of
 
848
    # logging in tests; we could test every single operation but doing that
 
849
    # still won't cause a test failure when the top level Transport API
 
850
    # changes; so there is little return doing that.
 
851
    def test_get(self):
 
852
        transport = get_transport('trace+memory:///')
 
853
        transport.put_bytes('foo', 'barish')
 
854
        transport.get('foo')
 
855
        expected_result = []
 
856
        # put_bytes records the bytes, not the content to avoid memory
 
857
        # pressure.
 
858
        expected_result.append(('put_bytes', 'foo', 6, None))
 
859
        # get records the file name only.
 
860
        expected_result.append(('get', 'foo'))
 
861
        self.assertEqual(expected_result, transport._activity)
 
862
 
 
863
    def test_readv(self):
 
864
        transport = get_transport('trace+memory:///')
 
865
        transport.put_bytes('foo', 'barish')
 
866
        list(transport.readv('foo', [(0, 1), (3, 2)], adjust_for_latency=True,
 
867
            upper_limit=6))
 
868
        expected_result = []
 
869
        # put_bytes records the bytes, not the content to avoid memory
 
870
        # pressure.
 
871
        expected_result.append(('put_bytes', 'foo', 6, None))
 
872
        # readv records the supplied offset request
 
873
        expected_result.append(('readv', 'foo', [(0, 1), (3, 2)], True, 6))
 
874
        self.assertEqual(expected_result, transport._activity)