/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/per_transport.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2011-05-02 21:13:54 UTC
  • mfrom: (5815.1.1 rm-get-vf-store)
  • Revision ID: pqm@pqm.ubuntu.com-20110502211354-cbni6j20jyhhmvfw
(jelmer) Remove unused RepositoryFormat._get_versioned_file_store. (Jelmer
 Vernooij)

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"""
251
247
 
252
248
    def test_get_bytes_unknown_file(self):
253
249
        t = self.get_transport()
254
 
 
255
250
        self.assertRaises(NoSuchFile, t.get_bytes, 'c')
256
251
 
257
252
    def test_get_with_open_write_stream_sees_all_content(self):
261
256
        handle = t.open_write_stream('foo')
262
257
        try:
263
258
            handle.write('b')
264
 
            self.assertEqual('b', t.get('foo').read())
 
259
            self.assertEqual('b', t.get_bytes('foo'))
265
260
        finally:
266
261
            handle.close()
267
262
 
273
268
        try:
274
269
            handle.write('b')
275
270
            self.assertEqual('b', t.get_bytes('foo'))
276
 
            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()
277
276
        finally:
278
277
            handle.close()
279
278
 
286
285
            return
287
286
 
288
287
        t.put_bytes('a', 'some text for a\n')
289
 
        self.failUnless(t.has('a'))
 
288
        self.assertTrue(t.has('a'))
290
289
        self.check_transport_contents('some text for a\n', t, 'a')
291
290
 
292
291
        # The contents should be overwritten
304
303
                    t.put_bytes_non_atomic, 'a', 'some text for a\n')
305
304
            return
306
305
 
307
 
        self.failIf(t.has('a'))
 
306
        self.assertFalse(t.has('a'))
308
307
        t.put_bytes_non_atomic('a', 'some text for a\n')
309
 
        self.failUnless(t.has('a'))
 
308
        self.assertTrue(t.has('a'))
310
309
        self.check_transport_contents('some text for a\n', t, 'a')
311
310
        # Put also replaces contents
312
311
        t.put_bytes_non_atomic('a', 'new\ncontents for\na\n')
324
323
        # Now test the create_parent flag
325
324
        self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'dir/a',
326
325
                                       'contents\n')
327
 
        self.failIf(t.has('dir/a'))
 
326
        self.assertFalse(t.has('dir/a'))
328
327
        t.put_bytes_non_atomic('dir/a', 'contents for dir/a\n',
329
328
                               create_parent_dir=True)
330
329
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
402
401
        result = t.put_file('a', StringIO('some text for a\n'))
403
402
        # put_file returns the length of the data written
404
403
        self.assertEqual(16, result)
405
 
        self.failUnless(t.has('a'))
 
404
        self.assertTrue(t.has('a'))
406
405
        self.check_transport_contents('some text for a\n', t, 'a')
407
406
        # Put also replaces contents
408
407
        result = t.put_file('a', StringIO('new\ncontents for\na\n'))
420
419
                    t.put_file_non_atomic, 'a', StringIO('some text for a\n'))
421
420
            return
422
421
 
423
 
        self.failIf(t.has('a'))
 
422
        self.assertFalse(t.has('a'))
424
423
        t.put_file_non_atomic('a', StringIO('some text for a\n'))
425
 
        self.failUnless(t.has('a'))
 
424
        self.assertTrue(t.has('a'))
426
425
        self.check_transport_contents('some text for a\n', t, 'a')
427
426
        # Put also replaces contents
428
427
        t.put_file_non_atomic('a', StringIO('new\ncontents for\na\n'))
440
439
        # Now test the create_parent flag
441
440
        self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'dir/a',
442
441
                                       StringIO('contents\n'))
443
 
        self.failIf(t.has('dir/a'))
 
442
        self.assertFalse(t.has('dir/a'))
444
443
        t.put_file_non_atomic('dir/a', StringIO('contents for dir/a\n'),
445
444
                              create_parent_dir=True)
446
445
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
645
644
            self.build_tree(files, transport=transport_from)
646
645
            self.assertEqual(4, transport_from.copy_to(files, transport_to))
647
646
            for f in files:
648
 
                self.check_transport_contents(transport_to.get(f).read(),
 
647
                self.check_transport_contents(transport_to.get_bytes(f),
649
648
                                              transport_from, f)
650
649
 
651
650
        t = self.get_transport()
674
673
        files = ['a', 'b', 'c', 'd']
675
674
        t.copy_to(iter(files), temp_transport)
676
675
        for f in files:
677
 
            self.check_transport_contents(temp_transport.get(f).read(),
 
676
            self.check_transport_contents(temp_transport.get_bytes(f),
678
677
                                          t, f)
679
678
        del temp_transport
680
679
 
823
822
            return
824
823
 
825
824
        t.put_bytes('a', 'a little bit of text\n')
826
 
        self.failUnless(t.has('a'))
 
825
        self.assertTrue(t.has('a'))
827
826
        t.delete('a')
828
 
        self.failIf(t.has('a'))
 
827
        self.assertFalse(t.has('a'))
829
828
 
830
829
        self.assertRaises(NoSuchFile, t.delete, 'a')
831
830
 
837
836
        t.delete_multi(['a', 'c'])
838
837
        self.assertEqual([False, True, False],
839
838
                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'))
 
839
        self.assertFalse(t.has('a'))
 
840
        self.assertTrue(t.has('b'))
 
841
        self.assertFalse(t.has('c'))
843
842
 
844
843
        self.assertRaises(NoSuchFile,
845
844
                t.delete_multi, ['a', 'b', 'c'])
906
905
        t.mkdir('foo-baz')
907
906
        t.rmdir('foo')
908
907
        self.assertRaises((NoSuchFile, PathError), t.rmdir, 'foo')
909
 
        self.failUnless(t.has('foo-bar'))
 
908
        self.assertTrue(t.has('foo-bar'))
910
909
 
911
910
    def test_rename_dir_succeeds(self):
912
911
        t = self.get_transport()
995
994
        self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
996
995
 
997
996
        t.move('a', 'b')
998
 
        self.failUnless(t.has('b'))
999
 
        self.failIf(t.has('a'))
 
997
        self.assertTrue(t.has('b'))
 
998
        self.assertFalse(t.has('a'))
1000
999
 
1001
1000
        self.check_transport_contents('a first file\n', t, 'b')
1002
1001
        self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
1004
1003
        # Overwrite a file
1005
1004
        t.put_bytes('c', 'c this file\n')
1006
1005
        t.move('c', 'b')
1007
 
        self.failIf(t.has('c'))
 
1006
        self.assertFalse(t.has('c'))
1008
1007
        self.check_transport_contents('c this file\n', t, 'b')
1009
1008
 
1010
1009
        # TODO: Try to write a test for atomicity
1042
1041
        except NotImplementedError:
1043
1042
            raise TestSkipped("Transport %s has no bogus URL support." %
1044
1043
                              self._server.__class__)
1045
 
        t = get_transport(url)
 
1044
        t = _mod_transport.get_transport(url)
1046
1045
        self.assertRaises((ConnectionError, NoSuchFile), t.get, '.bzr/branch')
1047
1046
 
1048
1047
    def test_stat(self):
1064
1063
        for path, size in zip(paths, sizes):
1065
1064
            st = t.stat(path)
1066
1065
            if path.endswith('/'):
1067
 
                self.failUnless(S_ISDIR(st.st_mode))
 
1066
                self.assertTrue(S_ISDIR(st.st_mode))
1068
1067
                # directory sizes are meaningless
1069
1068
            else:
1070
 
                self.failUnless(S_ISREG(st.st_mode))
 
1069
                self.assertTrue(S_ISREG(st.st_mode))
1071
1070
                self.assertEqual(size, st.st_size)
1072
1071
 
1073
1072
        remote_stats = list(t.stat_multi(paths))
1096
1095
        try:
1097
1096
            t.hardlink(source_name, link_name)
1098
1097
 
1099
 
            self.failUnless(t.has(source_name))
1100
 
            self.failUnless(t.has(link_name))
 
1098
            self.assertTrue(t.has(source_name))
 
1099
            self.assertTrue(t.has(link_name))
1101
1100
 
1102
1101
            st = t.stat(link_name)
1103
 
            self.failUnlessEqual(st[ST_NLINK], 2)
 
1102
            self.assertEqual(st[ST_NLINK], 2)
1104
1103
        except TransportNotPossible:
1105
1104
            raise TestSkipped("Transport %s does not support hardlinks." %
1106
1105
                              self._server.__class__)
1118
1117
        try:
1119
1118
            t.symlink(source_name, link_name)
1120
1119
 
1121
 
            self.failUnless(t.has(source_name))
1122
 
            self.failUnless(t.has(link_name))
 
1120
            self.assertTrue(t.has(source_name))
 
1121
            self.assertTrue(t.has(link_name))
1123
1122
 
1124
1123
            st = t.stat(link_name)
1125
 
            self.failUnless(S_ISLNK(st.st_mode))
 
1124
            self.assertTrue(S_ISLNK(st.st_mode),
 
1125
                "expected symlink, got mode %o" % st.st_mode)
1126
1126
        except TransportNotPossible:
1127
1127
            raise TestSkipped("Transport %s does not support symlinks." %
1128
1128
                              self._server.__class__)
1265
1265
        self.assertIs(t._get_connection(), c._get_connection())
1266
1266
 
1267
1267
        # Temporary failure, we need to create a new dummy connection
1268
 
        new_connection = object()
 
1268
        new_connection = None
1269
1269
        t._set_connection(new_connection)
1270
1270
        # Check that both transports use the same connection
1271
1271
        self.assertIs(new_connection, t._get_connection())
1293
1293
 
1294
1294
        self.build_tree(['a', 'b/', 'b/c'], transport=t1)
1295
1295
 
1296
 
        self.failUnless(t1.has('a'))
1297
 
        self.failUnless(t1.has('b/c'))
1298
 
        self.failIf(t1.has('c'))
 
1296
        self.assertTrue(t1.has('a'))
 
1297
        self.assertTrue(t1.has('b/c'))
 
1298
        self.assertFalse(t1.has('c'))
1299
1299
 
1300
1300
        t2 = t1.clone('b')
1301
1301
        self.assertEqual(t1.base + 'b/', t2.base)
1302
1302
 
1303
 
        self.failUnless(t2.has('c'))
1304
 
        self.failIf(t2.has('a'))
 
1303
        self.assertTrue(t2.has('c'))
 
1304
        self.assertFalse(t2.has('a'))
1305
1305
 
1306
1306
        t3 = t2.clone('..')
1307
 
        self.failUnless(t3.has('a'))
1308
 
        self.failIf(t3.has('c'))
 
1307
        self.assertTrue(t3.has('a'))
 
1308
        self.assertFalse(t3.has('c'))
1309
1309
 
1310
 
        self.failIf(t1.has('b/d'))
1311
 
        self.failIf(t2.has('d'))
1312
 
        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'))
1313
1313
 
1314
1314
        if t1.is_readonly():
1315
1315
            self.build_tree_contents([('b/d', 'newfile\n')])
1316
1316
        else:
1317
1317
            t2.put_bytes('d', 'newfile\n')
1318
1318
 
1319
 
        self.failUnless(t1.has('b/d'))
1320
 
        self.failUnless(t2.has('d'))
1321
 
        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'))
1322
1322
 
1323
1323
    def test_clone_to_root(self):
1324
1324
        orig_transport = self.get_transport()
1398
1398
        self.assertEqual(transport.clone("/").abspath('foo'),
1399
1399
                         transport.abspath("/foo"))
1400
1400
 
 
1401
    # GZ 2011-01-26: Test in per_transport but not using self.get_transport?
1401
1402
    def test_win32_abspath(self):
1402
1403
        # Note: we tried to set sys.platform='win32' so we could test on
1403
1404
        # other platforms too, but then osutils does platform specific
1408
1409
 
1409
1410
        # smoke test for abspath on win32.
1410
1411
        # a transport based on 'file:///' never fully qualifies the drive.
1411
 
        transport = get_transport("file:///")
1412
 
        self.failUnlessEqual(transport.abspath("/"), "file:///")
 
1412
        transport = _mod_transport.get_transport("file:///")
 
1413
        self.assertEqual(transport.abspath("/"), "file:///")
1413
1414
 
1414
1415
        # 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:/")
 
1416
        transport = _mod_transport.get_transport("file:///C:/")
 
1417
        self.assertEqual(transport.abspath("/"), "file:///C:/")
1417
1418
 
1418
1419
    def test_local_abspath(self):
1419
1420
        transport = self.get_transport()
1765
1766
        # also raise a special error
1766
1767
        self.assertListRaises((errors.ShortReadvError, errors.InvalidRange),
1767
1768
                              transport.readv, 'a', [(12,2)])
 
1769
 
 
1770
    def test_stat_symlink(self):
 
1771
        # if a transport points directly to a symlink (and supports symlinks
 
1772
        # at all) you can tell this.  helps with bug 32669.
 
1773
        t = self.get_transport()
 
1774
        try:
 
1775
            t.symlink('target', 'link')
 
1776
        except TransportNotPossible:
 
1777
            raise TestSkipped("symlinks not supported")
 
1778
        t2 = t.clone('link')
 
1779
        st = t2.stat('')
 
1780
        self.assertTrue(stat.S_ISLNK(st.st_mode))