← 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: Martin
  • Date: 2011-08-04 00:17:53 UTC
  • mto: This revision was merged to the branch mainline in revision 6055.
  • Revision ID: gzlist@googlemail.com-20110804001753-plgpwcpsxcum16yb
Make tests raising KnownFailure use the knownFailure method instead

Show diffs side-by-side

added added

removed removed

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