← 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: Robert Collins
  • Date: 2007-10-02 05:33:39 UTC
  • mto: This revision was merged to the branch mainline in revision 2885.
  • Revision ID: robertc@robertcollins.net-20071002053339-vnyjf4jrxv0jeekf
* Move transport logging into a new transport class
  TransportTraceDecorator (trace+ to get it from a url).
* Give Registry a useful __repr__.
* Fix a bug introduced by the change to use a Registry for transport where
  the transport loading tests left global state behind due to the
  _get_protocol_handlers method returning a reference, not a value object.
* Add an upper_limit parameter to readv, because when asking for byte
  ranges within the latency-adjustment window near the end of the file
  causes errors that are tricky to manage.
* A few minor drive-by formatting fixes.
* The TransportDecorator constructor now accepts a _from_transport
  parameter for decorators that want to share state (used by the trace
  decorator)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
38
38
                           UnsupportedProtocol,
39
39
                           )
40
40
from bzrlib.tests import TestCase, TestCaseInTempDir
41
 
from bzrlib.transport import (_CoalescedOffset,
 
41
from bzrlib.transport import (_clear_protocol_handlers,
 
42
                              _CoalescedOffset,
42
43
                              ConnectedTransport,
43
44
                              _get_protocol_handlers,
44
45
                              _set_protocol_handlers,
73
74
 
74
75
    def test_get_transport_modules(self):
75
76
        handlers = _get_protocol_handlers()
 
77
        # don't pollute the current handlers
 
78
        _clear_protocol_handlers()
76
79
        class SampleHandler(object):
77
80
            """I exist, isnt that enough?"""
78
81
        try:
89
92
    def test_transport_dependency(self):
90
93
        """Transport with missing dependency causes no error"""
91
94
        saved_handlers = _get_protocol_handlers()
 
95
        # don't pollute the current handlers
 
96
        _clear_protocol_handlers()
92
97
        try:
93
98
            register_transport_proto('foo')
94
99
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
712
717
        self.assertIsNot(t1, t2)
713
718
 
714
719
 
715
 
def get_test_permutations():
716
 
    """Return transport permutations to be used in testing.
717
 
 
718
 
    This module registers some transports, but they're only for testing
719
 
    registration.  We don't really want to run all the transport tests against
720
 
    them.
721
 
    """
722
 
    return []
 
720
class TestTransportTrace(TestCase):
 
721
 
 
722
    def test_get(self):
 
723
        transport = get_transport('trace+memory://')
 
724
        self.assertIsInstance(
 
725
            transport, bzrlib.transport.trace.TransportTraceDecorator)
 
726
 
 
727
    def test_clone_preserves_activity(self):
 
728
        transport = get_transport('trace+memory://')
 
729
        transport2 = transport.clone('.')
 
730
        self.assertTrue(transport is not transport2)
 
731
        self.assertTrue(transport._activity is transport2._activity)
 
732
 
 
733
    # the following specific tests are for the operations that have made use of
 
734
    # logging in tests; we could test every single operation but doing that
 
735
    # still won't cause a test failure when the top level Transport API
 
736
    # changes; so there is little return doing that.
 
737
    def test_get(self):
 
738
        transport = get_transport('trace+memory:///')
 
739
        transport.put_bytes('foo', 'barish')
 
740
        transport.get('foo')
 
741
        expected_result = []
 
742
        # put_bytes records the bytes, not the content to avoid memory
 
743
        # pressure.
 
744
        expected_result.append(('put_bytes', 'foo', 6, None))
 
745
        # get records the file name only.
 
746
        expected_result.append(('get', 'foo'))
 
747
        self.assertEqual(expected_result, transport._activity)
 
748
 
 
749
    def test_readv(self):
 
750
        transport = get_transport('trace+memory:///')
 
751
        transport.put_bytes('foo', 'barish')
 
752
        list(transport.readv('foo', [(0, 1), (3, 2)], adjust_for_latency=True,
 
753
            upper_limit=6))
 
754
        expected_result = []
 
755
        # put_bytes records the bytes, not the content to avoid memory
 
756
        # pressure.
 
757
        expected_result.append(('put_bytes', 'foo', 6, None))
 
758
        # readv records the supplied offset request
 
759
        expected_result.append(('readv', 'foo', [(0, 1), (3, 2)], True, 6))
 
760
        self.assertEqual(expected_result, transport._activity)