← 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-04-18 04:55:00 UTC
  • mfrom: (5784.2.1 754188-apport-test)
  • Revision ID: pqm@pqm.ubuntu.com-20110418045500-ce6lkgyiq7f47q43
(mbp) Rewrite test_report_bug_legacy away from using doctest (see bug
 764188) (Martin Pool)

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
32
32
    osutils,
33
33
    pyutils,
34
34
    tests,
 
35
    transport as _mod_transport,
35
36
    urlutils,
36
37
    )
37
38
from bzrlib.errors import (ConnectionError,
52
53
from bzrlib.tests.test_transport import TestTransportImplementation
53
54
from bzrlib.transport import (
54
55
    ConnectedTransport,
55
 
    get_transport,
56
56
    _get_transport_modules,
57
57
    )
58
58
from bzrlib.transport.memory import MemoryTransport
98
98
 
99
99
    def setUp(self):
100
100
        super(TransportTests, self).setUp()
101
 
        self._captureVar('BZR_NO_SMART_VFS', None)
 
101
        self.overrideEnv('BZR_NO_SMART_VFS', None)
102
102
 
103
103
    def check_transport_contents(self, content, transport, relpath):
104
 
        """Check that transport.get(relpath).read() == content."""
105
 
        self.assertEqualDiff(content, transport.get(relpath).read())
 
104
        """Check that transport.get_bytes(relpath) == content."""
 
105
        self.assertEqualDiff(content, transport.get_bytes(relpath))
106
106
 
107
107
    def test_ensure_base_missing(self):
108
108
        """.ensure_base() should create the directory if it doesn't exist"""
256
256
        handle = t.open_write_stream('foo')
257
257
        try:
258
258
            handle.write('b')
259
 
            self.assertEqual('b', t.get('foo').read())
 
259
            self.assertEqual('b', t.get_bytes('foo'))
260
260
        finally:
261
261
            handle.close()
262
262
 
268
268
        try:
269
269
            handle.write('b')
270
270
            self.assertEqual('b', t.get_bytes('foo'))
271
 
            self.assertEqual('b', t.get('foo').read())
 
271
            f = t.get('foo')
 
272
            try:
 
273
                self.assertEqual('b', f.read())
 
274
            finally:
 
275
                f.close()
272
276
        finally:
273
277
            handle.close()
274
278
 
281
285
            return
282
286
 
283
287
        t.put_bytes('a', 'some text for a\n')
284
 
        self.failUnless(t.has('a'))
 
288
        self.assertTrue(t.has('a'))
285
289
        self.check_transport_contents('some text for a\n', t, 'a')
286
290
 
287
291
        # The contents should be overwritten
299
303
                    t.put_bytes_non_atomic, 'a', 'some text for a\n')
300
304
            return
301
305
 
302
 
        self.failIf(t.has('a'))
 
306
        self.assertFalse(t.has('a'))
303
307
        t.put_bytes_non_atomic('a', 'some text for a\n')
304
 
        self.failUnless(t.has('a'))
 
308
        self.assertTrue(t.has('a'))
305
309
        self.check_transport_contents('some text for a\n', t, 'a')
306
310
        # Put also replaces contents
307
311
        t.put_bytes_non_atomic('a', 'new\ncontents for\na\n')
319
323
        # Now test the create_parent flag
320
324
        self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'dir/a',
321
325
                                       'contents\n')
322
 
        self.failIf(t.has('dir/a'))
 
326
        self.assertFalse(t.has('dir/a'))
323
327
        t.put_bytes_non_atomic('dir/a', 'contents for dir/a\n',
324
328
                               create_parent_dir=True)
325
329
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
397
401
        result = t.put_file('a', StringIO('some text for a\n'))
398
402
        # put_file returns the length of the data written
399
403
        self.assertEqual(16, result)
400
 
        self.failUnless(t.has('a'))
 
404
        self.assertTrue(t.has('a'))
401
405
        self.check_transport_contents('some text for a\n', t, 'a')
402
406
        # Put also replaces contents
403
407
        result = t.put_file('a', StringIO('new\ncontents for\na\n'))
415
419
                    t.put_file_non_atomic, 'a', StringIO('some text for a\n'))
416
420
            return
417
421
 
418
 
        self.failIf(t.has('a'))
 
422
        self.assertFalse(t.has('a'))
419
423
        t.put_file_non_atomic('a', StringIO('some text for a\n'))
420
 
        self.failUnless(t.has('a'))
 
424
        self.assertTrue(t.has('a'))
421
425
        self.check_transport_contents('some text for a\n', t, 'a')
422
426
        # Put also replaces contents
423
427
        t.put_file_non_atomic('a', StringIO('new\ncontents for\na\n'))
435
439
        # Now test the create_parent flag
436
440
        self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'dir/a',
437
441
                                       StringIO('contents\n'))
438
 
        self.failIf(t.has('dir/a'))
 
442
        self.assertFalse(t.has('dir/a'))
439
443
        t.put_file_non_atomic('dir/a', StringIO('contents for dir/a\n'),
440
444
                              create_parent_dir=True)
441
445
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
640
644
            self.build_tree(files, transport=transport_from)
641
645
            self.assertEqual(4, transport_from.copy_to(files, transport_to))
642
646
            for f in files:
643
 
                self.check_transport_contents(transport_to.get(f).read(),
 
647
                self.check_transport_contents(transport_to.get_bytes(f),
644
648
                                              transport_from, f)
645
649
 
646
650
        t = self.get_transport()
669
673
        files = ['a', 'b', 'c', 'd']
670
674
        t.copy_to(iter(files), temp_transport)
671
675
        for f in files:
672
 
            self.check_transport_contents(temp_transport.get(f).read(),
 
676
            self.check_transport_contents(temp_transport.get_bytes(f),
673
677
                                          t, f)
674
678
        del temp_transport
675
679
 
818
822
            return
819
823
 
820
824
        t.put_bytes('a', 'a little bit of text\n')
821
 
        self.failUnless(t.has('a'))
 
825
        self.assertTrue(t.has('a'))
822
826
        t.delete('a')
823
 
        self.failIf(t.has('a'))
 
827
        self.assertFalse(t.has('a'))
824
828
 
825
829
        self.assertRaises(NoSuchFile, t.delete, 'a')
826
830
 
832
836
        t.delete_multi(['a', 'c'])
833
837
        self.assertEqual([False, True, False],
834
838
                list(t.has_multi(['a', 'b', 'c'])))
835
 
        self.failIf(t.has('a'))
836
 
        self.failUnless(t.has('b'))
837
 
        self.failIf(t.has('c'))
 
839
        self.assertFalse(t.has('a'))
 
840
        self.assertTrue(t.has('b'))
 
841
        self.assertFalse(t.has('c'))
838
842
 
839
843
        self.assertRaises(NoSuchFile,
840
844
                t.delete_multi, ['a', 'b', 'c'])
901
905
        t.mkdir('foo-baz')
902
906
        t.rmdir('foo')
903
907
        self.assertRaises((NoSuchFile, PathError), t.rmdir, 'foo')
904
 
        self.failUnless(t.has('foo-bar'))
 
908
        self.assertTrue(t.has('foo-bar'))
905
909
 
906
910
    def test_rename_dir_succeeds(self):
907
911
        t = self.get_transport()
990
994
        self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
991
995
 
992
996
        t.move('a', 'b')
993
 
        self.failUnless(t.has('b'))
994
 
        self.failIf(t.has('a'))
 
997
        self.assertTrue(t.has('b'))
 
998
        self.assertFalse(t.has('a'))
995
999
 
996
1000
        self.check_transport_contents('a first file\n', t, 'b')
997
1001
        self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
999
1003
        # Overwrite a file
1000
1004
        t.put_bytes('c', 'c this file\n')
1001
1005
        t.move('c', 'b')
1002
 
        self.failIf(t.has('c'))
 
1006
        self.assertFalse(t.has('c'))
1003
1007
        self.check_transport_contents('c this file\n', t, 'b')
1004
1008
 
1005
1009
        # TODO: Try to write a test for atomicity
1037
1041
        except NotImplementedError:
1038
1042
            raise TestSkipped("Transport %s has no bogus URL support." %
1039
1043
                              self._server.__class__)
1040
 
        t = get_transport(url)
 
1044
        t = _mod_transport.get_transport(url)
1041
1045
        self.assertRaises((ConnectionError, NoSuchFile), t.get, '.bzr/branch')
1042
1046
 
1043
1047
    def test_stat(self):
1059
1063
        for path, size in zip(paths, sizes):
1060
1064
            st = t.stat(path)
1061
1065
            if path.endswith('/'):
1062
 
                self.failUnless(S_ISDIR(st.st_mode))
 
1066
                self.assertTrue(S_ISDIR(st.st_mode))
1063
1067
                # directory sizes are meaningless
1064
1068
            else:
1065
 
                self.failUnless(S_ISREG(st.st_mode))
 
1069
                self.assertTrue(S_ISREG(st.st_mode))
1066
1070
                self.assertEqual(size, st.st_size)
1067
1071
 
1068
1072
        remote_stats = list(t.stat_multi(paths))
1091
1095
        try:
1092
1096
            t.hardlink(source_name, link_name)
1093
1097
 
1094
 
            self.failUnless(t.has(source_name))
1095
 
            self.failUnless(t.has(link_name))
 
1098
            self.assertTrue(t.has(source_name))
 
1099
            self.assertTrue(t.has(link_name))
1096
1100
 
1097
1101
            st = t.stat(link_name)
1098
 
            self.failUnlessEqual(st[ST_NLINK], 2)
 
1102
            self.assertEqual(st[ST_NLINK], 2)
1099
1103
        except TransportNotPossible:
1100
1104
            raise TestSkipped("Transport %s does not support hardlinks." %
1101
1105
                              self._server.__class__)
1113
1117
        try:
1114
1118
            t.symlink(source_name, link_name)
1115
1119
 
1116
 
            self.failUnless(t.has(source_name))
1117
 
            self.failUnless(t.has(link_name))
 
1120
            self.assertTrue(t.has(source_name))
 
1121
            self.assertTrue(t.has(link_name))
1118
1122
 
1119
1123
            st = t.stat(link_name)
1120
 
            self.failUnless(S_ISLNK(st.st_mode),
 
1124
            self.assertTrue(S_ISLNK(st.st_mode),
1121
1125
                "expected symlink, got mode %o" % st.st_mode)
1122
1126
        except TransportNotPossible:
1123
1127
            raise TestSkipped("Transport %s does not support symlinks." %
1289
1293
 
1290
1294
        self.build_tree(['a', 'b/', 'b/c'], transport=t1)
1291
1295
 
1292
 
        self.failUnless(t1.has('a'))
1293
 
        self.failUnless(t1.has('b/c'))
1294
 
        self.failIf(t1.has('c'))
 
1296
        self.assertTrue(t1.has('a'))
 
1297
        self.assertTrue(t1.has('b/c'))
 
1298
        self.assertFalse(t1.has('c'))
1295
1299
 
1296
1300
        t2 = t1.clone('b')
1297
1301
        self.assertEqual(t1.base + 'b/', t2.base)
1298
1302
 
1299
 
        self.failUnless(t2.has('c'))
1300
 
        self.failIf(t2.has('a'))
 
1303
        self.assertTrue(t2.has('c'))
 
1304
        self.assertFalse(t2.has('a'))
1301
1305
 
1302
1306
        t3 = t2.clone('..')
1303
 
        self.failUnless(t3.has('a'))
1304
 
        self.failIf(t3.has('c'))
 
1307
        self.assertTrue(t3.has('a'))
 
1308
        self.assertFalse(t3.has('c'))
1305
1309
 
1306
 
        self.failIf(t1.has('b/d'))
1307
 
        self.failIf(t2.has('d'))
1308
 
        self.failIf(t3.has('b/d'))
 
1310
        self.assertFalse(t1.has('b/d'))
 
1311
        self.assertFalse(t2.has('d'))
 
1312
        self.assertFalse(t3.has('b/d'))
1309
1313
 
1310
1314
        if t1.is_readonly():
1311
1315
            self.build_tree_contents([('b/d', 'newfile\n')])
1312
1316
        else:
1313
1317
            t2.put_bytes('d', 'newfile\n')
1314
1318
 
1315
 
        self.failUnless(t1.has('b/d'))
1316
 
        self.failUnless(t2.has('d'))
1317
 
        self.failUnless(t3.has('b/d'))
 
1319
        self.assertTrue(t1.has('b/d'))
 
1320
        self.assertTrue(t2.has('d'))
 
1321
        self.assertTrue(t3.has('b/d'))
1318
1322
 
1319
1323
    def test_clone_to_root(self):
1320
1324
        orig_transport = self.get_transport()
1394
1398
        self.assertEqual(transport.clone("/").abspath('foo'),
1395
1399
                         transport.abspath("/foo"))
1396
1400
 
 
1401
    # GZ 2011-01-26: Test in per_transport but not using self.get_transport?
1397
1402
    def test_win32_abspath(self):
1398
1403
        # Note: we tried to set sys.platform='win32' so we could test on
1399
1404
        # other platforms too, but then osutils does platform specific
1404
1409
 
1405
1410
        # smoke test for abspath on win32.
1406
1411
        # a transport based on 'file:///' never fully qualifies the drive.
1407
 
        transport = get_transport("file:///")
1408
 
        self.failUnlessEqual(transport.abspath("/"), "file:///")
 
1412
        transport = _mod_transport.get_transport("file:///")
 
1413
        self.assertEqual(transport.abspath("/"), "file:///")
1409
1414
 
1410
1415
        # but a transport that starts with a drive spec must keep it.
1411
 
        transport = get_transport("file:///C:/")
1412
 
        self.failUnlessEqual(transport.abspath("/"), "file:///C:/")
 
1416
        transport = _mod_transport.get_transport("file:///C:/")
 
1417
        self.assertEqual(transport.abspath("/"), "file:///C:/")
1413
1418
 
1414
1419
    def test_local_abspath(self):
1415
1420
        transport = self.get_transport()