← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

Late bind to PatienceSequenceMatcher to allow plugin to override.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
1
1
# Copyright (C) 2004, 2005, 2006 by 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
5
5
# the Free Software Foundation; either version 2 of the License, or
6
6
# (at your option) any later version.
7
 
#
 
7
 
8
8
# This program is distributed in the hope that it will be useful,
9
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
11
# GNU General Public License for more details.
12
 
#
 
12
 
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
25
25
                           TransportNotPossible,
26
26
                           ConnectionError,
27
27
                           DependencyNotPresent,
28
 
                           UnsupportedProtocol,
29
28
                           )
30
29
from bzrlib.tests import TestCase, TestCaseInTempDir
31
 
from bzrlib.transport import (_CoalescedOffset,
32
 
                              _get_protocol_handlers,
 
30
from bzrlib.transport import (_get_protocol_handlers,
33
31
                              _get_transport_modules,
34
32
                              get_transport,
35
33
                              register_lazy_transport,
36
34
                              _set_protocol_handlers,
 
35
                              urlescape,
37
36
                              Transport,
38
37
                              )
39
38
from bzrlib.transport.memory import MemoryTransport
43
42
class TestTransport(TestCase):
44
43
    """Test the non transport-concrete class functionality."""
45
44
 
 
45
    def test_urlescape(self):
 
46
        self.assertEqual('%25', urlescape('%'))
 
47
 
46
48
    def test__get_set_protocol_handlers(self):
47
49
        handlers = _get_protocol_handlers()
48
50
        self.assertNotEqual({}, handlers)
72
74
        try:
73
75
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
74
76
                    'BadTransportHandler')
75
 
            try:
76
 
                get_transport('foo://fooserver/foo')
77
 
            except UnsupportedProtocol, e:
78
 
                e_str = str(e)
79
 
                self.assertEquals('Unsupported protocol'
80
 
                                  ' for url "foo://fooserver/foo":'
81
 
                                  ' Unable to import library "some_lib":'
82
 
                                  ' testing missing dependency', str(e))
83
 
            else:
84
 
                self.fail('Did not raise UnsupportedProtocol')
 
77
            t = get_transport('foo://fooserver/foo')
 
78
            # because we failed to load the transport
 
79
            self.assertTrue(isinstance(t, LocalTransport))
85
80
        finally:
86
81
            # restore original values
87
82
            _set_protocol_handlers(saved_handlers)
99
94
            self.assertTrue(isinstance(t, BackupTransportHandler))
100
95
        finally:
101
96
            _set_protocol_handlers(saved_handlers)
102
 
 
103
 
 
104
 
class TestCoalesceOffsets(TestCase):
105
 
    
106
 
    def check(self, expected, offsets, limit=0, fudge=0):
107
 
        coalesce = Transport._coalesce_offsets
108
 
        exp = [_CoalescedOffset(*x) for x in expected]
109
 
        out = list(coalesce(offsets, limit=limit, fudge_factor=fudge))
110
 
        self.assertEqual(exp, out)
111
 
 
112
 
    def test_coalesce_empty(self):
113
 
        self.check([], [])
114
 
 
115
 
    def test_coalesce_simple(self):
116
 
        self.check([(0, 10, [(0, 10)])], [(0, 10)])
117
 
 
118
 
    def test_coalesce_unrelated(self):
119
 
        self.check([(0, 10, [(0, 10)]),
120
 
                    (20, 10, [(0, 10)]),
121
 
                   ], [(0, 10), (20, 10)])
122
97
            
123
 
    def test_coalesce_unsorted(self):
124
 
        self.check([(20, 10, [(0, 10)]),
125
 
                    (0, 10, [(0, 10)]),
126
 
                   ], [(20, 10), (0, 10)])
127
 
 
128
 
    def test_coalesce_nearby(self):
129
 
        self.check([(0, 20, [(0, 10), (10, 10)])],
130
 
                   [(0, 10), (10, 10)])
131
 
 
132
 
    def test_coalesce_overlapped(self):
133
 
        self.check([(0, 15, [(0, 10), (5, 10)])],
134
 
                   [(0, 10), (5, 10)])
135
 
 
136
 
    def test_coalesce_limit(self):
137
 
        self.check([(10, 50, [(0, 10), (10, 10), (20, 10),
138
 
                              (30, 10), (40, 10)]),
139
 
                    (60, 50, [(0, 10), (10, 10), (20, 10),
140
 
                              (30, 10), (40, 10)]),
141
 
                   ], [(10, 10), (20, 10), (30, 10), (40, 10),
142
 
                       (50, 10), (60, 10), (70, 10), (80, 10),
143
 
                       (90, 10), (100, 10)],
144
 
                    limit=5)
145
 
 
146
 
    def test_coalesce_no_limit(self):
147
 
        self.check([(10, 100, [(0, 10), (10, 10), (20, 10),
148
 
                               (30, 10), (40, 10), (50, 10),
149
 
                               (60, 10), (70, 10), (80, 10),
150
 
                               (90, 10)]),
151
 
                   ], [(10, 10), (20, 10), (30, 10), (40, 10),
152
 
                       (50, 10), (60, 10), (70, 10), (80, 10),
153
 
                       (90, 10), (100, 10)])
154
 
 
155
 
    def test_coalesce_fudge(self):
156
 
        self.check([(10, 30, [(0, 10), (20, 10)]),
157
 
                    (100, 10, [(0, 10),]),
158
 
                   ], [(10, 10), (30, 10), (100, 10)],
159
 
                   fudge=10
160
 
                  )
161
 
 
162
98
 
163
99
class TestMemoryTransport(TestCase):
164
100
 
171
107
 
172
108
    def test_abspath(self):
173
109
        transport = MemoryTransport()
174
 
        self.assertEqual("memory:///relpath", transport.abspath('relpath'))
 
110
        self.assertEqual("memory:/relpath", transport.abspath('relpath'))
175
111
 
176
112
    def test_relpath(self):
177
113
        transport = MemoryTransport()
371
307
class BackupTransportHandler(Transport):
372
308
    """Test transport that works as a backup for the BadTransportHandler"""
373
309
    pass
374
 
 
375
 
 
376
 
class TestTransportImplementation(TestCaseInTempDir):
377
 
    """Implementation verification for transports.
378
 
    
379
 
    To verify a transport we need a server factory, which is a callable
380
 
    that accepts no parameters and returns an implementation of
381
 
    bzrlib.transport.Server.
382
 
    
383
 
    That Server is then used to construct transport instances and test
384
 
    the transport via loopback activity.
385
 
 
386
 
    Currently this assumes that the Transport object is connected to the 
387
 
    current working directory.  So that whatever is done 
388
 
    through the transport, should show up in the working 
389
 
    directory, and vice-versa. This is a bug, because its possible to have
390
 
    URL schemes which provide access to something that may not be 
391
 
    result in storage on the local disk, i.e. due to file system limits, or 
392
 
    due to it being a database or some other non-filesystem tool.
393
 
 
394
 
    This also tests to make sure that the functions work with both
395
 
    generators and lists (assuming iter(list) is effectively a generator)
396
 
    """
397
 
    
398
 
    def setUp(self):
399
 
        super(TestTransportImplementation, self).setUp()
400
 
        self._server = self.transport_server()
401
 
        self._server.setUp()
402
 
 
403
 
    def tearDown(self):
404
 
        super(TestTransportImplementation, self).tearDown()
405
 
        self._server.tearDown()
406
 
        
407
 
    def get_transport(self):
408
 
        """Return a connected transport to the local directory."""
409
 
        base_url = self._server.get_url()
410
 
        # try getting the transport via the regular interface:
411
 
        t = get_transport(base_url)
412
 
        if not isinstance(t, self.transport_class): 
413
 
            # we did not get the correct transport class type. Override the
414
 
            # regular connection behaviour by direct construction.
415
 
            t = self.transport_class(base_url)
416
 
        return t