← 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: 2011-07-22 14:11:10 UTC
  • mfrom: (6024.3.13 2.5-verbosity-knob-812928)
  • Revision ID: pqm@pqm.ubuntu.com-20110722141110-t853v5hoqst5iwr3
(jam) Fix hte release notes for verbosity knob.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
16
16
 
17
17
 
18
18
from cStringIO import StringIO
19
 
import os
20
19
import subprocess
21
20
import sys
22
21
import threading
39
38
    pathfilter,
40
39
    readonly,
41
40
    )
42
 
import bzrlib.transport.trace
43
41
from bzrlib.tests import (
44
42
    features,
45
43
    test_server,
91
89
        transport.register_lazy_transport(
92
90
            'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
93
91
        try:
94
 
            transport.get_transport_from_url('foo://fooserver/foo')
 
92
            transport.get_transport('foo://fooserver/foo')
95
93
        except errors.UnsupportedProtocol, e:
96
94
            e_str = str(e)
97
95
            self.assertEquals('Unsupported protocol'
111
109
            'foo', 'bzrlib.tests.test_transport', 'BackupTransportHandler')
112
110
        transport.register_lazy_transport(
113
111
            'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
114
 
        t = transport.get_transport_from_url('foo://fooserver/foo')
 
112
        t = transport.get_transport('foo://fooserver/foo')
115
113
        self.assertTrue(isinstance(t, BackupTransportHandler))
116
114
 
117
115
    def test_ssh_hints(self):
118
116
        """Transport ssh:// should raise an error pointing out bzr+ssh://"""
119
117
        try:
120
 
            transport.get_transport_from_url('ssh://fooserver/foo')
 
118
            transport.get_transport('ssh://fooserver/foo')
121
119
        except errors.UnsupportedProtocol, e:
122
120
            e_str = str(e)
123
121
            self.assertEquals('Unsupported protocol'
247
245
        server.start_server()
248
246
        url = server.get_url()
249
247
        self.assertTrue(url in transport.transport_list_registry)
250
 
        t = transport.get_transport_from_url(url)
 
248
        t = transport.get_transport(url)
251
249
        del t
252
250
        server.stop_server()
253
251
        self.assertFalse(url in transport.transport_list_registry)
368
366
    def test_abspath(self):
369
367
        # The abspath is always relative to the chroot_url.
370
368
        server = chroot.ChrootServer(
371
 
            transport.get_transport_from_url('memory:///foo/bar/'))
 
369
            transport.get_transport('memory:///foo/bar/'))
372
370
        self.start_server(server)
373
 
        t = transport.get_transport_from_url(server.get_url())
 
371
        t = transport.get_transport(server.get_url())
374
372
        self.assertEqual(server.get_url(), t.abspath('/'))
375
373
 
376
374
        subdir_t = t.clone('subdir')
378
376
 
379
377
    def test_clone(self):
380
378
        server = chroot.ChrootServer(
381
 
            transport.get_transport_from_url('memory:///foo/bar/'))
 
379
            transport.get_transport('memory:///foo/bar/'))
382
380
        self.start_server(server)
383
 
        t = transport.get_transport_from_url(server.get_url())
 
381
        t = transport.get_transport(server.get_url())
384
382
        # relpath from root and root path are the same
385
383
        relpath_cloned = t.clone('foo')
386
384
        abspath_cloned = t.clone('/foo')
395
393
        This is so that it is not possible to escape a chroot by doing::
396
394
            url = chroot_transport.base
397
395
            parent_url = urlutils.join(url, '..')
398
 
            new_t = transport.get_transport_from_url(parent_url)
 
396
            new_t = transport.get_transport(parent_url)
399
397
        """
400
398
        server = chroot.ChrootServer(
401
 
            transport.get_transport_from_url('memory:///path/subpath'))
 
399
            transport.get_transport('memory:///path/subpath'))
402
400
        self.start_server(server)
403
 
        t = transport.get_transport_from_url(server.get_url())
404
 
        new_t = transport.get_transport_from_url(t.base)
 
401
        t = transport.get_transport(server.get_url())
 
402
        new_t = transport.get_transport(t.base)
405
403
        self.assertEqual(t.server, new_t.server)
406
404
        self.assertEqual(t.base, new_t.base)
407
405
 
412
410
        This is so that it is not possible to escape a chroot by doing::
413
411
            url = chroot_transport.base
414
412
            parent_url = urlutils.join(url, '..')
415
 
            new_t = transport.get_transport_from_url(parent_url)
 
413
            new_t = transport.get_transport(parent_url)
416
414
        """
417
415
        server = chroot.ChrootServer(
418
 
            transport.get_transport_from_url('memory:///path/'))
 
416
            transport.get_transport('memory:///path/'))
419
417
        self.start_server(server)
420
 
        t = transport.get_transport_from_url(server.get_url())
 
418
        t = transport.get_transport(server.get_url())
421
419
        self.assertRaises(
422
420
            errors.InvalidURLJoin, urlutils.join, t.base, '..')
423
421
 
459
457
    def test_abspath(self):
460
458
        # The abspath is always relative to the base of the backing transport.
461
459
        server = pathfilter.PathFilteringServer(
462
 
            transport.get_transport_from_url('memory:///foo/bar/'),
 
460
            transport.get_transport('memory:///foo/bar/'),
463
461
            lambda x: x)
464
462
        server.start_server()
465
 
        t = transport.get_transport_from_url(server.get_url())
 
463
        t = transport.get_transport(server.get_url())
466
464
        self.assertEqual(server.get_url(), t.abspath('/'))
467
465
 
468
466
        subdir_t = t.clone('subdir')
477
475
        if filter_func is None:
478
476
            filter_func = lambda x: x
479
477
        server = pathfilter.PathFilteringServer(
480
 
            transport.get_transport_from_url('memory:///foo/bar/'), filter_func)
 
478
            transport.get_transport('memory:///foo/bar/'), filter_func)
481
479
        server.start_server()
482
480
        self.addCleanup(server.stop_server)
483
 
        return transport.get_transport_from_url(server.get_url())
 
481
        return transport.get_transport(server.get_url())
484
482
 
485
483
    def test__filter(self):
486
484
        # _filter (with an identity func as filter_func) always returns
530
528
        otherwise) the filtering by doing::
531
529
            url = filtered_transport.base
532
530
            parent_url = urlutils.join(url, '..')
533
 
            new_t = transport.get_transport_from_url(parent_url)
 
531
            new_t = transport.get_transport(parent_url)
534
532
        """
535
533
        t = self.make_pf_transport()
536
 
        new_t = transport.get_transport_from_url(t.base)
 
534
        new_t = transport.get_transport(t.base)
537
535
        self.assertEqual(t.server, new_t.server)
538
536
        self.assertEqual(t.base, new_t.base)
539
537
 
552
550
        # connect to '.' via http which is not listable
553
551
        server = HttpServer()
554
552
        self.start_server(server)
555
 
        t = transport.get_transport_from_url('readonly+' + server.get_url())
 
553
        t = transport.get_transport('readonly+' + server.get_url())
556
554
        self.assertIsInstance(t, readonly.ReadonlyTransportDecorator)
557
555
        self.assertEqual(False, t.listable())
558
556
        self.assertEqual(True, t.is_readonly())
591
589
        # the url should be decorated appropriately
592
590
        self.assertStartsWith(server.get_url(), 'fakenfs+')
593
591
        # and we should be able to get a transport for it
594
 
        t = transport.get_transport_from_url(server.get_url())
 
592
        t = transport.get_transport(server.get_url())
595
593
        # which must be a FakeNFSTransportDecorator instance.
596
594
        self.assertIsInstance(t, fakenfs.FakeNFSTransportDecorator)
597
595
 
674
672
        base_url = self._server.get_url()
675
673
        url = self._adjust_url(base_url, relpath)
676
674
        # try getting the transport via the regular interface:
677
 
        t = transport.get_transport_from_url(url)
 
675
        t = transport.get_transport(url)
678
676
        # vila--20070607 if the following are commented out the test suite
679
677
        # still pass. Is this really still needed or was it a forgotten
680
678
        # temporary fix ?
685
683
        return t
686
684
 
687
685
 
688
 
class TestTransportFromPath(tests.TestCaseInTempDir):
689
 
 
690
 
    def test_with_path(self):
691
 
        t = transport.get_transport_from_path(self.test_dir)
692
 
        self.assertIsInstance(t, local.LocalTransport)
693
 
        self.assertEquals(t.base.rstrip("/"),
694
 
            urlutils.local_path_to_url(self.test_dir))
695
 
 
696
 
    def test_with_url(self):
697
 
        t = transport.get_transport_from_path("file:")
698
 
        self.assertIsInstance(t, local.LocalTransport)
699
 
        self.assertEquals(t.base.rstrip("/"),
700
 
            urlutils.local_path_to_url(os.path.join(self.test_dir, "file:")))
701
 
 
702
 
 
703
 
class TestTransportFromUrl(tests.TestCaseInTempDir):
704
 
 
705
 
    def test_with_path(self):
706
 
        self.assertRaises(errors.InvalidURL, transport.get_transport_from_url,
707
 
            self.test_dir)
708
 
 
709
 
    def test_with_url(self):
710
 
        url = urlutils.local_path_to_url(self.test_dir)
711
 
        t = transport.get_transport_from_url(url)
712
 
        self.assertIsInstance(t, local.LocalTransport)
713
 
        self.assertEquals(t.base.rstrip("/"), url)
714
 
 
715
 
 
716
686
class TestLocalTransports(tests.TestCase):
717
687
 
718
688
    def test_get_transport_from_abspath(self):
740
710
        self.assertEquals(t.local_abspath(''), here)
741
711
 
742
712
 
743
 
class TestLocalTransportWriteStream(tests.TestCaseWithTransport):
744
 
 
745
 
    def test_local_fdatasync_calls_fdatasync(self):
746
 
        """Check fdatasync on a stream tries to flush the data to the OS.
747
 
        
748
 
        We can't easily observe the external effect but we can at least see
749
 
        it's called.
750
 
        """
751
 
        t = self.get_transport('.')
752
 
        calls = self.recordCalls(os, 'fdatasync')
753
 
        w = t.open_write_stream('out')
754
 
        w.write('foo')
755
 
        w.fdatasync()
756
 
        with open('out', 'rb') as f:
757
 
            # Should have been flushed.
758
 
            self.assertEquals(f.read(), 'foo')
759
 
        self.assertEquals(len(calls), 1, calls)
760
 
 
761
 
 
762
713
class TestWin32LocalTransport(tests.TestCase):
763
714
 
764
715
    def test_unc_clone_to_root(self):
864
815
 
865
816
    def test_reuse_same_transport(self):
866
817
        possible_transports = []
867
 
        t1 = transport.get_transport_from_url('http://foo/',
 
818
        t1 = transport.get_transport('http://foo/',
868
819
                                     possible_transports=possible_transports)
869
820
        self.assertEqual([t1], possible_transports)
870
 
        t2 = transport.get_transport_from_url('http://foo/',
 
821
        t2 = transport.get_transport('http://foo/',
871
822
                                     possible_transports=[t1])
872
823
        self.assertIs(t1, t2)
873
824
 
874
825
        # Also check that final '/' are handled correctly
875
 
        t3 = transport.get_transport_from_url('http://foo/path/')
876
 
        t4 = transport.get_transport_from_url('http://foo/path',
 
826
        t3 = transport.get_transport('http://foo/path/')
 
827
        t4 = transport.get_transport('http://foo/path',
877
828
                                     possible_transports=[t3])
878
829
        self.assertIs(t3, t4)
879
830
 
880
 
        t5 = transport.get_transport_from_url('http://foo/path')
881
 
        t6 = transport.get_transport_from_url('http://foo/path/',
 
831
        t5 = transport.get_transport('http://foo/path')
 
832
        t6 = transport.get_transport('http://foo/path/',
882
833
                                     possible_transports=[t5])
883
834
        self.assertIs(t5, t6)
884
835
 
885
836
    def test_don_t_reuse_different_transport(self):
886
 
        t1 = transport.get_transport_from_url('http://foo/path')
887
 
        t2 = transport.get_transport_from_url('http://bar/path',
 
837
        t1 = transport.get_transport('http://foo/path')
 
838
        t2 = transport.get_transport('http://bar/path',
888
839
                                     possible_transports=[t1])
889
840
        self.assertIsNot(t1, t2)
890
841
 
891
842
 
892
843
class TestTransportTrace(tests.TestCase):
893
844
 
894
 
    def test_decorator(self):
895
 
        t = transport.get_transport_from_url('trace+memory://')
 
845
    def test_get(self):
 
846
        t = transport.get_transport('trace+memory://')
896
847
        self.assertIsInstance(
897
848
            t, bzrlib.transport.trace.TransportTraceDecorator)
898
849
 
899
850
    def test_clone_preserves_activity(self):
900
 
        t = transport.get_transport_from_url('trace+memory://')
 
851
        t = transport.get_transport('trace+memory://')
901
852
        t2 = t.clone('.')
902
853
        self.assertTrue(t is not t2)
903
854
        self.assertTrue(t._activity is t2._activity)
907
858
    # still won't cause a test failure when the top level Transport API
908
859
    # changes; so there is little return doing that.
909
860
    def test_get(self):
910
 
        t = transport.get_transport_from_url('trace+memory:///')
 
861
        t = transport.get_transport('trace+memory:///')
911
862
        t.put_bytes('foo', 'barish')
912
863
        t.get('foo')
913
864
        expected_result = []
919
870
        self.assertEqual(expected_result, t._activity)
920
871
 
921
872
    def test_readv(self):
922
 
        t = transport.get_transport_from_url('trace+memory:///')
 
873
        t = transport.get_transport('trace+memory:///')
923
874
        t.put_bytes('foo', 'barish')
924
875
        list(t.readv('foo', [(0, 1), (3, 2)],
925
876
                     adjust_for_latency=True, upper_limit=6))
1072
1023
 
1073
1024
    def test_path(self):
1074
1025
        self.assertEquals("file:///foo/bar", location_to_url("/foo/bar"))
1075
 
 
1076
 
    def test_relative_file_url(self):
1077
 
        self.assertEquals(urlutils.local_path_to_url(".") + "/bar",
1078
 
            location_to_url("file:bar"))
1079
 
 
1080
 
    def test_absolute_file_url(self):
1081
 
        self.assertEquals("file:///bar", location_to_url("file:/bar"))