← 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: 2015-09-08 07:59:37 UTC
  • mto: This revision was merged to the branch mainline in revision 6604.
  • Revision ID: v.ladeuil+lp@free.fr-20150908075937-u63hgo67v3c2lapt
Skip the missed test in test_transport.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
1
 
# Copyright (C) 2005-2011 Canonical Ltd
 
1
# Copyright (C) 2005-2011, 2015 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
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):
 
818
        self.requireFeature(features.win32_feature)
714
819
        # Win32 UNC path like \\HOST\path
715
820
        # clone to root should stop at least at \\HOST part
716
821
        # not on \\
729
834
    def test_parse_url(self):
730
835
        t = transport.ConnectedTransport(
731
836
            '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)
 
837
        self.assertEquals(t._parsed_url.host, 'simple.example.com')
 
838
        self.assertEquals(t._parsed_url.port, None)
 
839
        self.assertEquals(t._parsed_url.path, '/home/source/')
 
840
        self.assertTrue(t._parsed_url.user is None)
 
841
        self.assertTrue(t._parsed_url.password is None)
737
842
 
738
843
        self.assertEquals(t.base, 'http://simple.example.com/home/source/')
739
844
 
740
845
    def test_parse_url_with_at_in_user(self):
741
846
        # Bug 228058
742
847
        t = transport.ConnectedTransport('ftp://user@host.com@www.host.com/')
743
 
        self.assertEquals(t._user, 'user@host.com')
 
848
        self.assertEquals(t._parsed_url.user, 'user@host.com')
744
849
 
745
850
    def test_parse_quoted_url(self):
746
851
        t = transport.ConnectedTransport(
747
852
            '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/')
 
853
        self.assertEquals(t._parsed_url.host, 'exAmple.com')
 
854
        self.assertEquals(t._parsed_url.port, 2222)
 
855
        self.assertEquals(t._parsed_url.user, 'robey')
 
856
        self.assertEquals(t._parsed_url.password, 'h@t')
 
857
        self.assertEquals(t._parsed_url.path, '/path/')
753
858
 
754
859
        # Base should not keep track of the password
755
 
        self.assertEquals(t.base, 'http://robey@exAmple.com:2222/path/')
 
860
        self.assertEquals(t.base, 'http://ro%62ey@ex%41mple.com:2222/path/')
756
861
 
757
862
    def test_parse_invalid_url(self):
758
863
        self.assertRaises(errors.InvalidURL,
782
887
 
783
888
    def test_connection_sharing_propagate_credentials(self):
784
889
        t = transport.ConnectedTransport('ftp://user@host.com/abs/path')
785
 
        self.assertEquals('user', t._user)
786
 
        self.assertEquals('host.com', t._host)
 
890
        self.assertEquals('user', t._parsed_url.user)
 
891
        self.assertEquals('host.com', t._parsed_url.host)
787
892
        self.assertIs(None, t._get_connection())
788
 
        self.assertIs(None, t._password)
 
893
        self.assertIs(None, t._parsed_url.password)
789
894
        c = t.clone('subdir')
790
895
        self.assertIs(None, c._get_connection())
791
 
        self.assertIs(None, t._password)
 
896
        self.assertIs(None, t._parsed_url.password)
792
897
 
793
898
        # Simulate the user entering a password
794
899
        password = 'secret'
813
918
 
814
919
    def test_reuse_same_transport(self):
815
920
        possible_transports = []
816
 
        t1 = transport.get_transport('http://foo/',
 
921
        t1 = transport.get_transport_from_url('http://foo/',
817
922
                                     possible_transports=possible_transports)
818
923
        self.assertEqual([t1], possible_transports)
819
 
        t2 = transport.get_transport('http://foo/',
 
924
        t2 = transport.get_transport_from_url('http://foo/',
820
925
                                     possible_transports=[t1])
821
926
        self.assertIs(t1, t2)
822
927
 
823
928
        # Also check that final '/' are handled correctly
824
 
        t3 = transport.get_transport('http://foo/path/')
825
 
        t4 = transport.get_transport('http://foo/path',
 
929
        t3 = transport.get_transport_from_url('http://foo/path/')
 
930
        t4 = transport.get_transport_from_url('http://foo/path',
826
931
                                     possible_transports=[t3])
827
932
        self.assertIs(t3, t4)
828
933
 
829
 
        t5 = transport.get_transport('http://foo/path')
830
 
        t6 = transport.get_transport('http://foo/path/',
 
934
        t5 = transport.get_transport_from_url('http://foo/path')
 
935
        t6 = transport.get_transport_from_url('http://foo/path/',
831
936
                                     possible_transports=[t5])
832
937
        self.assertIs(t5, t6)
833
938
 
834
939
    def test_don_t_reuse_different_transport(self):
835
 
        t1 = transport.get_transport('http://foo/path')
836
 
        t2 = transport.get_transport('http://bar/path',
 
940
        t1 = transport.get_transport_from_url('http://foo/path')
 
941
        t2 = transport.get_transport_from_url('http://bar/path',
837
942
                                     possible_transports=[t1])
838
943
        self.assertIsNot(t1, t2)
839
944
 
840
945
 
841
946
class TestTransportTrace(tests.TestCase):
842
947
 
843
 
    def test_get(self):
844
 
        t = transport.get_transport('trace+memory://')
 
948
    def test_decorator(self):
 
949
        t = transport.get_transport_from_url('trace+memory://')
845
950
        self.assertIsInstance(
846
951
            t, bzrlib.transport.trace.TransportTraceDecorator)
847
952
 
848
953
    def test_clone_preserves_activity(self):
849
 
        t = transport.get_transport('trace+memory://')
 
954
        t = transport.get_transport_from_url('trace+memory://')
850
955
        t2 = t.clone('.')
851
956
        self.assertTrue(t is not t2)
852
957
        self.assertTrue(t._activity is t2._activity)
856
961
    # still won't cause a test failure when the top level Transport API
857
962
    # changes; so there is little return doing that.
858
963
    def test_get(self):
859
 
        t = transport.get_transport('trace+memory:///')
 
964
        t = transport.get_transport_from_url('trace+memory:///')
860
965
        t.put_bytes('foo', 'barish')
861
966
        t.get('foo')
862
967
        expected_result = []
868
973
        self.assertEqual(expected_result, t._activity)
869
974
 
870
975
    def test_readv(self):
871
 
        t = transport.get_transport('trace+memory:///')
 
976
        t = transport.get_transport_from_url('trace+memory:///')
872
977
        t.put_bytes('foo', 'barish')
873
978
        list(t.readv('foo', [(0, 1), (3, 2)],
874
979
                     adjust_for_latency=True, upper_limit=6))
993
1098
        result = http.unhtml_roughly(fake_html)
994
1099
        self.assertEquals(len(result), 1000)
995
1100
        self.assertStartsWith(result, " something!")
 
1101
 
 
1102
 
 
1103
class SomeDirectory(object):
 
1104
 
 
1105
    def look_up(self, name, url):
 
1106
        return "http://bar"
 
1107
 
 
1108
 
 
1109
class TestLocationToUrl(tests.TestCase):
 
1110
 
 
1111
    def get_base_location(self):
 
1112
        path = osutils.abspath('/foo/bar')
 
1113
        if path.startswith('/'):
 
1114
            url = 'file://%s' % (path,)
 
1115
        else:
 
1116
            # On Windows, abspaths start with the drive letter, so we have to
 
1117
            # add in the extra '/'
 
1118
            url = 'file:///%s' % (path,)
 
1119
        return path, url
 
1120
 
 
1121
    def test_regular_url(self):
 
1122
        self.assertEquals("file://foo", location_to_url("file://foo"))
 
1123
 
 
1124
    def test_directory(self):
 
1125
        directories.register("bar:", SomeDirectory, "Dummy directory")
 
1126
        self.addCleanup(directories.remove, "bar:")
 
1127
        self.assertEquals("http://bar", location_to_url("bar:"))
 
1128
 
 
1129
    def test_unicode_url(self):
 
1130
        self.assertRaises(errors.InvalidURL, location_to_url,
 
1131
            "http://fo/\xc3\xaf".decode("utf-8"))
 
1132
 
 
1133
    def test_unicode_path(self):
 
1134
        path, url = self.get_base_location()
 
1135
        location = path + "\xc3\xaf".decode("utf-8")
 
1136
        url += '%C3%AF'
 
1137
        self.assertEquals(url, location_to_url(location))
 
1138
 
 
1139
    def test_path(self):
 
1140
        path, url = self.get_base_location()
 
1141
        self.assertEquals(url, location_to_url(path))
 
1142
 
 
1143
    def test_relative_file_url(self):
 
1144
        self.assertEquals(urlutils.local_path_to_url(".") + "/bar",
 
1145
            location_to_url("file:bar"))
 
1146
 
 
1147
    def test_absolute_file_url(self):
 
1148
        self.assertEquals("file:///bar", location_to_url("file:/bar"))