← 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: Canonical.com Patch Queue Manager
  • Date: 2007-06-28 07:08:27 UTC
  • mfrom: (2553.1.3 integration)
  • Revision ID: pqm@pqm.ubuntu.com-20070628070827-h5s313dg5tnag9vj
(robertc) Show the names of commit hooks during commit.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
32
32
                           NoSuchFile,
33
33
                           PathNotChild,
34
34
                           TransportNotPossible,
35
 
                           ConnectionError,
36
 
                           DependencyNotPresent,
37
 
                           ReadError,
38
35
                           UnsupportedProtocol,
39
36
                           )
40
37
from bzrlib.tests import TestCase, TestCaseInTempDir
41
38
from bzrlib.transport import (_CoalescedOffset,
42
 
                              ConnectedTransport,
43
39
                              _get_protocol_handlers,
44
40
                              _set_protocol_handlers,
45
41
                              _get_transport_modules,
46
42
                              get_transport,
47
 
                              LateReadError,
48
43
                              register_lazy_transport,
49
44
                              register_transport_proto,
50
45
                              _clear_protocol_handlers,
122
117
        finally:
123
118
            _set_protocol_handlers(saved_handlers)
124
119
 
125
 
    def test_LateReadError(self):
126
 
        """The LateReadError helper should raise on read()."""
127
 
        a_file = LateReadError('a path')
128
 
        try:
129
 
            a_file.read()
130
 
        except ReadError, error:
131
 
            self.assertEqual('a path', error.path)
132
 
        self.assertRaises(ReadError, a_file.read, 40)
133
 
        a_file.close()
134
 
 
135
120
    def test__combine_paths(self):
136
121
        t = Transport('/')
137
122
        self.assertEqual('/home/sarah/project/foo',
555
540
        self._server.setUp()
556
541
        self.addCleanup(self._server.tearDown)
557
542
 
558
 
    def get_transport(self, relpath=None):
559
 
        """Return a connected transport to the local directory.
560
 
 
561
 
        :param relpath: a path relative to the base url.
562
 
        """
 
543
    def get_transport(self):
 
544
        """Return a connected transport to the local directory."""
563
545
        base_url = self._server.get_url()
564
 
        url = self._adjust_url(base_url, relpath)
565
546
        # try getting the transport via the regular interface:
566
 
        t = get_transport(url)
567
 
        # vila--20070607 if the following are commented out the test suite
568
 
        # still pass. Is this really still needed or was it a forgotten
569
 
        # temporary fix ?
 
547
        t = get_transport(base_url)
570
548
        if not isinstance(t, self.transport_class):
571
549
            # we did not get the correct transport class type. Override the
572
550
            # regular connection behaviour by direct construction.
573
 
            t = self.transport_class(url)
 
551
            t = self.transport_class(base_url)
574
552
        return t
575
553
 
576
554
 
616
594
        self.assertEquals(t.base, 'file://HOST/')
617
595
 
618
596
 
619
 
class TestConnectedTransport(TestCase):
620
 
    """Tests for connected to remote server transports"""
621
 
 
622
 
    def test_parse_url(self):
623
 
        t = ConnectedTransport('sftp://simple.example.com/home/source')
624
 
        self.assertEquals(t._host, 'simple.example.com')
625
 
        self.assertEquals(t._port, None)
626
 
        self.assertEquals(t._path, '/home/source/')
627
 
        self.failUnless(t._user is None)
628
 
        self.failUnless(t._password is None)
629
 
 
630
 
        self.assertEquals(t.base, 'sftp://simple.example.com/home/source/')
631
 
 
632
 
    def test_parse_quoted_url(self):
633
 
        t = ConnectedTransport('http://ro%62ey:h%40t@ex%41mple.com:2222/path')
634
 
        self.assertEquals(t._host, 'exAmple.com')
635
 
        self.assertEquals(t._port, 2222)
636
 
        self.assertEquals(t._user, 'robey')
637
 
        self.assertEquals(t._password, 'h@t')
638
 
        self.assertEquals(t._path, '/path/')
639
 
 
640
 
        # Base should not keep track of the password
641
 
        self.assertEquals(t.base, 'http://robey@exAmple.com:2222/path/')
642
 
 
643
 
    def test_parse_invalid_url(self):
644
 
        self.assertRaises(errors.InvalidURL,
645
 
                          ConnectedTransport,
646
 
                          'sftp://lily.org:~janneke/public/bzr/gub')
647
 
 
648
 
    def test_relpath(self):
649
 
        t = ConnectedTransport('sftp://user@host.com/abs/path')
650
 
 
651
 
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
652
 
        self.assertRaises(errors.PathNotChild, t.relpath,
653
 
                          'http://user@host.com/abs/path/sub')
654
 
        self.assertRaises(errors.PathNotChild, t.relpath,
655
 
                          'sftp://user2@host.com/abs/path/sub')
656
 
        self.assertRaises(errors.PathNotChild, t.relpath,
657
 
                          'sftp://user@otherhost.com/abs/path/sub')
658
 
        self.assertRaises(errors.PathNotChild, t.relpath,
659
 
                          'sftp://user@host.com:33/abs/path/sub')
660
 
        # Make sure it works when we don't supply a username
661
 
        t = ConnectedTransport('sftp://host.com/abs/path')
662
 
        self.assertEquals(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
663
 
 
664
 
        # Make sure it works when parts of the path will be url encoded
665
 
        t = ConnectedTransport('sftp://host.com/dev/%path')
666
 
        self.assertEquals(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
667
 
 
668
 
    def test_connection_sharing_propagate_credentials(self):
669
 
        t = ConnectedTransport('foo://user@host.com/abs/path')
670
 
        self.assertIs(None, t._get_connection())
671
 
        self.assertIs(None, t._password)
672
 
        c = t.clone('subdir')
673
 
        self.assertEquals(None, c._get_connection())
674
 
        self.assertIs(None, t._password)
675
 
 
676
 
        # Simulate the user entering a password
677
 
        password = 'secret'
678
 
        connection = object()
679
 
        t._set_connection(connection, password)
680
 
        self.assertIs(connection, t._get_connection())
681
 
        self.assertIs(password, t._get_credentials())
682
 
        self.assertIs(connection, c._get_connection())
683
 
        self.assertIs(password, c._get_credentials())
684
 
 
685
 
        # credentials can be updated
686
 
        new_password = 'even more secret'
687
 
        c._update_credentials(new_password)
688
 
        self.assertIs(connection, t._get_connection())
689
 
        self.assertIs(new_password, t._get_credentials())
690
 
        self.assertIs(connection, c._get_connection())
691
 
        self.assertIs(new_password, c._get_credentials())
692
 
 
693
 
 
694
 
class TestReusedTransports(TestCase):
695
 
    """Tests for transport reuse"""
696
 
 
697
 
    def test_reuse_same_transport(self):
698
 
        t1 = get_transport('http://foo/')
699
 
        t2 = get_transport('http://foo/', possible_transports=[t1])
700
 
        self.assertIs(t1, t2)
701
 
 
702
 
        # Also check that final '/' are handled correctly
703
 
        t3 = get_transport('http://foo/path/')
704
 
        t4 = get_transport('http://foo/path', possible_transports=[t3])
705
 
        self.assertIs(t3, t4)
706
 
 
707
 
        t5 = get_transport('http://foo/path')
708
 
        t6 = get_transport('http://foo/path/', possible_transports=[t5])
709
 
        self.assertIs(t5, t6)
710
 
 
711
 
    def test_don_t_reuse_different_transport(self):
712
 
        t1 = get_transport('http://foo/path')
713
 
        t2 = get_transport('http://bar/path', possible_transports=[t1])
714
 
        self.assertIsNot(t1, t2)
715
 
 
716
 
 
717
597
def get_test_permutations():
718
598
    """Return transport permutations to be used in testing.
719
599