/breezy/unstable

To get this branch, use:
bzr branch https://code.breezy-vcs.org/breezy/unstable

« back to all changes in this revision

Viewing changes to breezy/tests/test_transport.py

  • Committer: Jelmer Vernooij
  • Date: 2017-05-24 01:39:33 UTC
  • mfrom: (3815.3776.6)
  • Revision ID: jelmer@jelmer.uk-20170524013933-ir4y4tqtrsiz2ka2
New upstream snapshot.

Show diffs side-by-side

added added

removed removed

Lines of Context:
15
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
17
 
18
 
from cStringIO import StringIO
19
18
import errno
20
19
import os
21
20
import subprocess
22
21
import sys
23
22
import threading
24
23
 
25
 
from bzrlib import (
 
24
from .. import (
26
25
    errors,
27
26
    osutils,
28
27
    tests,
29
28
    transport,
30
29
    urlutils,
31
30
    )
32
 
from bzrlib.directory_service import directories
33
 
from bzrlib.transport import (
 
31
from ..directory_service import directories
 
32
from ..sixish import (
 
33
    BytesIO,
 
34
    )
 
35
from ..transport import (
34
36
    chroot,
35
37
    fakenfs,
36
38
    http,
40
42
    pathfilter,
41
43
    readonly,
42
44
    )
43
 
import bzrlib.transport.trace
44
 
from bzrlib.tests import (
 
45
import breezy.transport.trace
 
46
from . import (
45
47
    features,
46
48
    test_server,
47
49
    )
71
73
        transport._clear_protocol_handlers()
72
74
        transport.register_transport_proto('foo')
73
75
        transport.register_lazy_transport('foo',
74
 
                                            'bzrlib.tests.test_transport',
 
76
                                            'breezy.tests.test_transport',
75
77
                                            'TestTransport.SampleHandler')
76
78
        transport.register_transport_proto('bar')
77
79
        transport.register_lazy_transport('bar',
78
 
                                            'bzrlib.tests.test_transport',
 
80
                                            'breezy.tests.test_transport',
79
81
                                            'TestTransport.SampleHandler')
80
82
        self.assertEqual([SampleHandler.__module__,
81
 
                            'bzrlib.transport.chroot',
82
 
                            'bzrlib.transport.pathfilter'],
 
83
                            'breezy.transport.chroot',
 
84
                            'breezy.transport.pathfilter'],
83
85
                            transport._get_transport_modules())
84
86
 
85
87
    def test_transport_dependency(self):
90
92
        transport._clear_protocol_handlers()
91
93
        transport.register_transport_proto('foo')
92
94
        transport.register_lazy_transport(
93
 
            'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
 
95
            'foo', 'breezy.tests.test_transport', 'BadTransportHandler')
94
96
        try:
95
97
            transport.get_transport_from_url('foo://fooserver/foo')
96
 
        except errors.UnsupportedProtocol, e:
 
98
        except errors.UnsupportedProtocol as e:
97
99
            e_str = str(e)
98
100
            self.assertEqual('Unsupported protocol'
99
101
                                ' for url "foo://fooserver/foo":'
109
111
        transport._clear_protocol_handlers()
110
112
        transport.register_transport_proto('foo')
111
113
        transport.register_lazy_transport(
112
 
            'foo', 'bzrlib.tests.test_transport', 'BackupTransportHandler')
 
114
            'foo', 'breezy.tests.test_transport', 'BackupTransportHandler')
113
115
        transport.register_lazy_transport(
114
 
            'foo', 'bzrlib.tests.test_transport', 'BadTransportHandler')
 
116
            'foo', 'breezy.tests.test_transport', 'BadTransportHandler')
115
117
        t = transport.get_transport_from_url('foo://fooserver/foo')
116
118
        self.assertTrue(isinstance(t, BackupTransportHandler))
117
119
 
119
121
        """Transport ssh:// should raise an error pointing out bzr+ssh://"""
120
122
        try:
121
123
            transport.get_transport_from_url('ssh://fooserver/foo')
122
 
        except errors.UnsupportedProtocol, e:
 
124
        except errors.UnsupportedProtocol as e:
123
125
            e_str = str(e)
124
126
            self.assertEqual('Unsupported protocol'
125
127
                              ' for url "ssh://fooserver/foo":'
134
136
        a_file = transport.LateReadError('a path')
135
137
        try:
136
138
            a_file.read()
137
 
        except errors.ReadError, error:
 
139
        except errors.ReadError as error:
138
140
            self.assertEqual('a path', error.path)
139
141
        self.assertRaises(errors.ReadError, a_file.read, 40)
140
142
        a_file.close()
272
274
        t = memory.MemoryTransport()
273
275
        t.append_bytes('path', 'content')
274
276
        self.assertEqual(t.get('path').read(), 'content')
275
 
        t.append_file('path', StringIO('content'))
 
277
        t.append_file('path', BytesIO(b'content'))
276
278
        self.assertEqual(t.get('path').read(), 'contentcontent')
277
279
 
278
280
    def test_put_and_get(self):
279
281
        t = memory.MemoryTransport()
280
 
        t.put_file('path', StringIO('content'))
 
282
        t.put_file('path', BytesIO(b'content'))
281
283
        self.assertEqual(t.get('path').read(), 'content')
282
284
        t.put_bytes('path', 'content')
283
285
        self.assertEqual(t.get('path').read(), 'content')
290
292
    def test_put_without_dir_fails(self):
291
293
        t = memory.MemoryTransport()
292
294
        self.assertRaises(errors.NoSuchFile,
293
 
                          t.put_file, 'dir/path', StringIO('content'))
 
295
                          t.put_file, 'dir/path', BytesIO(b'content'))
294
296
 
295
297
    def test_get_missing(self):
296
298
        transport = memory.MemoryTransport()
342
344
        t.put_bytes('dir/bar', 'content')
343
345
        t.put_bytes('bar', 'content')
344
346
        paths = set(t.iter_files_recursive())
345
 
        self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
 
347
        self.assertEqual({'dir/foo', 'dir/bar', 'bar'}, paths)
346
348
 
347
349
    def test_stat(self):
348
350
        t = memory.MemoryTransport()
561
563
        self.assertEqual(True, t.is_readonly())
562
564
 
563
565
    def test_http_parameters(self):
564
 
        from bzrlib.tests.http_server import HttpServer
 
566
        from breezy.tests.http_server import HttpServer
565
567
        # connect to '.' via http which is not listable
566
568
        server = HttpServer()
567
569
        self.start_server(server)
588
590
    def test_http_parameters(self):
589
591
        # the listable and is_readonly parameters
590
592
        # are not changed by the fakenfs decorator
591
 
        from bzrlib.tests.http_server import HttpServer
 
593
        from breezy.tests.http_server import HttpServer
592
594
        # connect to '.' via http which is not listable
593
595
        server = HttpServer()
594
596
        self.start_server(server)
622
624
 
623
625
    def get_vfat_transport(self, url):
624
626
        """Return vfat-backed transport for test directory"""
625
 
        from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
 
627
        from breezy.transport.fakevfat import FakeVFATTransportDecorator
626
628
        return FakeVFATTransportDecorator('vfat+' + url)
627
629
 
628
630
    def test_transport_creation(self):
629
 
        from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
 
631
        from breezy.transport.fakevfat import FakeVFATTransportDecorator
630
632
        t = self.get_vfat_transport('.')
631
633
        self.assertIsInstance(t, FakeVFATTransportDecorator)
632
634
 
657
659
 
658
660
    To verify a transport we need a server factory, which is a callable
659
661
    that accepts no parameters and returns an implementation of
660
 
    bzrlib.transport.Server.
 
662
    breezy.transport.Server.
661
663
 
662
664
    That Server is then used to construct transport instances and test
663
665
    the transport via loopback activity.
780
782
            raise e
781
783
        self.overrideAttr(os, 'chmod', fake_chmod)
782
784
        t.mkdir('test')
783
 
        t.mkdir('test2', mode=0707)
 
785
        t.mkdir('test2', mode=0o707)
784
786
        self.assertTrue(os.path.exists('test'))
785
787
        self.assertTrue(os.path.exists('test2'))
786
788
 
948
950
    def test_decorator(self):
949
951
        t = transport.get_transport_from_url('trace+memory://')
950
952
        self.assertIsInstance(
951
 
            t, bzrlib.transport.trace.TransportTraceDecorator)
 
953
            t, breezy.transport.trace.TransportTraceDecorator)
952
954
 
953
955
    def test_clone_preserves_activity(self):
954
956
        t = transport.get_transport_from_url('trace+memory://')
1002
1004
        # SFTPFullAbsoluteServer has a get_url method, and doesn't
1003
1005
        # override the interface (doesn't change self._vendor).
1004
1006
        # Note that this does encryption, so can be slow.
1005
 
        from bzrlib.tests import stub_sftp
 
1007
        from breezy.tests import stub_sftp
1006
1008
 
1007
1009
        # Start an SSH server
1008
1010
        self.command_executed = []
1057
1059
        port = ssh_server.port
1058
1060
 
1059
1061
        if sys.platform == 'win32':
1060
 
            bzr_remote_path = sys.executable + ' ' + self.get_bzr_path()
 
1062
            bzr_remote_path = sys.executable + ' ' + self.get_brz_path()
1061
1063
        else:
1062
 
            bzr_remote_path = self.get_bzr_path()
 
1064
            bzr_remote_path = self.get_brz_path()
1063
1065
        self.overrideEnv('BZR_REMOTE_PATH', bzr_remote_path)
1064
1066
 
1065
1067
        # Access the branch via a bzr+ssh URL.  The BZR_REMOTE_PATH environment