← 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: Martin Pool
  • Date: 2010-09-14 09:47:23 UTC
  • mto: (5452.4.1 jam-integration)
  • mto: This revision was merged to the branch mainline in revision 5476.
  • Revision ID: mbp@sourcefrog.net-20100914094723-2dn6e9q5ktpa1m8r
Update existing script tests to not ignore their output

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
1
 
# Copyright (C) 2005-2011 Canonical Ltd
 
1
# Copyright (C) 2005-2010 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
16
16
 
17
17
 
18
18
from cStringIO import StringIO
 
19
import os
19
20
import subprocess
20
21
import sys
21
22
import threading
30
31
from bzrlib.transport import (
31
32
    chroot,
32
33
    fakenfs,
33
 
    http,
34
34
    local,
35
35
    memory,
36
36
    pathfilter,
48
48
class TestTransport(tests.TestCase):
49
49
    """Test the non transport-concrete class functionality."""
50
50
 
 
51
    # FIXME: These tests should use addCleanup() and/or overrideAttr() instead
 
52
    # of try/finally -- vila 20100205
 
53
 
51
54
    def test__get_set_protocol_handlers(self):
52
55
        handlers = transport._get_protocol_handlers()
53
 
        self.assertNotEqual([], handlers.keys())
54
 
        transport._clear_protocol_handlers()
55
 
        self.addCleanup(transport._set_protocol_handlers, handlers)
56
 
        self.assertEqual([], transport._get_protocol_handlers().keys())
 
56
        self.assertNotEqual([], handlers.keys( ))
 
57
        try:
 
58
            transport._clear_protocol_handlers()
 
59
            self.assertEqual([], transport._get_protocol_handlers().keys())
 
60
        finally:
 
61
            transport._set_protocol_handlers(handlers)
57
62
 
58
63
    def test_get_transport_modules(self):
59
64
        handlers = transport._get_protocol_handlers()
60
 
        self.addCleanup(transport._set_protocol_handlers, handlers)
61
65
        # don't pollute the current handlers
62
66
        transport._clear_protocol_handlers()
63
 
 
64
67
        class SampleHandler(object):
65
68
            """I exist, isnt that enough?"""
66
 
        transport._clear_protocol_handlers()
67
 
        transport.register_transport_proto('foo')
68
 
        transport.register_lazy_transport('foo',
69
 
                                            'bzrlib.tests.test_transport',
70
 
                                            'TestTransport.SampleHandler')
71
 
        transport.register_transport_proto('bar')
72
 
        transport.register_lazy_transport('bar',
73
 
                                            'bzrlib.tests.test_transport',
74
 
                                            'TestTransport.SampleHandler')
75
 
        self.assertEqual([SampleHandler.__module__,
76
 
                            'bzrlib.transport.chroot',
77
 
                            'bzrlib.transport.pathfilter'],
78
 
                            transport._get_transport_modules())
 
69
        try:
 
70
            transport._clear_protocol_handlers()
 
71
            transport.register_transport_proto('foo')
 
72
            transport.register_lazy_transport('foo',
 
73
                                              'bzrlib.tests.test_transport',
 
74
                                              'TestTransport.SampleHandler')
 
75
            transport.register_transport_proto('bar')
 
76
            transport.register_lazy_transport('bar',
 
77
                                              'bzrlib.tests.test_transport',
 
78
                                              'TestTransport.SampleHandler')
 
79
            self.assertEqual([SampleHandler.__module__,
 
80
                              'bzrlib.transport.chroot',
 
81
                              'bzrlib.transport.pathfilter'],
 
82
                             transport._get_transport_modules())
 
83
        finally:
 
84
            transport._set_protocol_handlers(handlers)
79
85
 
80
86
    def test_transport_dependency(self):
81
87
        """Transport with missing dependency causes no error"""
82
88
        saved_handlers = transport._get_protocol_handlers()
83
 
        self.addCleanup(transport._set_protocol_handlers, saved_handlers)
84
89
        # don't pollute the current handlers
85
90
        transport._clear_protocol_handlers()
86
 
        transport.register_transport_proto('foo')
87
 
        transport.register_lazy_transport(
88
 
            'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
89
91
        try:
90
 
            transport.get_transport('foo://fooserver/foo')
91
 
        except errors.UnsupportedProtocol, e:
92
 
            e_str = str(e)
93
 
            self.assertEquals('Unsupported protocol'
94
 
                                ' for url "foo://fooserver/foo":'
95
 
                                ' Unable to import library "some_lib":'
96
 
                                ' testing missing dependency', str(e))
97
 
        else:
98
 
            self.fail('Did not raise UnsupportedProtocol')
 
92
            transport.register_transport_proto('foo')
 
93
            transport.register_lazy_transport(
 
94
                'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
 
95
            try:
 
96
                transport.get_transport('foo://fooserver/foo')
 
97
            except errors.UnsupportedProtocol, e:
 
98
                e_str = str(e)
 
99
                self.assertEquals('Unsupported protocol'
 
100
                                  ' for url "foo://fooserver/foo":'
 
101
                                  ' Unable to import library "some_lib":'
 
102
                                  ' testing missing dependency', str(e))
 
103
            else:
 
104
                self.fail('Did not raise UnsupportedProtocol')
 
105
        finally:
 
106
            # restore original values
 
107
            transport._set_protocol_handlers(saved_handlers)
99
108
 
100
109
    def test_transport_fallback(self):
101
110
        """Transport with missing dependency causes no error"""
102
111
        saved_handlers = transport._get_protocol_handlers()
103
 
        self.addCleanup(transport._set_protocol_handlers, saved_handlers)
104
 
        transport._clear_protocol_handlers()
105
 
        transport.register_transport_proto('foo')
106
 
        transport.register_lazy_transport(
107
 
            'foo', 'bzrlib.tests.test_transport', 'BackupTransportHandler')
108
 
        transport.register_lazy_transport(
109
 
            'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
110
 
        t = transport.get_transport('foo://fooserver/foo')
111
 
        self.assertTrue(isinstance(t, BackupTransportHandler))
 
112
        try:
 
113
            transport._clear_protocol_handlers()
 
114
            transport.register_transport_proto('foo')
 
115
            transport.register_lazy_transport(
 
116
                'foo', 'bzrlib.tests.test_transport', 'BackupTransportHandler')
 
117
            transport.register_lazy_transport(
 
118
                'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
 
119
            t = transport.get_transport('foo://fooserver/foo')
 
120
            self.assertTrue(isinstance(t, BackupTransportHandler))
 
121
        finally:
 
122
            transport._set_protocol_handlers(saved_handlers)
112
123
 
113
124
    def test_ssh_hints(self):
114
125
        """Transport ssh:// should raise an error pointing out bzr+ssh://"""
207
218
 
208
219
    def test_coalesce_fudge(self):
209
220
        self.check([(10, 30, [(0, 10), (20, 10)]),
210
 
                    (100, 10, [(0, 10)]),
 
221
                    (100, 10, [(0, 10),]),
211
222
                   ], [(10, 10), (30, 10), (100, 10)],
212
 
                   fudge=10)
213
 
 
 
223
                   fudge=10
 
224
                  )
214
225
    def test_coalesce_max_size(self):
215
226
        self.check([(10, 20, [(0, 10), (10, 10)]),
216
227
                    (30, 50, [(0, 50)]),
217
228
                    # If one range is above max_size, it gets its own coalesced
218
229
                    # offset
219
 
                    (100, 80, [(0, 80)]),],
 
230
                    (100, 80, [(0, 80),]),],
220
231
                   [(10, 10), (20, 10), (30, 50), (100, 80)],
221
 
                   max_size=50)
 
232
                   max_size=50
 
233
                  )
222
234
 
223
235
    def test_coalesce_no_max_size(self):
224
 
        self.check([(10, 170, [(0, 10), (10, 10), (20, 50), (70, 100)])],
 
236
        self.check([(10, 170, [(0, 10), (10, 10), (20, 50), (70, 100)]),],
225
237
                   [(10, 10), (20, 10), (30, 50), (80, 100)],
226
238
                  )
227
239
 
228
240
    def test_coalesce_default_limit(self):
229
241
        # By default we use a 100MB max size.
230
 
        ten_mb = 10 * 1024 * 1024
231
 
        self.check([(0, 10 * ten_mb, [(i * ten_mb, ten_mb) for i in range(10)]),
 
242
        ten_mb = 10*1024*1024
 
243
        self.check([(0, 10*ten_mb, [(i*ten_mb, ten_mb) for i in range(10)]),
232
244
                    (10*ten_mb, ten_mb, [(0, ten_mb)])],
233
245
                   [(i*ten_mb, ten_mb) for i in range(11)])
234
 
        self.check([(0, 11 * ten_mb, [(i * ten_mb, ten_mb) for i in range(11)])],
235
 
                   [(i * ten_mb, ten_mb) for i in range(11)],
 
246
        self.check([(0, 11*ten_mb, [(i*ten_mb, ten_mb) for i in range(11)]),],
 
247
                   [(i*ten_mb, ten_mb) for i in range(11)],
236
248
                   max_size=1*1024*1024*1024)
237
249
 
238
250
 
410
422
            parent_url = urlutils.join(url, '..')
411
423
            new_t = transport.get_transport(parent_url)
412
424
        """
413
 
        server = chroot.ChrootServer(
414
 
            transport.get_transport('memory:///path/'))
 
425
        server = chroot.ChrootServer(transport.get_transport('memory:///path/'))
415
426
        self.start_server(server)
416
427
        t = transport.get_transport(server.get_url())
417
428
        self.assertRaises(
429
440
        backing_transport = memory.MemoryTransport()
430
441
        server = chroot.ChrootServer(backing_transport)
431
442
        server.start_server()
432
 
        self.addCleanup(server.stop_server)
433
 
        self.assertTrue(server.scheme
434
 
                        in transport._get_protocol_handlers().keys())
 
443
        try:
 
444
            self.assertTrue(server.scheme
 
445
                            in transport._get_protocol_handlers().keys())
 
446
        finally:
 
447
            server.stop_server()
435
448
 
436
449
    def test_stop_server(self):
437
450
        backing_transport = memory.MemoryTransport()
445
458
        backing_transport = memory.MemoryTransport()
446
459
        server = chroot.ChrootServer(backing_transport)
447
460
        server.start_server()
448
 
        self.addCleanup(server.stop_server)
449
 
        self.assertEqual('chroot-%d:///' % id(server), server.get_url())
 
461
        try:
 
462
            self.assertEqual('chroot-%d:///' % id(server), server.get_url())
 
463
        finally:
 
464
            server.stop_server()
450
465
 
451
466
 
452
467
class PathFilteringDecoratorTransportTest(tests.TestCase):
467
482
 
468
483
    def make_pf_transport(self, filter_func=None):
469
484
        """Make a PathFilteringTransport backed by a MemoryTransport.
470
 
 
 
485
        
471
486
        :param filter_func: by default this will be a no-op function.  Use this
472
487
            parameter to override it."""
473
488
        if filter_func is None:
495
510
 
496
511
    def test_filter_invocation(self):
497
512
        filter_log = []
498
 
 
499
513
        def filter(path):
500
514
            filter_log.append(path)
501
515
            return path
549
563
        server = HttpServer()
550
564
        self.start_server(server)
551
565
        t = transport.get_transport('readonly+' + server.get_url())
552
 
        self.assertIsInstance(t, readonly.ReadonlyTransportDecorator)
 
566
        self.failUnless(isinstance(t, readonly.ReadonlyTransportDecorator))
553
567
        self.assertEqual(False, t.listable())
554
568
        self.assertEqual(True, t.is_readonly())
555
569
 
732
746
        self.assertEquals(t._host, 'simple.example.com')
733
747
        self.assertEquals(t._port, None)
734
748
        self.assertEquals(t._path, '/home/source/')
735
 
        self.assertTrue(t._user is None)
736
 
        self.assertTrue(t._password is None)
 
749
        self.failUnless(t._user is None)
 
750
        self.failUnless(t._password is None)
737
751
 
738
752
        self.assertEquals(t.base, 'http://simple.example.com/home/source/')
739
753
 
762
776
    def test_relpath(self):
763
777
        t = transport.ConnectedTransport('sftp://user@host.com/abs/path')
764
778
 
765
 
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'),
766
 
            'sub')
 
779
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
767
780
        self.assertRaises(errors.PathNotChild, t.relpath,
768
781
                          'http://user@host.com/abs/path/sub')
769
782
        self.assertRaises(errors.PathNotChild, t.relpath,
842
855
 
843
856
    def test_get(self):
844
857
        t = transport.get_transport('trace+memory://')
845
 
        self.assertIsInstance(
846
 
            t, bzrlib.transport.trace.TransportTraceDecorator)
 
858
        self.assertIsInstance(t, bzrlib.transport.trace.TransportTraceDecorator)
847
859
 
848
860
    def test_clone_preserves_activity(self):
849
861
        t = transport.get_transport('trace+memory://')
884
896
class TestSSHConnections(tests.TestCaseWithTransport):
885
897
 
886
898
    def test_bzr_connect_to_bzr_ssh(self):
887
 
        """get_transport of a bzr+ssh:// behaves correctly.
 
899
        """User acceptance that get_transport of a bzr+ssh:// behaves correctly.
888
900
 
889
901
        bzr+ssh:// should cause bzr to run a remote bzr smart server over SSH.
890
902
        """
906
918
        # SSH channel ourselves.  Surely this has already been implemented
907
919
        # elsewhere?
908
920
        started = []
909
 
 
910
921
        class StubSSHServer(stub_sftp.StubServer):
911
922
 
912
923
            test = self
918
929
                    stdout=subprocess.PIPE, stderr=subprocess.PIPE)
919
930
 
920
931
                # XXX: horribly inefficient, not to mention ugly.
921
 
                # Start a thread for each of stdin/out/err, and relay bytes
922
 
                # from the subprocess to channel and vice versa.
 
932
                # Start a thread for each of stdin/out/err, and relay bytes from
 
933
                # the subprocess to channel and vice versa.
923
934
                def ferry_bytes(read, write, close):
924
935
                    while True:
925
936
                        bytes = read(1)
955
966
            bzr_remote_path = sys.executable + ' ' + self.get_bzr_path()
956
967
        else:
957
968
            bzr_remote_path = self.get_bzr_path()
958
 
        self.overrideEnv('BZR_REMOTE_PATH', bzr_remote_path)
 
969
        os.environ['BZR_REMOTE_PATH'] = bzr_remote_path
959
970
 
960
971
        # Access the branch via a bzr+ssh URL.  The BZR_REMOTE_PATH environment
961
972
        # variable is used to tell bzr what command to run on the remote end.
982
993
        # And the rest are threads
983
994
        for t in started[1:]:
984
995
            t.join()
985
 
 
986
 
 
987
 
class TestUnhtml(tests.TestCase):
988
 
 
989
 
    """Tests for unhtml_roughly"""
990
 
 
991
 
    def test_truncation(self):
992
 
        fake_html = "<p>something!\n" * 1000
993
 
        result = http.unhtml_roughly(fake_html)
994
 
        self.assertEquals(len(result), 1000)
995
 
        self.assertStartsWith(result, " something!")