← 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: Vincent Ladeuil
  • Date: 2012-03-13 17:25:29 UTC
  • mfrom: (6499 +trunk)
  • mto: This revision was merged to the branch mainline in revision 6501.
  • Revision ID: v.ladeuil+lp@free.fr-20120313172529-i0suyjnepsor25i7
Merge trunk

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 errno
 
20
import os
19
21
import subprocess
20
22
import sys
21
23
import threading
27
29
    transport,
28
30
    urlutils,
29
31
    )
 
32
from bzrlib.directory_service import directories
30
33
from bzrlib.transport import (
31
34
    chroot,
32
35
    fakenfs,
33
36
    http,
34
37
    local,
 
38
    location_to_url,
35
39
    memory,
36
40
    pathfilter,
37
41
    readonly,
38
42
    )
 
43
import bzrlib.transport.trace
39
44
from bzrlib.tests import (
40
45
    features,
41
46
    test_server,
87
92
        transport.register_lazy_transport(
88
93
            'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
89
94
        try:
90
 
            transport.get_transport('foo://fooserver/foo')
 
95
            transport.get_transport_from_url('foo://fooserver/foo')
91
96
        except errors.UnsupportedProtocol, e:
92
97
            e_str = str(e)
93
98
            self.assertEquals('Unsupported protocol'
107
112
            'foo', 'bzrlib.tests.test_transport', 'BackupTransportHandler')
108
113
        transport.register_lazy_transport(
109
114
            'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
110
 
        t = transport.get_transport('foo://fooserver/foo')
 
115
        t = transport.get_transport_from_url('foo://fooserver/foo')
111
116
        self.assertTrue(isinstance(t, BackupTransportHandler))
112
117
 
113
118
    def test_ssh_hints(self):
114
119
        """Transport ssh:// should raise an error pointing out bzr+ssh://"""
115
120
        try:
116
 
            transport.get_transport('ssh://fooserver/foo')
 
121
            transport.get_transport_from_url('ssh://fooserver/foo')
117
122
        except errors.UnsupportedProtocol, e:
118
123
            e_str = str(e)
119
124
            self.assertEquals('Unsupported protocol'
134
139
        self.assertRaises(errors.ReadError, a_file.read, 40)
135
140
        a_file.close()
136
141
 
137
 
    def test__combine_paths(self):
138
 
        t = transport.Transport('/')
139
 
        self.assertEqual('/home/sarah/project/foo',
140
 
                         t._combine_paths('/home/sarah', 'project/foo'))
141
 
        self.assertEqual('/etc',
142
 
                         t._combine_paths('/home/sarah', '../../etc'))
143
 
        self.assertEqual('/etc',
144
 
                         t._combine_paths('/home/sarah', '../../../etc'))
145
 
        self.assertEqual('/etc',
146
 
                         t._combine_paths('/home/sarah', '/etc'))
147
 
 
148
142
    def test_local_abspath_non_local_transport(self):
149
143
        # the base implementation should throw
150
144
        t = memory.MemoryTransport()
243
237
        server.start_server()
244
238
        url = server.get_url()
245
239
        self.assertTrue(url in transport.transport_list_registry)
246
 
        t = transport.get_transport(url)
 
240
        t = transport.get_transport_from_url(url)
247
241
        del t
248
242
        server.stop_server()
249
243
        self.assertFalse(url in transport.transport_list_registry)
364
358
    def test_abspath(self):
365
359
        # The abspath is always relative to the chroot_url.
366
360
        server = chroot.ChrootServer(
367
 
            transport.get_transport('memory:///foo/bar/'))
 
361
            transport.get_transport_from_url('memory:///foo/bar/'))
368
362
        self.start_server(server)
369
 
        t = transport.get_transport(server.get_url())
 
363
        t = transport.get_transport_from_url(server.get_url())
370
364
        self.assertEqual(server.get_url(), t.abspath('/'))
371
365
 
372
366
        subdir_t = t.clone('subdir')
374
368
 
375
369
    def test_clone(self):
376
370
        server = chroot.ChrootServer(
377
 
            transport.get_transport('memory:///foo/bar/'))
 
371
            transport.get_transport_from_url('memory:///foo/bar/'))
378
372
        self.start_server(server)
379
 
        t = transport.get_transport(server.get_url())
 
373
        t = transport.get_transport_from_url(server.get_url())
380
374
        # relpath from root and root path are the same
381
375
        relpath_cloned = t.clone('foo')
382
376
        abspath_cloned = t.clone('/foo')
391
385
        This is so that it is not possible to escape a chroot by doing::
392
386
            url = chroot_transport.base
393
387
            parent_url = urlutils.join(url, '..')
394
 
            new_t = transport.get_transport(parent_url)
 
388
            new_t = transport.get_transport_from_url(parent_url)
395
389
        """
396
390
        server = chroot.ChrootServer(
397
 
            transport.get_transport('memory:///path/subpath'))
 
391
            transport.get_transport_from_url('memory:///path/subpath'))
398
392
        self.start_server(server)
399
 
        t = transport.get_transport(server.get_url())
400
 
        new_t = transport.get_transport(t.base)
 
393
        t = transport.get_transport_from_url(server.get_url())
 
394
        new_t = transport.get_transport_from_url(t.base)
401
395
        self.assertEqual(t.server, new_t.server)
402
396
        self.assertEqual(t.base, new_t.base)
403
397
 
408
402
        This is so that it is not possible to escape a chroot by doing::
409
403
            url = chroot_transport.base
410
404
            parent_url = urlutils.join(url, '..')
411
 
            new_t = transport.get_transport(parent_url)
 
405
            new_t = transport.get_transport_from_url(parent_url)
412
406
        """
413
407
        server = chroot.ChrootServer(
414
 
            transport.get_transport('memory:///path/'))
 
408
            transport.get_transport_from_url('memory:///path/'))
415
409
        self.start_server(server)
416
 
        t = transport.get_transport(server.get_url())
 
410
        t = transport.get_transport_from_url(server.get_url())
417
411
        self.assertRaises(
418
412
            errors.InvalidURLJoin, urlutils.join, t.base, '..')
419
413
 
449
443
        self.assertEqual('chroot-%d:///' % id(server), server.get_url())
450
444
 
451
445
 
 
446
class TestHooks(tests.TestCase):
 
447
    """Basic tests for transport hooks"""
 
448
 
 
449
    def _get_connected_transport(self):
 
450
        return transport.ConnectedTransport("bogus:nowhere")
 
451
 
 
452
    def test_transporthooks_initialisation(self):
 
453
        """Check all expected transport hook points are set up"""
 
454
        hookpoint = transport.TransportHooks()
 
455
        self.assertTrue("post_connect" in hookpoint,
 
456
            "post_connect not in %s" % (hookpoint,))
 
457
 
 
458
    def test_post_connect(self):
 
459
        """Ensure the post_connect hook is called when _set_transport is"""
 
460
        calls = []
 
461
        transport.Transport.hooks.install_named_hook("post_connect",
 
462
            calls.append, None)
 
463
        t = self._get_connected_transport()
 
464
        self.assertLength(0, calls)
 
465
        t._set_connection("connection", "auth")
 
466
        self.assertEqual(calls, [t])
 
467
 
 
468
 
452
469
class PathFilteringDecoratorTransportTest(tests.TestCase):
453
470
    """Pathfilter decoration specific tests."""
454
471
 
455
472
    def test_abspath(self):
456
473
        # The abspath is always relative to the base of the backing transport.
457
474
        server = pathfilter.PathFilteringServer(
458
 
            transport.get_transport('memory:///foo/bar/'),
 
475
            transport.get_transport_from_url('memory:///foo/bar/'),
459
476
            lambda x: x)
460
477
        server.start_server()
461
 
        t = transport.get_transport(server.get_url())
 
478
        t = transport.get_transport_from_url(server.get_url())
462
479
        self.assertEqual(server.get_url(), t.abspath('/'))
463
480
 
464
481
        subdir_t = t.clone('subdir')
473
490
        if filter_func is None:
474
491
            filter_func = lambda x: x
475
492
        server = pathfilter.PathFilteringServer(
476
 
            transport.get_transport('memory:///foo/bar/'), filter_func)
 
493
            transport.get_transport_from_url('memory:///foo/bar/'), filter_func)
477
494
        server.start_server()
478
495
        self.addCleanup(server.stop_server)
479
 
        return transport.get_transport(server.get_url())
 
496
        return transport.get_transport_from_url(server.get_url())
480
497
 
481
498
    def test__filter(self):
482
499
        # _filter (with an identity func as filter_func) always returns
526
543
        otherwise) the filtering by doing::
527
544
            url = filtered_transport.base
528
545
            parent_url = urlutils.join(url, '..')
529
 
            new_t = transport.get_transport(parent_url)
 
546
            new_t = transport.get_transport_from_url(parent_url)
530
547
        """
531
548
        t = self.make_pf_transport()
532
 
        new_t = transport.get_transport(t.base)
 
549
        new_t = transport.get_transport_from_url(t.base)
533
550
        self.assertEqual(t.server, new_t.server)
534
551
        self.assertEqual(t.base, new_t.base)
535
552
 
548
565
        # connect to '.' via http which is not listable
549
566
        server = HttpServer()
550
567
        self.start_server(server)
551
 
        t = transport.get_transport('readonly+' + server.get_url())
 
568
        t = transport.get_transport_from_url('readonly+' + server.get_url())
552
569
        self.assertIsInstance(t, readonly.ReadonlyTransportDecorator)
553
570
        self.assertEqual(False, t.listable())
554
571
        self.assertEqual(True, t.is_readonly())
587
604
        # the url should be decorated appropriately
588
605
        self.assertStartsWith(server.get_url(), 'fakenfs+')
589
606
        # and we should be able to get a transport for it
590
 
        t = transport.get_transport(server.get_url())
 
607
        t = transport.get_transport_from_url(server.get_url())
591
608
        # which must be a FakeNFSTransportDecorator instance.
592
609
        self.assertIsInstance(t, fakenfs.FakeNFSTransportDecorator)
593
610
 
670
687
        base_url = self._server.get_url()
671
688
        url = self._adjust_url(base_url, relpath)
672
689
        # try getting the transport via the regular interface:
673
 
        t = transport.get_transport(url)
 
690
        t = transport.get_transport_from_url(url)
674
691
        # vila--20070607 if the following are commented out the test suite
675
692
        # still pass. Is this really still needed or was it a forgotten
676
693
        # temporary fix ?
681
698
        return t
682
699
 
683
700
 
 
701
class TestTransportFromPath(tests.TestCaseInTempDir):
 
702
 
 
703
    def test_with_path(self):
 
704
        t = transport.get_transport_from_path(self.test_dir)
 
705
        self.assertIsInstance(t, local.LocalTransport)
 
706
        self.assertEquals(t.base.rstrip("/"),
 
707
            urlutils.local_path_to_url(self.test_dir))
 
708
 
 
709
    def test_with_url(self):
 
710
        t = transport.get_transport_from_path("file:")
 
711
        self.assertIsInstance(t, local.LocalTransport)
 
712
        self.assertEquals(t.base.rstrip("/"),
 
713
            urlutils.local_path_to_url(os.path.join(self.test_dir, "file:")))
 
714
 
 
715
 
 
716
class TestTransportFromUrl(tests.TestCaseInTempDir):
 
717
 
 
718
    def test_with_path(self):
 
719
        self.assertRaises(errors.InvalidURL, transport.get_transport_from_url,
 
720
            self.test_dir)
 
721
 
 
722
    def test_with_url(self):
 
723
        url = urlutils.local_path_to_url(self.test_dir)
 
724
        t = transport.get_transport_from_url(url)
 
725
        self.assertIsInstance(t, local.LocalTransport)
 
726
        self.assertEquals(t.base.rstrip("/"), url)
 
727
 
 
728
    def test_with_url_and_segment_parameters(self):
 
729
        url = urlutils.local_path_to_url(self.test_dir)+",branch=foo"
 
730
        t = transport.get_transport_from_url(url)
 
731
        self.assertIsInstance(t, local.LocalTransport)
 
732
        self.assertEquals(t.base.rstrip("/"), url)
 
733
        with open(os.path.join(self.test_dir, "afile"), 'w') as f:
 
734
            f.write("data")
 
735
        self.assertTrue(t.has("afile"))
 
736
 
 
737
 
684
738
class TestLocalTransports(tests.TestCase):
685
739
 
686
740
    def test_get_transport_from_abspath(self):
708
762
        self.assertEquals(t.local_abspath(''), here)
709
763
 
710
764
 
 
765
class TestLocalTransportMutation(tests.TestCaseInTempDir):
 
766
 
 
767
    def test_local_transport_mkdir(self):
 
768
        here = osutils.abspath('.')
 
769
        t = transport.get_transport(here)
 
770
        t.mkdir('test')
 
771
        self.assertTrue(os.path.exists('test'))
 
772
 
 
773
    def test_local_transport_mkdir_permission_denied(self):
 
774
        # See https://bugs.launchpad.net/bzr/+bug/606537
 
775
        here = osutils.abspath('.')
 
776
        t = transport.get_transport(here)
 
777
        def fake_chmod(path, mode):
 
778
            e = OSError('permission denied')
 
779
            e.errno = errno.EPERM
 
780
            raise e
 
781
        self.overrideAttr(os, 'chmod', fake_chmod)
 
782
        t.mkdir('test')
 
783
        t.mkdir('test2', mode=0707)
 
784
        self.assertTrue(os.path.exists('test'))
 
785
        self.assertTrue(os.path.exists('test2'))
 
786
 
 
787
 
 
788
class TestLocalTransportWriteStream(tests.TestCaseWithTransport):
 
789
 
 
790
    def test_local_fdatasync_calls_fdatasync(self):
 
791
        """Check fdatasync on a stream tries to flush the data to the OS.
 
792
        
 
793
        We can't easily observe the external effect but we can at least see
 
794
        it's called.
 
795
        """
 
796
        sentinel = object()
 
797
        fdatasync = getattr(os, 'fdatasync', sentinel)
 
798
        if fdatasync is sentinel:
 
799
            raise tests.TestNotApplicable('fdatasync not supported')
 
800
        t = self.get_transport('.')
 
801
        calls = self.recordCalls(os, 'fdatasync')
 
802
        w = t.open_write_stream('out')
 
803
        w.write('foo')
 
804
        w.fdatasync()
 
805
        with open('out', 'rb') as f:
 
806
            # Should have been flushed.
 
807
            self.assertEquals(f.read(), 'foo')
 
808
        self.assertEquals(len(calls), 1, calls)
 
809
 
 
810
    def test_missing_directory(self):
 
811
        t = self.get_transport('.')
 
812
        self.assertRaises(errors.NoSuchFile, t.open_write_stream, 'dir/foo')
 
813
 
 
814
 
711
815
class TestWin32LocalTransport(tests.TestCase):
712
816
 
713
817
    def test_unc_clone_to_root(self):
729
833
    def test_parse_url(self):
730
834
        t = transport.ConnectedTransport(
731
835
            'http://simple.example.com/home/source')
732
 
        self.assertEquals(t._host, 'simple.example.com')
733
 
        self.assertEquals(t._port, None)
734
 
        self.assertEquals(t._path, '/home/source/')
735
 
        self.assertTrue(t._user is None)
736
 
        self.assertTrue(t._password is None)
 
836
        self.assertEquals(t._parsed_url.host, 'simple.example.com')
 
837
        self.assertEquals(t._parsed_url.port, None)
 
838
        self.assertEquals(t._parsed_url.path, '/home/source/')
 
839
        self.assertTrue(t._parsed_url.user is None)
 
840
        self.assertTrue(t._parsed_url.password is None)
737
841
 
738
842
        self.assertEquals(t.base, 'http://simple.example.com/home/source/')
739
843
 
740
844
    def test_parse_url_with_at_in_user(self):
741
845
        # Bug 228058
742
846
        t = transport.ConnectedTransport('ftp://user@host.com@www.host.com/')
743
 
        self.assertEquals(t._user, 'user@host.com')
 
847
        self.assertEquals(t._parsed_url.user, 'user@host.com')
744
848
 
745
849
    def test_parse_quoted_url(self):
746
850
        t = transport.ConnectedTransport(
747
851
            'http://ro%62ey:h%40t@ex%41mple.com:2222/path')
748
 
        self.assertEquals(t._host, 'exAmple.com')
749
 
        self.assertEquals(t._port, 2222)
750
 
        self.assertEquals(t._user, 'robey')
751
 
        self.assertEquals(t._password, 'h@t')
752
 
        self.assertEquals(t._path, '/path/')
 
852
        self.assertEquals(t._parsed_url.host, 'exAmple.com')
 
853
        self.assertEquals(t._parsed_url.port, 2222)
 
854
        self.assertEquals(t._parsed_url.user, 'robey')
 
855
        self.assertEquals(t._parsed_url.password, 'h@t')
 
856
        self.assertEquals(t._parsed_url.path, '/path/')
753
857
 
754
858
        # Base should not keep track of the password
755
 
        self.assertEquals(t.base, 'http://robey@exAmple.com:2222/path/')
 
859
        self.assertEquals(t.base, 'http://ro%62ey@ex%41mple.com:2222/path/')
756
860
 
757
861
    def test_parse_invalid_url(self):
758
862
        self.assertRaises(errors.InvalidURL,
782
886
 
783
887
    def test_connection_sharing_propagate_credentials(self):
784
888
        t = transport.ConnectedTransport('ftp://user@host.com/abs/path')
785
 
        self.assertEquals('user', t._user)
786
 
        self.assertEquals('host.com', t._host)
 
889
        self.assertEquals('user', t._parsed_url.user)
 
890
        self.assertEquals('host.com', t._parsed_url.host)
787
891
        self.assertIs(None, t._get_connection())
788
 
        self.assertIs(None, t._password)
 
892
        self.assertIs(None, t._parsed_url.password)
789
893
        c = t.clone('subdir')
790
894
        self.assertIs(None, c._get_connection())
791
 
        self.assertIs(None, t._password)
 
895
        self.assertIs(None, t._parsed_url.password)
792
896
 
793
897
        # Simulate the user entering a password
794
898
        password = 'secret'
813
917
 
814
918
    def test_reuse_same_transport(self):
815
919
        possible_transports = []
816
 
        t1 = transport.get_transport('http://foo/',
 
920
        t1 = transport.get_transport_from_url('http://foo/',
817
921
                                     possible_transports=possible_transports)
818
922
        self.assertEqual([t1], possible_transports)
819
 
        t2 = transport.get_transport('http://foo/',
 
923
        t2 = transport.get_transport_from_url('http://foo/',
820
924
                                     possible_transports=[t1])
821
925
        self.assertIs(t1, t2)
822
926
 
823
927
        # Also check that final '/' are handled correctly
824
 
        t3 = transport.get_transport('http://foo/path/')
825
 
        t4 = transport.get_transport('http://foo/path',
 
928
        t3 = transport.get_transport_from_url('http://foo/path/')
 
929
        t4 = transport.get_transport_from_url('http://foo/path',
826
930
                                     possible_transports=[t3])
827
931
        self.assertIs(t3, t4)
828
932
 
829
 
        t5 = transport.get_transport('http://foo/path')
830
 
        t6 = transport.get_transport('http://foo/path/',
 
933
        t5 = transport.get_transport_from_url('http://foo/path')
 
934
        t6 = transport.get_transport_from_url('http://foo/path/',
831
935
                                     possible_transports=[t5])
832
936
        self.assertIs(t5, t6)
833
937
 
834
938
    def test_don_t_reuse_different_transport(self):
835
 
        t1 = transport.get_transport('http://foo/path')
836
 
        t2 = transport.get_transport('http://bar/path',
 
939
        t1 = transport.get_transport_from_url('http://foo/path')
 
940
        t2 = transport.get_transport_from_url('http://bar/path',
837
941
                                     possible_transports=[t1])
838
942
        self.assertIsNot(t1, t2)
839
943
 
840
944
 
841
945
class TestTransportTrace(tests.TestCase):
842
946
 
843
 
    def test_get(self):
844
 
        t = transport.get_transport('trace+memory://')
 
947
    def test_decorator(self):
 
948
        t = transport.get_transport_from_url('trace+memory://')
845
949
        self.assertIsInstance(
846
950
            t, bzrlib.transport.trace.TransportTraceDecorator)
847
951
 
848
952
    def test_clone_preserves_activity(self):
849
 
        t = transport.get_transport('trace+memory://')
 
953
        t = transport.get_transport_from_url('trace+memory://')
850
954
        t2 = t.clone('.')
851
955
        self.assertTrue(t is not t2)
852
956
        self.assertTrue(t._activity is t2._activity)
856
960
    # still won't cause a test failure when the top level Transport API
857
961
    # changes; so there is little return doing that.
858
962
    def test_get(self):
859
 
        t = transport.get_transport('trace+memory:///')
 
963
        t = transport.get_transport_from_url('trace+memory:///')
860
964
        t.put_bytes('foo', 'barish')
861
965
        t.get('foo')
862
966
        expected_result = []
868
972
        self.assertEqual(expected_result, t._activity)
869
973
 
870
974
    def test_readv(self):
871
 
        t = transport.get_transport('trace+memory:///')
 
975
        t = transport.get_transport_from_url('trace+memory:///')
872
976
        t.put_bytes('foo', 'barish')
873
977
        list(t.readv('foo', [(0, 1), (3, 2)],
874
978
                     adjust_for_latency=True, upper_limit=6))
993
1097
        result = http.unhtml_roughly(fake_html)
994
1098
        self.assertEquals(len(result), 1000)
995
1099
        self.assertStartsWith(result, " something!")
 
1100
 
 
1101
 
 
1102
class SomeDirectory(object):
 
1103
 
 
1104
    def look_up(self, name, url):
 
1105
        return "http://bar"
 
1106
 
 
1107
 
 
1108
class TestLocationToUrl(tests.TestCase):
 
1109
 
 
1110
    def get_base_location(self):
 
1111
        path = osutils.abspath('/foo/bar')
 
1112
        if path.startswith('/'):
 
1113
            url = 'file://%s' % (path,)
 
1114
        else:
 
1115
            # On Windows, abspaths start with the drive letter, so we have to
 
1116
            # add in the extra '/'
 
1117
            url = 'file:///%s' % (path,)
 
1118
        return path, url
 
1119
 
 
1120
    def test_regular_url(self):
 
1121
        self.assertEquals("file://foo", location_to_url("file://foo"))
 
1122
 
 
1123
    def test_directory(self):
 
1124
        directories.register("bar:", SomeDirectory, "Dummy directory")
 
1125
        self.addCleanup(directories.remove, "bar:")
 
1126
        self.assertEquals("http://bar", location_to_url("bar:"))
 
1127
 
 
1128
    def test_unicode_url(self):
 
1129
        self.assertRaises(errors.InvalidURL, location_to_url,
 
1130
            "http://fo/\xc3\xaf".decode("utf-8"))
 
1131
 
 
1132
    def test_unicode_path(self):
 
1133
        path, url = self.get_base_location()
 
1134
        location = path + "\xc3\xaf".decode("utf-8")
 
1135
        url += '%C3%AF'
 
1136
        self.assertEquals(url, location_to_url(location))
 
1137
 
 
1138
    def test_path(self):
 
1139
        path, url = self.get_base_location()
 
1140
        self.assertEquals(url, location_to_url(path))
 
1141
 
 
1142
    def test_relative_file_url(self):
 
1143
        self.assertEquals(urlutils.local_path_to_url(".") + "/bar",
 
1144
            location_to_url("file:bar"))
 
1145
 
 
1146
    def test_absolute_file_url(self):
 
1147
        self.assertEquals("file:///bar", location_to_url("file:/bar"))