/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2007-10-04 07:12:14 UTC
  • mfrom: (2745.5.6 transport-get-file)
  • Revision ID: pqm@pqm.ubuntu.com-20071004071214-i0icltanhq59qtwt
(robertc) Transport improvements from packs: new trace+ transport, and readv adjust_for_latency feature, as well as test harness cleanup. (Robert Collins)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
39
39
                           UnsupportedProtocol,
40
40
                           )
41
41
from bzrlib.tests import TestCase, TestCaseInTempDir
42
 
from bzrlib.transport import (_CoalescedOffset,
 
42
from bzrlib.transport import (_clear_protocol_handlers,
 
43
                              _CoalescedOffset,
43
44
                              ConnectedTransport,
44
45
                              _get_protocol_handlers,
45
46
                              _set_protocol_handlers,
78
79
 
79
80
    def test_get_transport_modules(self):
80
81
        handlers = _get_protocol_handlers()
 
82
        # don't pollute the current handlers
 
83
        _clear_protocol_handlers()
81
84
        class SampleHandler(object):
82
85
            """I exist, isnt that enough?"""
83
86
        try:
94
97
    def test_transport_dependency(self):
95
98
        """Transport with missing dependency causes no error"""
96
99
        saved_handlers = _get_protocol_handlers()
 
100
        # don't pollute the current handlers
 
101
        _clear_protocol_handlers()
97
102
        try:
98
103
            register_transport_proto('foo')
99
104
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
760
765
        self.assertEquals('bzr://host.com:666/', t.base)
761
766
 
762
767
 
763
 
 
764
 
def get_test_permutations():
765
 
    """Return transport permutations to be used in testing.
766
 
 
767
 
    This module registers some transports, but they're only for testing
768
 
    registration.  We don't really want to run all the transport tests against
769
 
    them.
770
 
    """
771
 
    return []
 
768
class TestTransportTrace(TestCase):
 
769
 
 
770
    def test_get(self):
 
771
        transport = get_transport('trace+memory://')
 
772
        self.assertIsInstance(
 
773
            transport, bzrlib.transport.trace.TransportTraceDecorator)
 
774
 
 
775
    def test_clone_preserves_activity(self):
 
776
        transport = get_transport('trace+memory://')
 
777
        transport2 = transport.clone('.')
 
778
        self.assertTrue(transport is not transport2)
 
779
        self.assertTrue(transport._activity is transport2._activity)
 
780
 
 
781
    # the following specific tests are for the operations that have made use of
 
782
    # logging in tests; we could test every single operation but doing that
 
783
    # still won't cause a test failure when the top level Transport API
 
784
    # changes; so there is little return doing that.
 
785
    def test_get(self):
 
786
        transport = get_transport('trace+memory:///')
 
787
        transport.put_bytes('foo', 'barish')
 
788
        transport.get('foo')
 
789
        expected_result = []
 
790
        # put_bytes records the bytes, not the content to avoid memory
 
791
        # pressure.
 
792
        expected_result.append(('put_bytes', 'foo', 6, None))
 
793
        # get records the file name only.
 
794
        expected_result.append(('get', 'foo'))
 
795
        self.assertEqual(expected_result, transport._activity)
 
796
 
 
797
    def test_readv(self):
 
798
        transport = get_transport('trace+memory:///')
 
799
        transport.put_bytes('foo', 'barish')
 
800
        list(transport.readv('foo', [(0, 1), (3, 2)], adjust_for_latency=True,
 
801
            upper_limit=6))
 
802
        expected_result = []
 
803
        # put_bytes records the bytes, not the content to avoid memory
 
804
        # pressure.
 
805
        expected_result.append(('put_bytes', 'foo', 6, None))
 
806
        # readv records the supplied offset request
 
807
        expected_result.append(('readv', 'foo', [(0, 1), (3, 2)], True, 6))
 
808
        self.assertEqual(expected_result, transport._activity)