← 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: 2006-10-10 07:23:07 UTC
  • mfrom: (2067.1.1 urandom-56883)
  • Revision ID: pqm@pqm.ubuntu.com-20061010072307-037a6f63da8a1bdd
(John Arbash Meinel) Handle exceptions while opening /dev/urandom

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
1
 
# Copyright (C) 2004, 2005, 2006, 2007 Canonical Ltd
 
1
# Copyright (C) 2004, 2005, 2006 by 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
21
21
from cStringIO import StringIO
22
22
 
23
23
import bzrlib
24
 
from bzrlib import (
25
 
    errors,
26
 
    urlutils,
27
 
    )
28
 
from bzrlib.errors import (ConnectionError,
29
 
                           DependencyNotPresent,
30
 
                           FileExists,
31
 
                           InvalidURLJoin,
32
 
                           NoSuchFile,
33
 
                           PathNotChild,
 
24
from bzrlib.errors import (NoSuchFile, FileExists,
34
25
                           TransportNotPossible,
35
26
                           ConnectionError,
36
27
                           DependencyNotPresent,
37
 
                           ReadError,
38
28
                           UnsupportedProtocol,
39
29
                           )
40
30
from bzrlib.tests import TestCase, TestCaseInTempDir
41
31
from bzrlib.transport import (_CoalescedOffset,
42
 
                              ConnectedTransport,
43
32
                              _get_protocol_handlers,
44
 
                              _set_protocol_handlers,
45
33
                              _get_transport_modules,
46
34
                              get_transport,
47
 
                              LateReadError,
48
35
                              register_lazy_transport,
49
 
                              register_transport_proto,
50
 
                              _clear_protocol_handlers,
 
36
                              _set_protocol_handlers,
51
37
                              Transport,
52
38
                              )
53
 
from bzrlib.transport.chroot import ChrootServer
54
39
from bzrlib.transport.memory import MemoryTransport
55
 
from bzrlib.transport.local import (LocalTransport,
56
 
                                    EmulatedWin32LocalTransport)
57
 
 
58
 
 
59
 
# TODO: Should possibly split transport-specific tests into their own files.
 
40
from bzrlib.transport.local import LocalTransport
60
41
 
61
42
 
62
43
class TestTransport(TestCase):
64
45
 
65
46
    def test__get_set_protocol_handlers(self):
66
47
        handlers = _get_protocol_handlers()
67
 
        self.assertNotEqual([], handlers.keys( ))
 
48
        self.assertNotEqual({}, handlers)
68
49
        try:
69
 
            _clear_protocol_handlers()
70
 
            self.assertEqual([], _get_protocol_handlers().keys())
 
50
            _set_protocol_handlers({})
 
51
            self.assertEqual({}, _get_protocol_handlers())
71
52
        finally:
72
53
            _set_protocol_handlers(handlers)
73
54
 
76
57
        class SampleHandler(object):
77
58
            """I exist, isnt that enough?"""
78
59
        try:
79
 
            _clear_protocol_handlers()
80
 
            register_transport_proto('foo')
 
60
            my_handlers = {}
 
61
            _set_protocol_handlers(my_handlers)
81
62
            register_lazy_transport('foo', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
82
 
            register_transport_proto('bar')
83
63
            register_lazy_transport('bar', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
84
 
            self.assertEqual([SampleHandler.__module__, 'bzrlib.transport.chroot'],
 
64
            self.assertEqual([SampleHandler.__module__],
85
65
                             _get_transport_modules())
86
66
        finally:
87
67
            _set_protocol_handlers(handlers)
90
70
        """Transport with missing dependency causes no error"""
91
71
        saved_handlers = _get_protocol_handlers()
92
72
        try:
93
 
            register_transport_proto('foo')
94
73
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
95
74
                    'BadTransportHandler')
96
75
            try:
111
90
        """Transport with missing dependency causes no error"""
112
91
        saved_handlers = _get_protocol_handlers()
113
92
        try:
114
 
            _clear_protocol_handlers()
115
 
            register_transport_proto('foo')
 
93
            _set_protocol_handlers({})
116
94
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
117
95
                    'BackupTransportHandler')
118
96
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
122
100
        finally:
123
101
            _set_protocol_handlers(saved_handlers)
124
102
 
125
 
    def test_LateReadError(self):
126
 
        """The LateReadError helper should raise on read()."""
127
 
        a_file = LateReadError('a path')
128
 
        try:
129
 
            a_file.read()
130
 
        except ReadError, error:
131
 
            self.assertEqual('a path', error.path)
132
 
        self.assertRaises(ReadError, a_file.read, 40)
133
 
        a_file.close()
134
 
 
135
 
    def test__combine_paths(self):
136
 
        t = Transport('/')
137
 
        self.assertEqual('/home/sarah/project/foo',
138
 
                         t._combine_paths('/home/sarah', 'project/foo'))
139
 
        self.assertEqual('/etc',
140
 
                         t._combine_paths('/home/sarah', '../../etc'))
141
 
        self.assertEqual('/etc',
142
 
                         t._combine_paths('/home/sarah', '../../../etc'))
143
 
        self.assertEqual('/etc',
144
 
                         t._combine_paths('/home/sarah', '/etc'))
145
 
 
146
 
    def test_local_abspath_non_local_transport(self):
147
 
        # the base implementation should throw
148
 
        t = MemoryTransport()
149
 
        e = self.assertRaises(errors.NotLocalUrl, t.local_abspath, 't')
150
 
        self.assertEqual('memory:///t is not a local path.', str(e))
151
 
 
152
103
 
153
104
class TestCoalesceOffsets(TestCase):
154
105
    
228
179
        self.assertEqual("memory:///", transport.base)
229
180
        self.assertEqual("memory:///", transport.abspath('/'))
230
181
 
231
 
    def test_abspath_of_relpath_starting_at_root(self):
 
182
    def test_relpath(self):
232
183
        transport = MemoryTransport()
233
 
        self.assertEqual("memory:///foo", transport.abspath('/foo'))
234
184
 
235
185
    def test_append_and_get(self):
236
186
        transport = MemoryTransport()
269
219
        transport.append_bytes('foo', 'content')
270
220
        self.assertEquals(True, transport.has('foo'))
271
221
 
272
 
    def test_list_dir(self):
273
 
        transport = MemoryTransport()
274
 
        transport.put_bytes('foo', 'content')
275
 
        transport.mkdir('dir')
276
 
        transport.put_bytes('dir/subfoo', 'content')
277
 
        transport.put_bytes('dirlike', 'content')
278
 
 
279
 
        self.assertEquals(['dir', 'dirlike', 'foo'], sorted(transport.list_dir('.')))
280
 
        self.assertEquals(['subfoo'], sorted(transport.list_dir('dir')))
281
 
 
282
222
    def test_mkdir(self):
283
223
        transport = MemoryTransport()
284
224
        transport.mkdir('dir')
317
257
        self.assertEqual(7, transport.stat('foo').st_size)
318
258
        self.assertEqual(6, transport.stat('bar').st_size)
319
259
 
320
 
 
321
 
class ChrootDecoratorTransportTest(TestCase):
322
 
    """Chroot decoration specific tests."""
323
 
 
324
 
    def test_abspath(self):
325
 
        # The abspath is always relative to the chroot_url.
326
 
        server = ChrootServer(get_transport('memory:///foo/bar/'))
327
 
        server.setUp()
328
 
        transport = get_transport(server.get_url())
329
 
        self.assertEqual(server.get_url(), transport.abspath('/'))
330
 
 
331
 
        subdir_transport = transport.clone('subdir')
332
 
        self.assertEqual(server.get_url(), subdir_transport.abspath('/'))
333
 
        server.tearDown()
334
 
 
335
 
    def test_clone(self):
336
 
        server = ChrootServer(get_transport('memory:///foo/bar/'))
337
 
        server.setUp()
338
 
        transport = get_transport(server.get_url())
339
 
        # relpath from root and root path are the same
340
 
        relpath_cloned = transport.clone('foo')
341
 
        abspath_cloned = transport.clone('/foo')
342
 
        self.assertEqual(server, relpath_cloned.server)
343
 
        self.assertEqual(server, abspath_cloned.server)
344
 
        server.tearDown()
345
 
    
346
 
    def test_chroot_url_preserves_chroot(self):
347
 
        """Calling get_transport on a chroot transport's base should produce a
348
 
        transport with exactly the same behaviour as the original chroot
349
 
        transport.
350
 
 
351
 
        This is so that it is not possible to escape a chroot by doing::
352
 
            url = chroot_transport.base
353
 
            parent_url = urlutils.join(url, '..')
354
 
            new_transport = get_transport(parent_url)
355
 
        """
356
 
        server = ChrootServer(get_transport('memory:///path/subpath'))
357
 
        server.setUp()
358
 
        transport = get_transport(server.get_url())
359
 
        new_transport = get_transport(transport.base)
360
 
        self.assertEqual(transport.server, new_transport.server)
361
 
        self.assertEqual(transport.base, new_transport.base)
362
 
        server.tearDown()
363
260
        
364
 
    def test_urljoin_preserves_chroot(self):
365
 
        """Using urlutils.join(url, '..') on a chroot URL should not produce a
366
 
        URL that escapes the intended chroot.
367
 
 
368
 
        This is so that it is not possible to escape a chroot by doing::
369
 
            url = chroot_transport.base
370
 
            parent_url = urlutils.join(url, '..')
371
 
            new_transport = get_transport(parent_url)
372
 
        """
373
 
        server = ChrootServer(get_transport('memory:///path/'))
374
 
        server.setUp()
375
 
        transport = get_transport(server.get_url())
376
 
        self.assertRaises(
377
 
            InvalidURLJoin, urlutils.join, transport.base, '..')
378
 
        server.tearDown()
379
 
 
380
 
 
381
 
class ChrootServerTest(TestCase):
382
 
 
383
 
    def test_construct(self):
384
 
        backing_transport = MemoryTransport()
385
 
        server = ChrootServer(backing_transport)
386
 
        self.assertEqual(backing_transport, server.backing_transport)
387
 
 
388
 
    def test_setUp(self):
389
 
        backing_transport = MemoryTransport()
390
 
        server = ChrootServer(backing_transport)
391
 
        server.setUp()
392
 
        self.assertTrue(server.scheme in _get_protocol_handlers().keys())
393
 
 
394
 
    def test_tearDown(self):
395
 
        backing_transport = MemoryTransport()
396
 
        server = ChrootServer(backing_transport)
397
 
        server.setUp()
398
 
        server.tearDown()
399
 
        self.assertFalse(server.scheme in _get_protocol_handlers().keys())
400
 
 
401
 
    def test_get_url(self):
402
 
        backing_transport = MemoryTransport()
403
 
        server = ChrootServer(backing_transport)
404
 
        server.setUp()
405
 
        self.assertEqual('chroot-%d:///' % id(server), server.get_url())
406
 
        server.tearDown()
407
 
 
408
 
 
409
261
class ReadonlyDecoratorTransportTest(TestCase):
410
262
    """Readonly decoration specific tests."""
411
263
 
418
270
        self.assertEqual(True, transport.is_readonly())
419
271
 
420
272
    def test_http_parameters(self):
421
 
        from bzrlib.tests.HttpServer import HttpServer
422
273
        import bzrlib.transport.readonly as readonly
 
274
        from bzrlib.transport.http import HttpServer
423
275
        # connect to . via http which is not listable
424
276
        server = HttpServer()
425
277
        server.setUp()
453
305
    def test_http_parameters(self):
454
306
        # the listable, should_cache and is_readonly parameters
455
307
        # are not changed by the fakenfs decorator
456
 
        from bzrlib.tests.HttpServer import HttpServer
 
308
        from bzrlib.transport.http import HttpServer
457
309
        # connect to . via http which is not listable
458
310
        server = HttpServer()
459
311
        server.setUp()
473
325
        server = fakenfs.FakeNFSServer()
474
326
        server.setUp()
475
327
        try:
476
 
            # the url should be decorated appropriately
477
 
            self.assertStartsWith(server.get_url(), 'fakenfs+')
 
328
            # the server should be a relpath localhost server
 
329
            self.assertEqual(server.get_url(), 'fakenfs+.')
478
330
            # and we should be able to get a transport for it
479
331
            transport = get_transport(server.get_url())
480
332
            # which must be a FakeNFSTransportDecorator instance.
489
341
        transport = self.get_nfs_transport('.')
490
342
        self.build_tree(['from/', 'from/foo', 'to/', 'to/bar'],
491
343
                        transport=transport)
492
 
        self.assertRaises(errors.ResourceBusy,
 
344
        self.assertRaises(bzrlib.errors.ResourceBusy,
493
345
                          transport.rename, 'from', 'to')
494
346
 
495
347
 
553
405
        super(TestTransportImplementation, self).setUp()
554
406
        self._server = self.transport_server()
555
407
        self._server.setUp()
556
 
        self.addCleanup(self._server.tearDown)
557
 
 
558
 
    def get_transport(self, relpath=None):
559
 
        """Return a connected transport to the local directory.
560
 
 
561
 
        :param relpath: a path relative to the base url.
562
 
        """
 
408
 
 
409
    def tearDown(self):
 
410
        super(TestTransportImplementation, self).tearDown()
 
411
        self._server.tearDown()
 
412
        
 
413
    def get_transport(self):
 
414
        """Return a connected transport to the local directory."""
563
415
        base_url = self._server.get_url()
564
 
        url = self._adjust_url(base_url, relpath)
565
416
        # try getting the transport via the regular interface:
566
 
        t = get_transport(url)
567
 
        # vila--20070607 if the following are commented out the test suite
568
 
        # still pass. Is this really still needed or was it a forgotten
569
 
        # temporary fix ?
 
417
        t = get_transport(base_url)
570
418
        if not isinstance(t, self.transport_class):
571
419
            # we did not get the correct transport class type. Override the
572
420
            # regular connection behaviour by direct construction.
573
 
            t = self.transport_class(url)
 
421
            t = self.transport_class(base_url)
574
422
        return t
575
 
 
576
 
 
577
 
class TestLocalTransports(TestCase):
578
 
 
579
 
    def test_get_transport_from_abspath(self):
580
 
        here = os.path.abspath('.')
581
 
        t = get_transport(here)
582
 
        self.assertIsInstance(t, LocalTransport)
583
 
        self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
584
 
 
585
 
    def test_get_transport_from_relpath(self):
586
 
        here = os.path.abspath('.')
587
 
        t = get_transport('.')
588
 
        self.assertIsInstance(t, LocalTransport)
589
 
        self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
590
 
 
591
 
    def test_get_transport_from_local_url(self):
592
 
        here = os.path.abspath('.')
593
 
        here_url = urlutils.local_path_to_url(here) + '/'
594
 
        t = get_transport(here_url)
595
 
        self.assertIsInstance(t, LocalTransport)
596
 
        self.assertEquals(t.base, here_url)
597
 
 
598
 
    def test_local_abspath(self):
599
 
        here = os.path.abspath('.')
600
 
        t = get_transport(here)
601
 
        self.assertEquals(t.local_abspath(''), here)
602
 
 
603
 
 
604
 
class TestWin32LocalTransport(TestCase):
605
 
 
606
 
    def test_unc_clone_to_root(self):
607
 
        # Win32 UNC path like \\HOST\path
608
 
        # clone to root should stop at least at \\HOST part
609
 
        # not on \\
610
 
        t = EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
611
 
        for i in xrange(4):
612
 
            t = t.clone('..')
613
 
        self.assertEquals(t.base, 'file://HOST/')
614
 
        # make sure we reach the root
615
 
        t = t.clone('..')
616
 
        self.assertEquals(t.base, 'file://HOST/')
617
 
 
618
 
class TestConnectedTransport(TestCase):
619
 
    """Tests for connected to remote server transports"""
620
 
 
621
 
    def test_parse_url(self):
622
 
        t = ConnectedTransport('sftp://simple.example.com/home/source')
623
 
        self.assertEquals(t._host, 'simple.example.com')
624
 
        self.assertEquals(t._port, None)
625
 
        self.assertEquals(t._path, '/home/source/')
626
 
        self.failUnless(t._user is None)
627
 
        self.failUnless(t._password is None)
628
 
 
629
 
        self.assertEquals(t.base, 'sftp://simple.example.com/home/source/')
630
 
 
631
 
    def test_parse_quoted_url(self):
632
 
        t = ConnectedTransport('http://ro%62ey:h%40t@ex%41mple.com:2222/path')
633
 
        self.assertEquals(t._host, 'exAmple.com')
634
 
        self.assertEquals(t._port, 2222)
635
 
        self.assertEquals(t._user, 'robey')
636
 
        self.assertEquals(t._password, 'h@t')
637
 
        self.assertEquals(t._path, '/path/')
638
 
 
639
 
        # Base should not keep track of the password
640
 
        self.assertEquals(t.base, 'http://robey@exAmple.com:2222/path/')
641
 
 
642
 
    def test_parse_invalid_url(self):
643
 
        self.assertRaises(errors.InvalidURL,
644
 
                          ConnectedTransport,
645
 
                          'sftp://lily.org:~janneke/public/bzr/gub')
646
 
 
647
 
    def test_relpath(self):
648
 
        t = ConnectedTransport('sftp://user@host.com/abs/path')
649
 
 
650
 
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
651
 
        self.assertRaises(errors.PathNotChild, t.relpath,
652
 
                          'http://user@host.com/abs/path/sub')
653
 
        self.assertRaises(errors.PathNotChild, t.relpath,
654
 
                          'sftp://user2@host.com/abs/path/sub')
655
 
        self.assertRaises(errors.PathNotChild, t.relpath,
656
 
                          'sftp://user@otherhost.com/abs/path/sub')
657
 
        self.assertRaises(errors.PathNotChild, t.relpath,
658
 
                          'sftp://user@host.com:33/abs/path/sub')
659
 
        # Make sure it works when we don't supply a username
660
 
        t = ConnectedTransport('sftp://host.com/abs/path')
661
 
        self.assertEquals(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
662
 
 
663
 
        # Make sure it works when parts of the path will be url encoded
664
 
        t = ConnectedTransport('sftp://host.com/dev/%path')
665
 
        self.assertEquals(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
666
 
 
667
 
    def test_connection_sharing_propagate_credentials(self):
668
 
        t = ConnectedTransport('foo://user@host.com/abs/path')
669
 
        self.assertIs(None, t._get_connection())
670
 
        self.assertIs(None, t._password)
671
 
        c = t.clone('subdir')
672
 
        self.assertEquals(None, c._get_connection())
673
 
        self.assertIs(None, t._password)
674
 
 
675
 
        # Simulate the user entering a password
676
 
        password = 'secret'
677
 
        connection = object()
678
 
        t._set_connection(connection, password)
679
 
        self.assertIs(connection, t._get_connection())
680
 
        self.assertIs(password, t._get_credentials())
681
 
        self.assertIs(connection, c._get_connection())
682
 
        self.assertIs(password, c._get_credentials())
683
 
 
684
 
        # credentials can be updated
685
 
        new_password = 'even more secret'
686
 
        c._update_credentials(new_password)
687
 
        self.assertIs(connection, t._get_connection())
688
 
        self.assertIs(new_password, t._get_credentials())
689
 
        self.assertIs(connection, c._get_connection())
690
 
        self.assertIs(new_password, c._get_credentials())
691
 
 
692
 
 
693
 
class TestReusedTransports(TestCase):
694
 
    """Tests for transport reuse"""
695
 
 
696
 
    def test_reuse_same_transport(self):
697
 
        t1 = get_transport('http://foo/')
698
 
        t2 = get_transport('http://foo/', possible_transports=[t1])
699
 
        self.assertIs(t1, t2)
700
 
 
701
 
        # Also check that final '/' are handled correctly
702
 
        t3 = get_transport('http://foo/path/')
703
 
        t4 = get_transport('http://foo/path', possible_transports=[t3])
704
 
        self.assertIs(t3, t4)
705
 
 
706
 
        t5 = get_transport('http://foo/path')
707
 
        t6 = get_transport('http://foo/path/', possible_transports=[t5])
708
 
        self.assertIs(t5, t6)
709
 
 
710
 
    def test_don_t_reuse_different_transport(self):
711
 
        t1 = get_transport('http://foo/path')
712
 
        t2 = get_transport('http://bar/path', possible_transports=[t1])
713
 
        self.assertIsNot(t1, t2)
714
 
 
715
 
 
716
 
def get_test_permutations():
717
 
    """Return transport permutations to be used in testing.
718
 
 
719
 
    This module registers some transports, but they're only for testing
720
 
    registration.  We don't really want to run all the transport tests against
721
 
    them.
722
 
    """
723
 
    return []