/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

Merge from bzr.dev.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
23
23
import bzrlib
24
24
from bzrlib import (
25
25
    errors,
 
26
    osutils,
26
27
    urlutils,
27
28
    )
28
29
from bzrlib.errors import (ConnectionError,
38
39
                           UnsupportedProtocol,
39
40
                           )
40
41
from bzrlib.tests import TestCase, TestCaseInTempDir
41
 
from bzrlib.transport import (_CoalescedOffset,
 
42
from bzrlib.transport import (_clear_protocol_handlers,
 
43
                              _CoalescedOffset,
42
44
                              ConnectedTransport,
43
45
                              _get_protocol_handlers,
44
46
                              _set_protocol_handlers,
54
56
from bzrlib.transport.memory import MemoryTransport
55
57
from bzrlib.transport.local import (LocalTransport,
56
58
                                    EmulatedWin32LocalTransport)
 
59
from bzrlib.transport.remote import (
 
60
    BZR_DEFAULT_PORT,
 
61
    RemoteTCPTransport
 
62
    )
57
63
 
58
64
 
59
65
# TODO: Should possibly split transport-specific tests into their own files.
73
79
 
74
80
    def test_get_transport_modules(self):
75
81
        handlers = _get_protocol_handlers()
 
82
        # don't pollute the current handlers
 
83
        _clear_protocol_handlers()
76
84
        class SampleHandler(object):
77
85
            """I exist, isnt that enough?"""
78
86
        try:
89
97
    def test_transport_dependency(self):
90
98
        """Transport with missing dependency causes no error"""
91
99
        saved_handlers = _get_protocol_handlers()
 
100
        # don't pollute the current handlers
 
101
        _clear_protocol_handlers()
92
102
        try:
93
103
            register_transport_proto('foo')
94
104
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
572
582
class TestLocalTransports(TestCase):
573
583
 
574
584
    def test_get_transport_from_abspath(self):
575
 
        here = os.path.abspath('.')
 
585
        here = osutils.abspath('.')
576
586
        t = get_transport(here)
577
587
        self.assertIsInstance(t, LocalTransport)
578
588
        self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
579
589
 
580
590
    def test_get_transport_from_relpath(self):
581
 
        here = os.path.abspath('.')
 
591
        here = osutils.abspath('.')
582
592
        t = get_transport('.')
583
593
        self.assertIsInstance(t, LocalTransport)
584
594
        self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
585
595
 
586
596
    def test_get_transport_from_local_url(self):
587
 
        here = os.path.abspath('.')
 
597
        here = osutils.abspath('.')
588
598
        here_url = urlutils.local_path_to_url(here) + '/'
589
599
        t = get_transport(here_url)
590
600
        self.assertIsInstance(t, LocalTransport)
591
601
        self.assertEquals(t.base, here_url)
592
602
 
593
603
    def test_local_abspath(self):
594
 
        here = os.path.abspath('.')
 
604
        here = osutils.abspath('.')
595
605
        t = get_transport(here)
596
606
        self.assertEquals(t.local_abspath(''), here)
597
607
 
617
627
    def test_parse_url(self):
618
628
        t = ConnectedTransport('sftp://simple.example.com/home/source')
619
629
        self.assertEquals(t._host, 'simple.example.com')
620
 
        self.assertEquals(t._port, None)
 
630
        self.assertEquals(t._port, 22)
621
631
        self.assertEquals(t._path, '/home/source/')
622
632
        self.failUnless(t._user is None)
623
633
        self.failUnless(t._password is None)
712
722
        self.assertIsNot(t1, t2)
713
723
 
714
724
 
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 []
 
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)