/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: Martin Pool
  • Date: 2007-10-08 07:29:57 UTC
  • mfrom: (2894 +trunk)
  • mto: This revision was merged to the branch mainline in revision 2895.
  • Revision ID: mbp@sourcefrog.net-20071008072957-uhm1gl1mqcsdc377
merge trunk

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,
55
56
from bzrlib.transport.memory import MemoryTransport
56
57
from bzrlib.transport.local import (LocalTransport,
57
58
                                    EmulatedWin32LocalTransport)
 
59
from bzrlib.transport.remote import (
 
60
    BZR_DEFAULT_PORT,
 
61
    RemoteTCPTransport
 
62
    )
58
63
 
59
64
 
60
65
# TODO: Should possibly split transport-specific tests into their own files.
74
79
 
75
80
    def test_get_transport_modules(self):
76
81
        handlers = _get_protocol_handlers()
 
82
        # don't pollute the current handlers
 
83
        _clear_protocol_handlers()
77
84
        class SampleHandler(object):
78
85
            """I exist, isnt that enough?"""
79
86
        try:
90
97
    def test_transport_dependency(self):
91
98
        """Transport with missing dependency causes no error"""
92
99
        saved_handlers = _get_protocol_handlers()
 
100
        # don't pollute the current handlers
 
101
        _clear_protocol_handlers()
93
102
        try:
94
103
            register_transport_proto('foo')
95
104
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
618
627
    def test_parse_url(self):
619
628
        t = ConnectedTransport('sftp://simple.example.com/home/source')
620
629
        self.assertEquals(t._host, 'simple.example.com')
621
 
        self.assertEquals(t._port, None)
 
630
        self.assertEquals(t._port, 22)
622
631
        self.assertEquals(t._path, '/home/source/')
623
632
        self.failUnless(t._user is None)
624
633
        self.failUnless(t._password is None)
713
722
        self.assertIsNot(t1, t2)
714
723
 
715
724
 
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 []
 
725
class TestRemoteTCPTransport(TestCase):
 
726
    """Tests for bzr:// transport (RemoteTCPTransport)."""
 
727
 
 
728
    def test_relpath_with_implicit_port(self):
 
729
        """Connected transports with the same URL are the same, even if the
 
730
        port is implicit.
 
731
 
 
732
        So t.relpath(url) should always be '' if t.base is the same as url, or
 
733
        if the only difference is that one explicitly specifies the default
 
734
        port and the other doesn't specify a port.
 
735
        """
 
736
        t_implicit_port = RemoteTCPTransport('bzr://host.com/')
 
737
        self.assertEquals('', t_implicit_port.relpath('bzr://host.com/'))
 
738
        self.assertEquals('', t_implicit_port.relpath('bzr://host.com:4155/'))
 
739
        t_explicit_port = RemoteTCPTransport('bzr://host.com:4155/')
 
740
        self.assertEquals('', t_explicit_port.relpath('bzr://host.com/'))
 
741
        self.assertEquals('', t_explicit_port.relpath('bzr://host.com:4155/'))
 
742
 
 
743
    def test_construct_uses_default_port(self):
 
744
        """If no port is specified, then RemoteTCPTransport uses
 
745
        BZR_DEFAULT_PORT.
 
746
        """
 
747
        t = get_transport('bzr://host.com/')
 
748
        self.assertEquals(BZR_DEFAULT_PORT, t._port)
 
749
 
 
750
    def test_url_omits_default_port(self):
 
751
        """If a RemoteTCPTransport uses the default port, then its base URL
 
752
        will omit the port.
 
753
 
 
754
        This is like how ":80" is omitted from "http://example.com/".
 
755
        """
 
756
        t = get_transport('bzr://host.com:4155/')
 
757
        self.assertEquals('bzr://host.com/', t.base)
 
758
 
 
759
    def test_url_includes_non_default_port(self):
 
760
        """Non-default ports are included in the transport's URL.
 
761
 
 
762
        Contrast this to `test_url_omits_default_port`.
 
763
        """
 
764
        t = get_transport('bzr://host.com:666/')
 
765
        self.assertEquals('bzr://host.com:666/', t.base)
 
766
 
 
767
 
 
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)