← 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: Canonical.com Patch Queue Manager
  • Date: 2010-04-06 06:59:03 UTC
  • mfrom: (5051.5.1 subunit)
  • Revision ID: pqm@pqm.ubuntu.com-20100406065903-y9dxgwmog1pmw7dz
Use subunit when running tests in PQM.

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()."""
256
251
 
257
252
    def test_get_bytes_unknown_file(self):
258
253
        t = self.get_transport()
 
254
 
259
255
        self.assertRaises(NoSuchFile, t.get_bytes, 'c')
260
256
 
261
257
    def test_get_with_open_write_stream_sees_all_content(self):
265
261
        handle = t.open_write_stream('foo')
266
262
        try:
267
263
            handle.write('b')
268
 
            self.assertEqual('b', t.get_bytes('foo'))
 
264
            self.assertEqual('b', t.get('foo').read())
269
265
        finally:
270
266
            handle.close()
271
267
 
277
273
        try:
278
274
            handle.write('b')
279
275
            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()
 
276
            self.assertEqual('b', t.get('foo').read())
285
277
        finally:
286
278
            handle.close()
287
279
 
294
286
            return
295
287
 
296
288
        t.put_bytes('a', 'some text for a\n')
297
 
        self.assertTrue(t.has('a'))
 
289
        self.failUnless(t.has('a'))
298
290
        self.check_transport_contents('some text for a\n', t, 'a')
299
291
 
300
292
        # The contents should be overwritten
312
304
                    t.put_bytes_non_atomic, 'a', 'some text for a\n')
313
305
            return
314
306
 
315
 
        self.assertFalse(t.has('a'))
 
307
        self.failIf(t.has('a'))
316
308
        t.put_bytes_non_atomic('a', 'some text for a\n')
317
 
        self.assertTrue(t.has('a'))
 
309
        self.failUnless(t.has('a'))
318
310
        self.check_transport_contents('some text for a\n', t, 'a')
319
311
        # Put also replaces contents
320
312
        t.put_bytes_non_atomic('a', 'new\ncontents for\na\n')
332
324
        # Now test the create_parent flag
333
325
        self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'dir/a',
334
326
                                       'contents\n')
335
 
        self.assertFalse(t.has('dir/a'))
 
327
        self.failIf(t.has('dir/a'))
336
328
        t.put_bytes_non_atomic('dir/a', 'contents for dir/a\n',
337
329
                               create_parent_dir=True)
338
330
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
410
402
        result = t.put_file('a', StringIO('some text for a\n'))
411
403
        # put_file returns the length of the data written
412
404
        self.assertEqual(16, result)
413
 
        self.assertTrue(t.has('a'))
 
405
        self.failUnless(t.has('a'))
414
406
        self.check_transport_contents('some text for a\n', t, 'a')
415
407
        # Put also replaces contents
416
408
        result = t.put_file('a', StringIO('new\ncontents for\na\n'))
428
420
                    t.put_file_non_atomic, 'a', StringIO('some text for a\n'))
429
421
            return
430
422
 
431
 
        self.assertFalse(t.has('a'))
 
423
        self.failIf(t.has('a'))
432
424
        t.put_file_non_atomic('a', StringIO('some text for a\n'))
433
 
        self.assertTrue(t.has('a'))
 
425
        self.failUnless(t.has('a'))
434
426
        self.check_transport_contents('some text for a\n', t, 'a')
435
427
        # Put also replaces contents
436
428
        t.put_file_non_atomic('a', StringIO('new\ncontents for\na\n'))
448
440
        # Now test the create_parent flag
449
441
        self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'dir/a',
450
442
                                       StringIO('contents\n'))
451
 
        self.assertFalse(t.has('dir/a'))
 
443
        self.failIf(t.has('dir/a'))
452
444
        t.put_file_non_atomic('dir/a', StringIO('contents for dir/a\n'),
453
445
                              create_parent_dir=True)
454
446
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
653
645
            self.build_tree(files, transport=transport_from)
654
646
            self.assertEqual(4, transport_from.copy_to(files, transport_to))
655
647
            for f in files:
656
 
                self.check_transport_contents(transport_to.get_bytes(f),
 
648
                self.check_transport_contents(transport_to.get(f).read(),
657
649
                                              transport_from, f)
658
650
 
659
651
        t = self.get_transport()
682
674
        files = ['a', 'b', 'c', 'd']
683
675
        t.copy_to(iter(files), temp_transport)
684
676
        for f in files:
685
 
            self.check_transport_contents(temp_transport.get_bytes(f),
 
677
            self.check_transport_contents(temp_transport.get(f).read(),
686
678
                                          t, f)
687
679
        del temp_transport
688
680
 
831
823
            return
832
824
 
833
825
        t.put_bytes('a', 'a little bit of text\n')
834
 
        self.assertTrue(t.has('a'))
 
826
        self.failUnless(t.has('a'))
835
827
        t.delete('a')
836
 
        self.assertFalse(t.has('a'))
 
828
        self.failIf(t.has('a'))
837
829
 
838
830
        self.assertRaises(NoSuchFile, t.delete, 'a')
839
831
 
845
837
        t.delete_multi(['a', 'c'])
846
838
        self.assertEqual([False, True, False],
847
839
                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'))
 
840
        self.failIf(t.has('a'))
 
841
        self.failUnless(t.has('b'))
 
842
        self.failIf(t.has('c'))
851
843
 
852
844
        self.assertRaises(NoSuchFile,
853
845
                t.delete_multi, ['a', 'b', 'c'])
914
906
        t.mkdir('foo-baz')
915
907
        t.rmdir('foo')
916
908
        self.assertRaises((NoSuchFile, PathError), t.rmdir, 'foo')
917
 
        self.assertTrue(t.has('foo-bar'))
 
909
        self.failUnless(t.has('foo-bar'))
918
910
 
919
911
    def test_rename_dir_succeeds(self):
920
912
        t = self.get_transport()
1003
995
        self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
1004
996
 
1005
997
        t.move('a', 'b')
1006
 
        self.assertTrue(t.has('b'))
1007
 
        self.assertFalse(t.has('a'))
 
998
        self.failUnless(t.has('b'))
 
999
        self.failIf(t.has('a'))
1008
1000
 
1009
1001
        self.check_transport_contents('a first file\n', t, 'b')
1010
1002
        self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
1012
1004
        # Overwrite a file
1013
1005
        t.put_bytes('c', 'c this file\n')
1014
1006
        t.move('c', 'b')
1015
 
        self.assertFalse(t.has('c'))
 
1007
        self.failIf(t.has('c'))
1016
1008
        self.check_transport_contents('c this file\n', t, 'b')
1017
1009
 
1018
1010
        # TODO: Try to write a test for atomicity
1050
1042
        except NotImplementedError:
1051
1043
            raise TestSkipped("Transport %s has no bogus URL support." %
1052
1044
                              self._server.__class__)
1053
 
        t = _mod_transport.get_transport(url)
 
1045
        t = get_transport(url)
1054
1046
        self.assertRaises((ConnectionError, NoSuchFile), t.get, '.bzr/branch')
1055
1047
 
1056
1048
    def test_stat(self):
1072
1064
        for path, size in zip(paths, sizes):
1073
1065
            st = t.stat(path)
1074
1066
            if path.endswith('/'):
1075
 
                self.assertTrue(S_ISDIR(st.st_mode))
 
1067
                self.failUnless(S_ISDIR(st.st_mode))
1076
1068
                # directory sizes are meaningless
1077
1069
            else:
1078
 
                self.assertTrue(S_ISREG(st.st_mode))
 
1070
                self.failUnless(S_ISREG(st.st_mode))
1079
1071
                self.assertEqual(size, st.st_size)
1080
1072
 
1081
1073
        remote_stats = list(t.stat_multi(paths))
1088
1080
        self.assertListRaises(NoSuchFile, t.stat_multi, iter(['a', 'c', 'd']))
1089
1081
        self.build_tree(['subdir/', 'subdir/file'], transport=t)
1090
1082
        subdir = t.clone('subdir')
1091
 
        st = subdir.stat('./file')
1092
 
        st = subdir.stat('.')
 
1083
        subdir.stat('./file')
 
1084
        subdir.stat('.')
1093
1085
 
1094
1086
    def test_hardlink(self):
1095
1087
        from stat import ST_NLINK
1104
1096
        try:
1105
1097
            t.hardlink(source_name, link_name)
1106
1098
 
1107
 
            self.assertTrue(t.has(source_name))
1108
 
            self.assertTrue(t.has(link_name))
 
1099
            self.failUnless(t.has(source_name))
 
1100
            self.failUnless(t.has(link_name))
1109
1101
 
1110
1102
            st = t.stat(link_name)
1111
 
            self.assertEqual(st[ST_NLINK], 2)
 
1103
            self.failUnlessEqual(st[ST_NLINK], 2)
1112
1104
        except TransportNotPossible:
1113
1105
            raise TestSkipped("Transport %s does not support hardlinks." %
1114
1106
                              self._server.__class__)
1126
1118
        try:
1127
1119
            t.symlink(source_name, link_name)
1128
1120
 
1129
 
            self.assertTrue(t.has(source_name))
1130
 
            self.assertTrue(t.has(link_name))
 
1121
            self.failUnless(t.has(source_name))
 
1122
            self.failUnless(t.has(link_name))
1131
1123
 
1132
1124
            st = t.stat(link_name)
1133
 
            self.assertTrue(S_ISLNK(st.st_mode),
1134
 
                "expected symlink, got mode %o" % st.st_mode)
 
1125
            self.failUnless(S_ISLNK(st.st_mode))
1135
1126
        except TransportNotPossible:
1136
1127
            raise TestSkipped("Transport %s does not support symlinks." %
1137
1128
                              self._server.__class__)
1274
1265
        self.assertIs(t._get_connection(), c._get_connection())
1275
1266
 
1276
1267
        # Temporary failure, we need to create a new dummy connection
1277
 
        new_connection = None
 
1268
        new_connection = object()
1278
1269
        t._set_connection(new_connection)
1279
1270
        # Check that both transports use the same connection
1280
1271
        self.assertIs(new_connection, t._get_connection())
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
 
1775
1765
        # also raise a special error
1776
1766
        self.assertListRaises((errors.ShortReadvError, errors.InvalidRange),
1777
1767
                              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))