/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport_implementations.py

  • Committer: Robert Collins
  • Date: 2008-02-13 03:30:01 UTC
  • mfrom: (3221 +trunk)
  • mto: This revision was merged to the branch mainline in revision 3224.
  • Revision ID: robertc@robertcollins.net-20080213033001-rw70ul0zb02ph856
Merge to fix conflicts.

Show diffs side-by-side

added added

removed removed

Lines of Context:
20
20
TransportTestProviderAdapter.
21
21
"""
22
22
 
 
23
import itertools
23
24
import os
24
25
from cStringIO import StringIO
25
26
from StringIO import StringIO as pyStringIO
45
46
                           )
46
47
from bzrlib.osutils import getcwd
47
48
from bzrlib.smart import medium
48
 
from bzrlib.tests import TestCaseInTempDir, TestScenarioApplier, TestSkipped
 
49
from bzrlib.tests import (
 
50
    TestCaseInTempDir,
 
51
    TestScenarioApplier,
 
52
    TestSkipped,
 
53
    TestNotApplicable,
 
54
    )
49
55
from bzrlib.tests.test_transport import TestTransportImplementation
50
56
from bzrlib.transport import (
51
57
    ConnectedTransport,
181
187
        self.build_tree(files, transport=t, line_endings='binary')
182
188
        self.check_transport_contents('contents of a\n', t, 'a')
183
189
        content_f = t.get_multi(files)
184
 
        for content, f in zip(contents, content_f):
 
190
        # Use itertools.izip() instead of use zip() or map(), since they fully
 
191
        # evaluate their inputs, the transport requests should be issued and
 
192
        # handled sequentially (we don't want to force transport to buffer).
 
193
        for content, f in itertools.izip(contents, content_f):
185
194
            self.assertEqual(content, f.read())
186
195
 
187
196
        content_f = t.get_multi(iter(files))
188
 
        for content, f in zip(contents, content_f):
 
197
        # Use itertools.izip() for the same reason
 
198
        for content, f in itertools.izip(contents, content_f):
189
199
            self.assertEqual(content, f.read())
190
200
 
191
201
        self.assertRaises(NoSuchFile, t.get, 'c')
205
215
        except (errors.PathError, errors.RedirectRequested):
206
216
            # early failure return immediately.
207
217
            return
208
 
        # having got a file, read() must either work (i.e. http reading a dir listing) or
209
 
        # fail with ReadError
 
218
        # having got a file, read() must either work (i.e. http reading a dir
 
219
        # listing) or fail with ReadError
210
220
        try:
211
221
            a_file.read()
212
222
        except errors.ReadError:
900
910
        self.assertFalse(t.has('adir/bdir'))
901
911
        self.assertFalse(t.has('adir/bsubdir'))
902
912
 
 
913
    def test_rename_across_subdirs(self):
 
914
        t = self.get_transport()
 
915
        if t.is_readonly():
 
916
            raise TestNotApplicable("transport is readonly")
 
917
        t.mkdir('a')
 
918
        t.mkdir('b')
 
919
        ta = t.clone('a')
 
920
        tb = t.clone('b')
 
921
        ta.put_bytes('f', 'aoeu')
 
922
        ta.rename('f', '../b/f')
 
923
        self.assertTrue(tb.has('f'))
 
924
        self.assertFalse(ta.has('f'))
 
925
        self.assertTrue(t.has('b/f'))
 
926
 
903
927
    def test_delete_tree(self):
904
928
        t = self.get_transport()
905
929
 
959
983
        self.check_transport_contents('c this file\n', t, 'b')
960
984
 
961
985
        # TODO: Try to write a test for atomicity
962
 
        # TODO: Test moving into a non-existant subdirectory
 
986
        # TODO: Test moving into a non-existent subdirectory
963
987
        # TODO: Test Transport.move_multi
964
988
 
965
989
    def test_copy(self):
986
1010
    def test_connection_error(self):
987
1011
        """ConnectionError is raised when connection is impossible.
988
1012
        
989
 
        The error may be raised from either the constructor or the first
990
 
        operation on the transport.
 
1013
        The error should be raised from the first operation on the transport.
991
1014
        """
992
1015
        try:
993
1016
            url = self._server.get_bogus_url()
1263
1286
        self.assertEqual('', t.relpath(t.base))
1264
1287
        # base ends with /
1265
1288
        self.assertEqual('', t.relpath(t.base[:-1]))
1266
 
        # subdirs which dont exist should still give relpaths.
 
1289
        # subdirs which don't exist should still give relpaths.
1267
1290
        self.assertEqual('foo', t.relpath(t.base + 'foo'))
1268
1291
        # trailing slash should be the same.
1269
1292
        self.assertEqual('foo', t.relpath(t.base + 'foo/'))
1551
1574
                adjust_for_latency=True, upper_limit=content_size))
1552
1575
            self.assertEqual(1, len(result))
1553
1576
            data_len = len(result[0][1])
1554
 
            # minimmum length is from 400 to 1034 - 634
 
1577
            # minimum length is from 400 to 1034 - 634
1555
1578
            self.assertTrue(data_len >= 634)
1556
1579
            # must contain the region 400 to 1034
1557
1580
            self.assertTrue(result[0][0] <= 400)