← 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: John Arbash Meinel
  • Date: 2011-05-11 11:35:28 UTC
  • mto: This revision was merged to the branch mainline in revision 5851.
  • Revision ID: john@arbash-meinel.com-20110511113528-qepibuwxicjrbb2h
Break compatibility with python <2.6.

This includes auditing the code for places where we were doing
explicit 'sys.version' checks and removing them as appropriate.

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