← 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: 2011-01-21 22:45:45 UTC
  • mfrom: (5628.1.1 locale-failure)
  • Revision ID: pqm@pqm.ubuntu.com-20110121224545-j7ie4ubrplzw0tvv
(jelmer) Fix test_locale test on systems where the default locale is not C.
 (Jelmer Vernooij)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/per_transport.py
32
32
    osutils,
33
33
    pyutils,
34
34
    tests,
35
 
    transport as _mod_transport,
36
35
    urlutils,
37
36
    )
38
37
from bzrlib.errors import (ConnectionError,
53
52
from bzrlib.tests.test_transport import TestTransportImplementation
54
53
from bzrlib.transport import (
55
54
    ConnectedTransport,
 
55
    get_transport,
56
56
    _get_transport_modules,
57
57
    )
58
58
from bzrlib.transport.memory import MemoryTransport
207
207
                    ]
208
208
        self.build_tree(files, transport=t, line_endings='binary')
209
209
        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']))
 
210
        self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])
 
211
        self.assertListRaises(NoSuchFile, t.get_multi, iter(['a', 'b', 'c']))
221
212
 
222
213
    def test_get_directory_read_gives_ReadError(self):
223
214
        """consistent errors for read() on a file returned by get()."""
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')
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