← 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: Vincent Ladeuil
  • Date: 2008-01-03 08:49:38 UTC
  • mfrom: (3111.1.31 175524)
  • mto: This revision was merged to the branch mainline in revision 3158.
  • Revision ID: v.ladeuil+lp@free.fr-20080103084938-7kvurk5uvde2ui54
https://launchpad.net/bugs/175524
Fix bug #175524, http test servers are 1.1 compliant

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
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
21
from bzrlib import (
25
22
    errors,
 
23
    osutils,
26
24
    urlutils,
27
25
    )
28
 
from bzrlib.errors import (ConnectionError,
29
 
                           DependencyNotPresent,
 
26
from bzrlib.errors import (DependencyNotPresent,
30
27
                           FileExists,
31
28
                           InvalidURLJoin,
32
29
                           NoSuchFile,
33
30
                           PathNotChild,
34
 
                           TransportNotPossible,
 
31
                           ReadError,
35
32
                           UnsupportedProtocol,
36
33
                           )
37
34
from bzrlib.tests import TestCase, TestCaseInTempDir
38
 
from bzrlib.transport import (_CoalescedOffset,
 
35
from bzrlib.transport import (_clear_protocol_handlers,
 
36
                              _CoalescedOffset,
 
37
                              ConnectedTransport,
39
38
                              _get_protocol_handlers,
40
39
                              _set_protocol_handlers,
41
40
                              _get_transport_modules,
42
41
                              get_transport,
 
42
                              LateReadError,
43
43
                              register_lazy_transport,
44
44
                              register_transport_proto,
45
 
                              _clear_protocol_handlers,
46
45
                              Transport,
47
46
                              )
48
47
from bzrlib.transport.chroot import ChrootServer
68
67
 
69
68
    def test_get_transport_modules(self):
70
69
        handlers = _get_protocol_handlers()
 
70
        # don't pollute the current handlers
 
71
        _clear_protocol_handlers()
71
72
        class SampleHandler(object):
72
73
            """I exist, isnt that enough?"""
73
74
        try:
74
75
            _clear_protocol_handlers()
75
76
            register_transport_proto('foo')
76
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
 
77
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
 
78
                                    'TestTransport.SampleHandler')
77
79
            register_transport_proto('bar')
78
 
            register_lazy_transport('bar', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
79
 
            self.assertEqual([SampleHandler.__module__, 'bzrlib.transport.chroot'],
 
80
            register_lazy_transport('bar', 'bzrlib.tests.test_transport',
 
81
                                    'TestTransport.SampleHandler')
 
82
            self.assertEqual([SampleHandler.__module__,
 
83
                              'bzrlib.transport.chroot'],
80
84
                             _get_transport_modules())
81
85
        finally:
82
86
            _set_protocol_handlers(handlers)
84
88
    def test_transport_dependency(self):
85
89
        """Transport with missing dependency causes no error"""
86
90
        saved_handlers = _get_protocol_handlers()
 
91
        # don't pollute the current handlers
 
92
        _clear_protocol_handlers()
87
93
        try:
88
94
            register_transport_proto('foo')
89
95
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
117
123
        finally:
118
124
            _set_protocol_handlers(saved_handlers)
119
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
 
120
136
    def test__combine_paths(self):
121
137
        t = Transport('/')
122
138
        self.assertEqual('/home/sarah/project/foo',
136
152
 
137
153
 
138
154
class TestCoalesceOffsets(TestCase):
139
 
    
140
 
    def check(self, expected, offsets, limit=0, fudge=0):
 
155
 
 
156
    def check(self, expected, offsets, limit=0, max_size=0, fudge=0):
141
157
        coalesce = Transport._coalesce_offsets
142
158
        exp = [_CoalescedOffset(*x) for x in expected]
143
 
        out = list(coalesce(offsets, limit=limit, fudge_factor=fudge))
 
159
        out = list(coalesce(offsets, limit=limit, fudge_factor=fudge,
 
160
                            max_size=max_size))
144
161
        self.assertEqual(exp, out)
145
162
 
146
163
    def test_coalesce_empty(self):
153
170
        self.check([(0, 10, [(0, 10)]),
154
171
                    (20, 10, [(0, 10)]),
155
172
                   ], [(0, 10), (20, 10)])
156
 
            
 
173
 
157
174
    def test_coalesce_unsorted(self):
158
175
        self.check([(20, 10, [(0, 10)]),
159
176
                    (0, 10, [(0, 10)]),
163
180
        self.check([(0, 20, [(0, 10), (10, 10)])],
164
181
                   [(0, 10), (10, 10)])
165
182
 
 
183
    # XXX: scary, http.readv() can't handle that --vila20071209
166
184
    def test_coalesce_overlapped(self):
167
185
        self.check([(0, 15, [(0, 10), (5, 10)])],
168
186
                   [(0, 10), (5, 10)])
192
210
                   ], [(10, 10), (30, 10), (100, 10)],
193
211
                   fudge=10
194
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
                  )
195
227
 
196
228
 
197
229
class TestMemoryTransport(TestCase):
283
315
    def test_parameters(self):
284
316
        transport = MemoryTransport()
285
317
        self.assertEqual(True, transport.listable())
286
 
        self.assertEqual(False, transport.should_cache())
287
318
        self.assertEqual(False, transport.is_readonly())
288
319
 
289
320
    def test_iter_files_recursive(self):
399
430
        # connect to . in readonly mode
400
431
        transport = readonly.ReadonlyTransportDecorator('readonly+.')
401
432
        self.assertEqual(True, transport.listable())
402
 
        self.assertEqual(False, transport.should_cache())
403
433
        self.assertEqual(True, transport.is_readonly())
404
434
 
405
435
    def test_http_parameters(self):
406
 
        from bzrlib.tests.HttpServer import HttpServer
 
436
        from bzrlib.tests.http_server import HttpServer
407
437
        import bzrlib.transport.readonly as readonly
408
 
        # connect to . via http which is not listable
 
438
        # connect to '.' via http which is not listable
409
439
        server = HttpServer()
410
440
        server.setUp()
411
441
        try:
413
443
            self.failUnless(isinstance(transport,
414
444
                                       readonly.ReadonlyTransportDecorator))
415
445
            self.assertEqual(False, transport.listable())
416
 
            self.assertEqual(True, transport.should_cache())
417
446
            self.assertEqual(True, transport.is_readonly())
418
447
        finally:
419
448
            server.tearDown()
428
457
        return fakenfs.FakeNFSTransportDecorator('fakenfs+' + url)
429
458
 
430
459
    def test_local_parameters(self):
431
 
        # the listable, should_cache and is_readonly parameters
 
460
        # the listable and is_readonly parameters
432
461
        # are not changed by the fakenfs decorator
433
462
        transport = self.get_nfs_transport('.')
434
463
        self.assertEqual(True, transport.listable())
435
 
        self.assertEqual(False, transport.should_cache())
436
464
        self.assertEqual(False, transport.is_readonly())
437
465
 
438
466
    def test_http_parameters(self):
439
 
        # the listable, should_cache and is_readonly parameters
 
467
        # the listable and is_readonly parameters
440
468
        # are not changed by the fakenfs decorator
441
 
        from bzrlib.tests.HttpServer import HttpServer
442
 
        # 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
443
471
        server = HttpServer()
444
472
        server.setUp()
445
473
        try:
447
475
            self.assertIsInstance(
448
476
                transport, bzrlib.transport.fakenfs.FakeNFSTransportDecorator)
449
477
            self.assertEqual(False, transport.listable())
450
 
            self.assertEqual(True, transport.should_cache())
451
478
            self.assertEqual(True, transport.is_readonly())
452
479
        finally:
453
480
            server.tearDown()
540
567
        self._server.setUp()
541
568
        self.addCleanup(self._server.tearDown)
542
569
 
543
 
    def get_transport(self):
544
 
        """Return a connected transport to the local directory."""
 
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
        """
545
575
        base_url = self._server.get_url()
 
576
        url = self._adjust_url(base_url, relpath)
546
577
        # try getting the transport via the regular interface:
547
 
        t = get_transport(base_url)
 
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 ?
548
582
        if not isinstance(t, self.transport_class):
549
583
            # we did not get the correct transport class type. Override the
550
584
            # regular connection behaviour by direct construction.
551
 
            t = self.transport_class(base_url)
 
585
            t = self.transport_class(url)
552
586
        return t
553
587
 
554
588
 
555
589
class TestLocalTransports(TestCase):
556
590
 
557
591
    def test_get_transport_from_abspath(self):
558
 
        here = os.path.abspath('.')
 
592
        here = osutils.abspath('.')
559
593
        t = get_transport(here)
560
594
        self.assertIsInstance(t, LocalTransport)
561
595
        self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
562
596
 
563
597
    def test_get_transport_from_relpath(self):
564
 
        here = os.path.abspath('.')
 
598
        here = osutils.abspath('.')
565
599
        t = get_transport('.')
566
600
        self.assertIsInstance(t, LocalTransport)
567
601
        self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
568
602
 
569
603
    def test_get_transport_from_local_url(self):
570
 
        here = os.path.abspath('.')
 
604
        here = osutils.abspath('.')
571
605
        here_url = urlutils.local_path_to_url(here) + '/'
572
606
        t = get_transport(here_url)
573
607
        self.assertIsInstance(t, LocalTransport)
574
608
        self.assertEquals(t.base, here_url)
575
609
 
576
610
    def test_local_abspath(self):
577
 
        here = os.path.abspath('.')
 
611
        here = osutils.abspath('.')
578
612
        t = get_transport(here)
579
613
        self.assertEquals(t.local_abspath(''), here)
580
614
 
594
628
        self.assertEquals(t.base, 'file://HOST/')
595
629
 
596
630
 
597
 
def get_test_permutations():
598
 
    """Return transport permutations to be used in testing.
599
 
 
600
 
    This module registers some transports, but they're only for testing
601
 
    registration.  We don't really want to run all the transport tests against
602
 
    them.
603
 
    """
604
 
    return []
 
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_quoted_url(self):
 
645
        t = ConnectedTransport('http://ro%62ey:h%40t@ex%41mple.com:2222/path')
 
646
        self.assertEquals(t._host, 'exAmple.com')
 
647
        self.assertEquals(t._port, 2222)
 
648
        self.assertEquals(t._user, 'robey')
 
649
        self.assertEquals(t._password, 'h@t')
 
650
        self.assertEquals(t._path, '/path/')
 
651
 
 
652
        # Base should not keep track of the password
 
653
        self.assertEquals(t.base, 'http://robey@exAmple.com:2222/path/')
 
654
 
 
655
    def test_parse_invalid_url(self):
 
656
        self.assertRaises(errors.InvalidURL,
 
657
                          ConnectedTransport,
 
658
                          'sftp://lily.org:~janneke/public/bzr/gub')
 
659
 
 
660
    def test_relpath(self):
 
661
        t = ConnectedTransport('sftp://user@host.com/abs/path')
 
662
 
 
663
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
 
664
        self.assertRaises(errors.PathNotChild, t.relpath,
 
665
                          'http://user@host.com/abs/path/sub')
 
666
        self.assertRaises(errors.PathNotChild, t.relpath,
 
667
                          'sftp://user2@host.com/abs/path/sub')
 
668
        self.assertRaises(errors.PathNotChild, t.relpath,
 
669
                          'sftp://user@otherhost.com/abs/path/sub')
 
670
        self.assertRaises(errors.PathNotChild, t.relpath,
 
671
                          'sftp://user@host.com:33/abs/path/sub')
 
672
        # Make sure it works when we don't supply a username
 
673
        t = ConnectedTransport('sftp://host.com/abs/path')
 
674
        self.assertEquals(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
 
675
 
 
676
        # Make sure it works when parts of the path will be url encoded
 
677
        t = ConnectedTransport('sftp://host.com/dev/%path')
 
678
        self.assertEquals(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
 
679
 
 
680
    def test_connection_sharing_propagate_credentials(self):
 
681
        t = ConnectedTransport('ftp://user@host.com/abs/path')
 
682
        self.assertEquals('user', t._user)
 
683
        self.assertEquals('host.com', t._host)
 
684
        self.assertIs(None, t._get_connection())
 
685
        self.assertIs(None, t._password)
 
686
        c = t.clone('subdir')
 
687
        self.assertIs(None, c._get_connection())
 
688
        self.assertIs(None, t._password)
 
689
 
 
690
        # Simulate the user entering a password
 
691
        password = 'secret'
 
692
        connection = object()
 
693
        t._set_connection(connection, password)
 
694
        self.assertIs(connection, t._get_connection())
 
695
        self.assertIs(password, t._get_credentials())
 
696
        self.assertIs(connection, c._get_connection())
 
697
        self.assertIs(password, c._get_credentials())
 
698
 
 
699
        # credentials can be updated
 
700
        new_password = 'even more secret'
 
701
        c._update_credentials(new_password)
 
702
        self.assertIs(connection, t._get_connection())
 
703
        self.assertIs(new_password, t._get_credentials())
 
704
        self.assertIs(connection, c._get_connection())
 
705
        self.assertIs(new_password, c._get_credentials())
 
706
 
 
707
 
 
708
class TestReusedTransports(TestCase):
 
709
    """Tests for transport reuse"""
 
710
 
 
711
    def test_reuse_same_transport(self):
 
712
        possible_transports = []
 
713
        t1 = get_transport('http://foo/',
 
714
                           possible_transports=possible_transports)
 
715
        self.assertEqual([t1], possible_transports)
 
716
        t2 = get_transport('http://foo/', possible_transports=[t1])
 
717
        self.assertIs(t1, t2)
 
718
 
 
719
        # Also check that final '/' are handled correctly
 
720
        t3 = get_transport('http://foo/path/')
 
721
        t4 = get_transport('http://foo/path', possible_transports=[t3])
 
722
        self.assertIs(t3, t4)
 
723
 
 
724
        t5 = get_transport('http://foo/path')
 
725
        t6 = get_transport('http://foo/path/', possible_transports=[t5])
 
726
        self.assertIs(t5, t6)
 
727
 
 
728
    def test_don_t_reuse_different_transport(self):
 
729
        t1 = get_transport('http://foo/path')
 
730
        t2 = get_transport('http://bar/path', possible_transports=[t1])
 
731
        self.assertIsNot(t1, t2)
 
732
 
 
733
 
 
734
class TestTransportTrace(TestCase):
 
735
 
 
736
    def test_get(self):
 
737
        transport = get_transport('trace+memory://')
 
738
        self.assertIsInstance(
 
739
            transport, bzrlib.transport.trace.TransportTraceDecorator)
 
740
 
 
741
    def test_clone_preserves_activity(self):
 
742
        transport = get_transport('trace+memory://')
 
743
        transport2 = transport.clone('.')
 
744
        self.assertTrue(transport is not transport2)
 
745
        self.assertTrue(transport._activity is transport2._activity)
 
746
 
 
747
    # the following specific tests are for the operations that have made use of
 
748
    # logging in tests; we could test every single operation but doing that
 
749
    # still won't cause a test failure when the top level Transport API
 
750
    # changes; so there is little return doing that.
 
751
    def test_get(self):
 
752
        transport = get_transport('trace+memory:///')
 
753
        transport.put_bytes('foo', 'barish')
 
754
        transport.get('foo')
 
755
        expected_result = []
 
756
        # put_bytes records the bytes, not the content to avoid memory
 
757
        # pressure.
 
758
        expected_result.append(('put_bytes', 'foo', 6, None))
 
759
        # get records the file name only.
 
760
        expected_result.append(('get', 'foo'))
 
761
        self.assertEqual(expected_result, transport._activity)
 
762
 
 
763
    def test_readv(self):
 
764
        transport = get_transport('trace+memory:///')
 
765
        transport.put_bytes('foo', 'barish')
 
766
        list(transport.readv('foo', [(0, 1), (3, 2)], adjust_for_latency=True,
 
767
            upper_limit=6))
 
768
        expected_result = []
 
769
        # put_bytes records the bytes, not the content to avoid memory
 
770
        # pressure.
 
771
        expected_result.append(('put_bytes', 'foo', 6, None))
 
772
        # readv records the supplied offset request
 
773
        expected_result.append(('readv', 'foo', [(0, 1), (3, 2)], True, 6))
 
774
        self.assertEqual(expected_result, transport._activity)