← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

merge bzr.dev

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
28
28
                           UnsupportedProtocol,
29
29
                           )
30
30
from bzrlib.tests import TestCase, TestCaseInTempDir
31
 
from bzrlib.transport import (_get_protocol_handlers,
 
31
from bzrlib.transport import (_CoalescedOffset,
 
32
                              _get_protocol_handlers,
32
33
                              _get_transport_modules,
33
34
                              get_transport,
34
35
                              register_lazy_transport,
98
99
            self.assertTrue(isinstance(t, BackupTransportHandler))
99
100
        finally:
100
101
            _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)])
101
122
            
 
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
 
102
162
 
103
163
class TestMemoryTransport(TestCase):
104
164
 
311
371
class BackupTransportHandler(Transport):
312
372
    """Test transport that works as a backup for the BadTransportHandler"""
313
373
    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