/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-09-14 06:31:28 UTC
  • mfrom: (2822 +trunk)
  • mto: This revision was merged to the branch mainline in revision 2823.
  • Revision ID: mbp@sourcefrog.net-20070914063128-0p7mh6zfb4pzdg9p
merge trunk

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
39
39
                           )
40
40
from bzrlib.tests import TestCase, TestCaseInTempDir
41
41
from bzrlib.transport import (_CoalescedOffset,
 
42
                              ConnectedTransport,
42
43
                              _get_protocol_handlers,
43
44
                              _set_protocol_handlers,
44
45
                              _get_transport_modules,
297
298
    def test_parameters(self):
298
299
        transport = MemoryTransport()
299
300
        self.assertEqual(True, transport.listable())
300
 
        self.assertEqual(False, transport.should_cache())
301
301
        self.assertEqual(False, transport.is_readonly())
302
302
 
303
303
    def test_iter_files_recursive(self):
413
413
        # connect to . in readonly mode
414
414
        transport = readonly.ReadonlyTransportDecorator('readonly+.')
415
415
        self.assertEqual(True, transport.listable())
416
 
        self.assertEqual(False, transport.should_cache())
417
416
        self.assertEqual(True, transport.is_readonly())
418
417
 
419
418
    def test_http_parameters(self):
427
426
            self.failUnless(isinstance(transport,
428
427
                                       readonly.ReadonlyTransportDecorator))
429
428
            self.assertEqual(False, transport.listable())
430
 
            self.assertEqual(True, transport.should_cache())
431
429
            self.assertEqual(True, transport.is_readonly())
432
430
        finally:
433
431
            server.tearDown()
442
440
        return fakenfs.FakeNFSTransportDecorator('fakenfs+' + url)
443
441
 
444
442
    def test_local_parameters(self):
445
 
        # the listable, should_cache and is_readonly parameters
 
443
        # the listable and is_readonly parameters
446
444
        # are not changed by the fakenfs decorator
447
445
        transport = self.get_nfs_transport('.')
448
446
        self.assertEqual(True, transport.listable())
449
 
        self.assertEqual(False, transport.should_cache())
450
447
        self.assertEqual(False, transport.is_readonly())
451
448
 
452
449
    def test_http_parameters(self):
453
 
        # the listable, should_cache and is_readonly parameters
 
450
        # the listable and is_readonly parameters
454
451
        # are not changed by the fakenfs decorator
455
452
        from bzrlib.tests.HttpServer import HttpServer
456
453
        # connect to . via http which is not listable
461
458
            self.assertIsInstance(
462
459
                transport, bzrlib.transport.fakenfs.FakeNFSTransportDecorator)
463
460
            self.assertEqual(False, transport.listable())
464
 
            self.assertEqual(True, transport.should_cache())
465
461
            self.assertEqual(True, transport.is_readonly())
466
462
        finally:
467
463
            server.tearDown()
563
559
        url = self._adjust_url(base_url, relpath)
564
560
        # try getting the transport via the regular interface:
565
561
        t = get_transport(url)
 
562
        # vila--20070607 if the following are commented out the test suite
 
563
        # still pass. Is this really still needed or was it a forgotten
 
564
        # temporary fix ?
566
565
        if not isinstance(t, self.transport_class):
567
566
            # we did not get the correct transport class type. Override the
568
567
            # regular connection behaviour by direct construction.
612
611
        self.assertEquals(t.base, 'file://HOST/')
613
612
 
614
613
 
 
614
class TestConnectedTransport(TestCase):
 
615
    """Tests for connected to remote server transports"""
 
616
 
 
617
    def test_parse_url(self):
 
618
        t = ConnectedTransport('sftp://simple.example.com/home/source')
 
619
        self.assertEquals(t._host, 'simple.example.com')
 
620
        self.assertEquals(t._port, None)
 
621
        self.assertEquals(t._path, '/home/source/')
 
622
        self.failUnless(t._user is None)
 
623
        self.failUnless(t._password is None)
 
624
 
 
625
        self.assertEquals(t.base, 'sftp://simple.example.com/home/source/')
 
626
 
 
627
    def test_parse_quoted_url(self):
 
628
        t = ConnectedTransport('http://ro%62ey:h%40t@ex%41mple.com:2222/path')
 
629
        self.assertEquals(t._host, 'exAmple.com')
 
630
        self.assertEquals(t._port, 2222)
 
631
        self.assertEquals(t._user, 'robey')
 
632
        self.assertEquals(t._password, 'h@t')
 
633
        self.assertEquals(t._path, '/path/')
 
634
 
 
635
        # Base should not keep track of the password
 
636
        self.assertEquals(t.base, 'http://robey@exAmple.com:2222/path/')
 
637
 
 
638
    def test_parse_invalid_url(self):
 
639
        self.assertRaises(errors.InvalidURL,
 
640
                          ConnectedTransport,
 
641
                          'sftp://lily.org:~janneke/public/bzr/gub')
 
642
 
 
643
    def test_relpath(self):
 
644
        t = ConnectedTransport('sftp://user@host.com/abs/path')
 
645
 
 
646
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
 
647
        self.assertRaises(errors.PathNotChild, t.relpath,
 
648
                          'http://user@host.com/abs/path/sub')
 
649
        self.assertRaises(errors.PathNotChild, t.relpath,
 
650
                          'sftp://user2@host.com/abs/path/sub')
 
651
        self.assertRaises(errors.PathNotChild, t.relpath,
 
652
                          'sftp://user@otherhost.com/abs/path/sub')
 
653
        self.assertRaises(errors.PathNotChild, t.relpath,
 
654
                          'sftp://user@host.com:33/abs/path/sub')
 
655
        # Make sure it works when we don't supply a username
 
656
        t = ConnectedTransport('sftp://host.com/abs/path')
 
657
        self.assertEquals(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
 
658
 
 
659
        # Make sure it works when parts of the path will be url encoded
 
660
        t = ConnectedTransport('sftp://host.com/dev/%path')
 
661
        self.assertEquals(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
 
662
 
 
663
    def test_connection_sharing_propagate_credentials(self):
 
664
        t = ConnectedTransport('foo://user@host.com/abs/path')
 
665
        self.assertIs(None, t._get_connection())
 
666
        self.assertIs(None, t._password)
 
667
        c = t.clone('subdir')
 
668
        self.assertEquals(None, c._get_connection())
 
669
        self.assertIs(None, t._password)
 
670
 
 
671
        # Simulate the user entering a password
 
672
        password = 'secret'
 
673
        connection = object()
 
674
        t._set_connection(connection, password)
 
675
        self.assertIs(connection, t._get_connection())
 
676
        self.assertIs(password, t._get_credentials())
 
677
        self.assertIs(connection, c._get_connection())
 
678
        self.assertIs(password, c._get_credentials())
 
679
 
 
680
        # credentials can be updated
 
681
        new_password = 'even more secret'
 
682
        c._update_credentials(new_password)
 
683
        self.assertIs(connection, t._get_connection())
 
684
        self.assertIs(new_password, t._get_credentials())
 
685
        self.assertIs(connection, c._get_connection())
 
686
        self.assertIs(new_password, c._get_credentials())
 
687
 
 
688
 
 
689
class TestReusedTransports(TestCase):
 
690
    """Tests for transport reuse"""
 
691
 
 
692
    def test_reuse_same_transport(self):
 
693
        possible_transports = []
 
694
        t1 = get_transport('http://foo/',
 
695
                           possible_transports=possible_transports)
 
696
        self.assertEqual([t1], possible_transports)
 
697
        t2 = get_transport('http://foo/', possible_transports=[t1])
 
698
        self.assertIs(t1, t2)
 
699
 
 
700
        # Also check that final '/' are handled correctly
 
701
        t3 = get_transport('http://foo/path/')
 
702
        t4 = get_transport('http://foo/path', possible_transports=[t3])
 
703
        self.assertIs(t3, t4)
 
704
 
 
705
        t5 = get_transport('http://foo/path')
 
706
        t6 = get_transport('http://foo/path/', possible_transports=[t5])
 
707
        self.assertIs(t5, t6)
 
708
 
 
709
    def test_don_t_reuse_different_transport(self):
 
710
        t1 = get_transport('http://foo/path')
 
711
        t2 = get_transport('http://bar/path', possible_transports=[t1])
 
712
        self.assertIsNot(t1, t2)
 
713
 
 
714
 
615
715
def get_test_permutations():
616
716
    """Return transport permutations to be used in testing.
617
717