← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/per_transport.py

  • Committer: Andrew Bennetts
  • Date: 2010-10-13 00:26:41 UTC
  • mto: This revision was merged to the branch mainline in revision 5498.
  • Revision ID: andrew.bennetts@canonical.com-20101013002641-9tlh9k89mlj1666m
Keep docs-plain working.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/per_transport.py
1
 
# Copyright (C) 2005-2011 Canonical Ltd
 
1
# Copyright (C) 2005-2010 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
26
26
from StringIO import StringIO as pyStringIO
27
27
import stat
28
28
import sys
 
29
import unittest
29
30
 
30
31
from bzrlib import (
31
32
    errors,
32
33
    osutils,
33
 
    pyutils,
34
34
    tests,
35
 
    transport as _mod_transport,
36
35
    urlutils,
37
36
    )
38
37
from bzrlib.errors import (ConnectionError,
 
38
                           DirectoryNotEmpty,
39
39
                           FileExists,
40
40
                           InvalidURL,
 
41
                           LockError,
41
42
                           NoSuchFile,
 
43
                           NotLocalUrl,
42
44
                           PathError,
43
45
                           TransportNotPossible,
44
46
                           )
45
47
from bzrlib.osutils import getcwd
46
48
from bzrlib.smart import medium
47
49
from bzrlib.tests import (
 
50
    TestCaseInTempDir,
48
51
    TestSkipped,
49
52
    TestNotApplicable,
50
53
    multiply_tests,
53
56
from bzrlib.tests.test_transport import TestTransportImplementation
54
57
from bzrlib.transport import (
55
58
    ConnectedTransport,
 
59
    get_transport,
56
60
    _get_transport_modules,
57
61
    )
58
62
from bzrlib.transport.memory import MemoryTransport
74
78
    for module in _get_transport_modules():
75
79
        try:
76
80
            permutations = get_transport_test_permutations(
77
 
                pyutils.get_named_object(module))
 
81
                reduce(getattr, (module).split('.')[1:], __import__(module)))
78
82
            for (klass, server_factory) in permutations:
79
83
                scenario = ('%s,%s' % (klass.__name__, server_factory.__name__),
80
84
                    {"transport_class":klass,
98
102
 
99
103
    def setUp(self):
100
104
        super(TransportTests, self).setUp()
101
 
        self.overrideEnv('BZR_NO_SMART_VFS', None)
 
105
        self._captureVar('BZR_NO_SMART_VFS', None)
102
106
 
103
107
    def check_transport_contents(self, content, transport, relpath):
104
 
        """Check that transport.get_bytes(relpath) == content."""
105
 
        self.assertEqualDiff(content, transport.get_bytes(relpath))
 
108
        """Check that transport.get(relpath).read() == content."""
 
109
        self.assertEqualDiff(content, transport.get(relpath).read())
106
110
 
107
111
    def test_ensure_base_missing(self):
108
112
        """.ensure_base() should create the directory if it doesn't exist"""
207
211
                    ]
208
212
        self.build_tree(files, transport=t, line_endings='binary')
209
213
        self.assertRaises(NoSuchFile, t.get, 'c')
210
 
        def iterate_and_close(func, *args):
211
 
            for f in func(*args):
212
 
                # We call f.read() here because things like paramiko actually
213
 
                # spawn a thread to prefetch the content, which we want to
214
 
                # consume before we close the handle.
215
 
                content = f.read()
216
 
                f.close()
217
 
        self.assertRaises(NoSuchFile, iterate_and_close,
218
 
                          t.get_multi, ['a', 'b', 'c'])
219
 
        self.assertRaises(NoSuchFile, iterate_and_close,
220
 
                          t.get_multi, iter(['a', 'b', 'c']))
 
214
        self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])
 
215
        self.assertListRaises(NoSuchFile, t.get_multi, iter(['a', 'b', 'c']))
221
216
 
222
217
    def test_get_directory_read_gives_ReadError(self):
223
218
        """consistent errors for read() on a file returned by get()."""
265
260
        handle = t.open_write_stream('foo')
266
261
        try:
267
262
            handle.write('b')
268
 
            self.assertEqual('b', t.get_bytes('foo'))
 
263
            self.assertEqual('b', t.get('foo').read())
269
264
        finally:
270
265
            handle.close()
271
266
 
277
272
        try:
278
273
            handle.write('b')
279
274
            self.assertEqual('b', t.get_bytes('foo'))
280
 
            f = t.get('foo')
281
 
            try:
282
 
                self.assertEqual('b', f.read())
283
 
            finally:
284
 
                f.close()
 
275
            self.assertEqual('b', t.get('foo').read())
285
276
        finally:
286
277
            handle.close()
287
278
 
294
285
            return
295
286
 
296
287
        t.put_bytes('a', 'some text for a\n')
297
 
        self.assertTrue(t.has('a'))
 
288
        self.failUnless(t.has('a'))
298
289
        self.check_transport_contents('some text for a\n', t, 'a')
299
290
 
300
291
        # The contents should be overwritten
312
303
                    t.put_bytes_non_atomic, 'a', 'some text for a\n')
313
304
            return
314
305
 
315
 
        self.assertFalse(t.has('a'))
 
306
        self.failIf(t.has('a'))
316
307
        t.put_bytes_non_atomic('a', 'some text for a\n')
317
 
        self.assertTrue(t.has('a'))
 
308
        self.failUnless(t.has('a'))
318
309
        self.check_transport_contents('some text for a\n', t, 'a')
319
310
        # Put also replaces contents
320
311
        t.put_bytes_non_atomic('a', 'new\ncontents for\na\n')
332
323
        # Now test the create_parent flag
333
324
        self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'dir/a',
334
325
                                       'contents\n')
335
 
        self.assertFalse(t.has('dir/a'))
 
326
        self.failIf(t.has('dir/a'))
336
327
        t.put_bytes_non_atomic('dir/a', 'contents for dir/a\n',
337
328
                               create_parent_dir=True)
338
329
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
410
401
        result = t.put_file('a', StringIO('some text for a\n'))
411
402
        # put_file returns the length of the data written
412
403
        self.assertEqual(16, result)
413
 
        self.assertTrue(t.has('a'))
 
404
        self.failUnless(t.has('a'))
414
405
        self.check_transport_contents('some text for a\n', t, 'a')
415
406
        # Put also replaces contents
416
407
        result = t.put_file('a', StringIO('new\ncontents for\na\n'))
428
419
                    t.put_file_non_atomic, 'a', StringIO('some text for a\n'))
429
420
            return
430
421
 
431
 
        self.assertFalse(t.has('a'))
 
422
        self.failIf(t.has('a'))
432
423
        t.put_file_non_atomic('a', StringIO('some text for a\n'))
433
 
        self.assertTrue(t.has('a'))
 
424
        self.failUnless(t.has('a'))
434
425
        self.check_transport_contents('some text for a\n', t, 'a')
435
426
        # Put also replaces contents
436
427
        t.put_file_non_atomic('a', StringIO('new\ncontents for\na\n'))
448
439
        # Now test the create_parent flag
449
440
        self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'dir/a',
450
441
                                       StringIO('contents\n'))
451
 
        self.assertFalse(t.has('dir/a'))
 
442
        self.failIf(t.has('dir/a'))
452
443
        t.put_file_non_atomic('dir/a', StringIO('contents for dir/a\n'),
453
444
                              create_parent_dir=True)
454
445
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
653
644
            self.build_tree(files, transport=transport_from)
654
645
            self.assertEqual(4, transport_from.copy_to(files, transport_to))
655
646
            for f in files:
656
 
                self.check_transport_contents(transport_to.get_bytes(f),
 
647
                self.check_transport_contents(transport_to.get(f).read(),
657
648
                                              transport_from, f)
658
649
 
659
650
        t = self.get_transport()
682
673
        files = ['a', 'b', 'c', 'd']
683
674
        t.copy_to(iter(files), temp_transport)
684
675
        for f in files:
685
 
            self.check_transport_contents(temp_transport.get_bytes(f),
 
676
            self.check_transport_contents(temp_transport.get(f).read(),
686
677
                                          t, f)
687
678
        del temp_transport
688
679
 
831
822
            return
832
823
 
833
824
        t.put_bytes('a', 'a little bit of text\n')
834
 
        self.assertTrue(t.has('a'))
 
825
        self.failUnless(t.has('a'))
835
826
        t.delete('a')
836
 
        self.assertFalse(t.has('a'))
 
827
        self.failIf(t.has('a'))
837
828
 
838
829
        self.assertRaises(NoSuchFile, t.delete, 'a')
839
830
 
845
836
        t.delete_multi(['a', 'c'])
846
837
        self.assertEqual([False, True, False],
847
838
                list(t.has_multi(['a', 'b', 'c'])))
848
 
        self.assertFalse(t.has('a'))
849
 
        self.assertTrue(t.has('b'))
850
 
        self.assertFalse(t.has('c'))
 
839
        self.failIf(t.has('a'))
 
840
        self.failUnless(t.has('b'))
 
841
        self.failIf(t.has('c'))
851
842
 
852
843
        self.assertRaises(NoSuchFile,
853
844
                t.delete_multi, ['a', 'b', 'c'])
914
905
        t.mkdir('foo-baz')
915
906
        t.rmdir('foo')
916
907
        self.assertRaises((NoSuchFile, PathError), t.rmdir, 'foo')
917
 
        self.assertTrue(t.has('foo-bar'))
 
908
        self.failUnless(t.has('foo-bar'))
918
909
 
919
910
    def test_rename_dir_succeeds(self):
920
911
        t = self.get_transport()
1003
994
        self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
1004
995
 
1005
996
        t.move('a', 'b')
1006
 
        self.assertTrue(t.has('b'))
1007
 
        self.assertFalse(t.has('a'))
 
997
        self.failUnless(t.has('b'))
 
998
        self.failIf(t.has('a'))
1008
999
 
1009
1000
        self.check_transport_contents('a first file\n', t, 'b')
1010
1001
        self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
1012
1003
        # Overwrite a file
1013
1004
        t.put_bytes('c', 'c this file\n')
1014
1005
        t.move('c', 'b')
1015
 
        self.assertFalse(t.has('c'))
 
1006
        self.failIf(t.has('c'))
1016
1007
        self.check_transport_contents('c this file\n', t, 'b')
1017
1008
 
1018
1009
        # TODO: Try to write a test for atomicity
1050
1041
        except NotImplementedError:
1051
1042
            raise TestSkipped("Transport %s has no bogus URL support." %
1052
1043
                              self._server.__class__)
1053
 
        t = _mod_transport.get_transport(url)
 
1044
        t = get_transport(url)
1054
1045
        self.assertRaises((ConnectionError, NoSuchFile), t.get, '.bzr/branch')
1055
1046
 
1056
1047
    def test_stat(self):
1072
1063
        for path, size in zip(paths, sizes):
1073
1064
            st = t.stat(path)
1074
1065
            if path.endswith('/'):
1075
 
                self.assertTrue(S_ISDIR(st.st_mode))
 
1066
                self.failUnless(S_ISDIR(st.st_mode))
1076
1067
                # directory sizes are meaningless
1077
1068
            else:
1078
 
                self.assertTrue(S_ISREG(st.st_mode))
 
1069
                self.failUnless(S_ISREG(st.st_mode))
1079
1070
                self.assertEqual(size, st.st_size)
1080
1071
 
1081
1072
        remote_stats = list(t.stat_multi(paths))
1088
1079
        self.assertListRaises(NoSuchFile, t.stat_multi, iter(['a', 'c', 'd']))
1089
1080
        self.build_tree(['subdir/', 'subdir/file'], transport=t)
1090
1081
        subdir = t.clone('subdir')
1091
 
        st = subdir.stat('./file')
1092
 
        st = subdir.stat('.')
 
1082
        subdir.stat('./file')
 
1083
        subdir.stat('.')
1093
1084
 
1094
1085
    def test_hardlink(self):
1095
1086
        from stat import ST_NLINK
1104
1095
        try:
1105
1096
            t.hardlink(source_name, link_name)
1106
1097
 
1107
 
            self.assertTrue(t.has(source_name))
1108
 
            self.assertTrue(t.has(link_name))
 
1098
            self.failUnless(t.has(source_name))
 
1099
            self.failUnless(t.has(link_name))
1109
1100
 
1110
1101
            st = t.stat(link_name)
1111
 
            self.assertEqual(st[ST_NLINK], 2)
 
1102
            self.failUnlessEqual(st[ST_NLINK], 2)
1112
1103
        except TransportNotPossible:
1113
1104
            raise TestSkipped("Transport %s does not support hardlinks." %
1114
1105
                              self._server.__class__)
1126
1117
        try:
1127
1118
            t.symlink(source_name, link_name)
1128
1119
 
1129
 
            self.assertTrue(t.has(source_name))
1130
 
            self.assertTrue(t.has(link_name))
 
1120
            self.failUnless(t.has(source_name))
 
1121
            self.failUnless(t.has(link_name))
1131
1122
 
1132
1123
            st = t.stat(link_name)
1133
 
            self.assertTrue(S_ISLNK(st.st_mode),
 
1124
            self.failUnless(S_ISLNK(st.st_mode),
1134
1125
                "expected symlink, got mode %o" % st.st_mode)
1135
1126
        except TransportNotPossible:
1136
1127
            raise TestSkipped("Transport %s does not support symlinks." %
1302
1293
 
1303
1294
        self.build_tree(['a', 'b/', 'b/c'], transport=t1)
1304
1295
 
1305
 
        self.assertTrue(t1.has('a'))
1306
 
        self.assertTrue(t1.has('b/c'))
1307
 
        self.assertFalse(t1.has('c'))
 
1296
        self.failUnless(t1.has('a'))
 
1297
        self.failUnless(t1.has('b/c'))
 
1298
        self.failIf(t1.has('c'))
1308
1299
 
1309
1300
        t2 = t1.clone('b')
1310
1301
        self.assertEqual(t1.base + 'b/', t2.base)
1311
1302
 
1312
 
        self.assertTrue(t2.has('c'))
1313
 
        self.assertFalse(t2.has('a'))
 
1303
        self.failUnless(t2.has('c'))
 
1304
        self.failIf(t2.has('a'))
1314
1305
 
1315
1306
        t3 = t2.clone('..')
1316
 
        self.assertTrue(t3.has('a'))
1317
 
        self.assertFalse(t3.has('c'))
 
1307
        self.failUnless(t3.has('a'))
 
1308
        self.failIf(t3.has('c'))
1318
1309
 
1319
 
        self.assertFalse(t1.has('b/d'))
1320
 
        self.assertFalse(t2.has('d'))
1321
 
        self.assertFalse(t3.has('b/d'))
 
1310
        self.failIf(t1.has('b/d'))
 
1311
        self.failIf(t2.has('d'))
 
1312
        self.failIf(t3.has('b/d'))
1322
1313
 
1323
1314
        if t1.is_readonly():
1324
1315
            self.build_tree_contents([('b/d', 'newfile\n')])
1325
1316
        else:
1326
1317
            t2.put_bytes('d', 'newfile\n')
1327
1318
 
1328
 
        self.assertTrue(t1.has('b/d'))
1329
 
        self.assertTrue(t2.has('d'))
1330
 
        self.assertTrue(t3.has('b/d'))
 
1319
        self.failUnless(t1.has('b/d'))
 
1320
        self.failUnless(t2.has('d'))
 
1321
        self.failUnless(t3.has('b/d'))
1331
1322
 
1332
1323
    def test_clone_to_root(self):
1333
1324
        orig_transport = self.get_transport()
1407
1398
        self.assertEqual(transport.clone("/").abspath('foo'),
1408
1399
                         transport.abspath("/foo"))
1409
1400
 
1410
 
    # GZ 2011-01-26: Test in per_transport but not using self.get_transport?
1411
1401
    def test_win32_abspath(self):
1412
1402
        # Note: we tried to set sys.platform='win32' so we could test on
1413
1403
        # other platforms too, but then osutils does platform specific
1418
1408
 
1419
1409
        # smoke test for abspath on win32.
1420
1410
        # a transport based on 'file:///' never fully qualifies the drive.
1421
 
        transport = _mod_transport.get_transport("file:///")
1422
 
        self.assertEqual(transport.abspath("/"), "file:///")
 
1411
        transport = get_transport("file:///")
 
1412
        self.failUnlessEqual(transport.abspath("/"), "file:///")
1423
1413
 
1424
1414
        # but a transport that starts with a drive spec must keep it.
1425
 
        transport = _mod_transport.get_transport("file:///C:/")
1426
 
        self.assertEqual(transport.abspath("/"), "file:///C:/")
 
1415
        transport = get_transport("file:///C:/")
 
1416
        self.failUnlessEqual(transport.abspath("/"), "file:///C:/")
1427
1417
 
1428
1418
    def test_local_abspath(self):
1429
1419
        transport = self.get_transport()
1626
1616
    def test_readv(self):
1627
1617
        transport = self.get_transport()
1628
1618
        if transport.is_readonly():
1629
 
            with file('a', 'w') as f: f.write('0123456789')
 
1619
            file('a', 'w').write('0123456789')
1630
1620
        else:
1631
1621
            transport.put_bytes('a', '0123456789')
1632
1622
 
1642
1632
    def test_readv_out_of_order(self):
1643
1633
        transport = self.get_transport()
1644
1634
        if transport.is_readonly():
1645
 
            with file('a', 'w') as f: f.write('0123456789')
 
1635
            file('a', 'w').write('0123456789')
1646
1636
        else:
1647
1637
            transport.put_bytes('a', '01234567890')
1648
1638
 
1720
1710
        transport = self.get_transport()
1721
1711
        # test from observed failure case.
1722
1712
        if transport.is_readonly():
1723
 
            with file('a', 'w') as f: f.write('a'*1024*1024)
 
1713
            file('a', 'w').write('a'*1024*1024)
1724
1714
        else:
1725
1715
            transport.put_bytes('a', 'a'*1024*1024)
1726
1716
        broken_vector = [(465219, 800), (225221, 800), (445548, 800),
1760
1750
    def test_readv_short_read(self):
1761
1751
        transport = self.get_transport()
1762
1752
        if transport.is_readonly():
1763
 
            with file('a', 'w') as f: f.write('0123456789')
 
1753
            file('a', 'w').write('0123456789')
1764
1754
        else:
1765
1755
            transport.put_bytes('a', '01234567890')
1766
1756