/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 brzlib/tests/test_atomicfile.py

  • Committer: Jelmer Vernooij
  • Date: 2017-05-21 12:41:27 UTC
  • mto: This revision was merged to the branch mainline in revision 6623.
  • Revision ID: jelmer@jelmer.uk-20170521124127-iv8etg0vwymyai6y
s/bzr/brz/ in apport config.

Show diffs side-by-side

added added

removed removed

Lines of Context:
20
20
import stat
21
21
import sys
22
22
 
23
 
from .. import (
 
23
from brzlib import (
24
24
    atomicfile,
 
25
    errors,
25
26
    osutils,
 
27
    symbol_versioning,
26
28
    )
27
 
from . import TestCaseInTempDir, TestSkipped
 
29
from brzlib.tests import TestCaseInTempDir, TestSkipped
28
30
 
29
31
 
30
32
class TestAtomicFile(TestCaseInTempDir):
32
34
    def test_commit(self):
33
35
        f = atomicfile.AtomicFile('test')
34
36
        self.assertPathDoesNotExist('test')
35
 
        f.write(b'foo\n')
 
37
        f.write('foo\n')
36
38
        f.commit()
37
39
 
38
40
        self.assertEqual(['test'], os.listdir('.'))
39
 
        self.check_file_contents('test', b'foo\n')
40
 
        self.assertRaises(atomicfile.AtomicFileAlreadyClosed, f.commit)
41
 
        self.assertRaises(atomicfile.AtomicFileAlreadyClosed, f.abort)
 
41
        self.check_file_contents('test', 'foo\n')
 
42
        self.assertRaises(errors.AtomicFileAlreadyClosed, f.commit)
 
43
        self.assertRaises(errors.AtomicFileAlreadyClosed, f.abort)
42
44
        # close is re-entrant safe
43
45
        f.close()
44
46
 
45
47
    def test_abort(self):
46
48
        f = atomicfile.AtomicFile('test')
47
 
        f.write(b'foo\n')
 
49
        f.write('foo\n')
48
50
        f.abort()
49
51
        self.assertEqual([], os.listdir('.'))
50
52
 
51
 
        self.assertRaises(atomicfile.AtomicFileAlreadyClosed, f.abort)
52
 
        self.assertRaises(atomicfile.AtomicFileAlreadyClosed, f.commit)
 
53
        self.assertRaises(errors.AtomicFileAlreadyClosed, f.abort)
 
54
        self.assertRaises(errors.AtomicFileAlreadyClosed, f.commit)
53
55
 
54
56
        # close is re-entrant safe
55
57
        f.close()
56
58
 
57
59
    def test_close(self):
58
60
        f = atomicfile.AtomicFile('test')
59
 
        f.write(b'foo\n')
 
61
        f.write('foo\n')
60
62
        # close on an open file is an abort
61
63
        f.close()
62
64
        self.assertEqual([], os.listdir('.'))
63
65
 
64
 
        self.assertRaises(atomicfile.AtomicFileAlreadyClosed, f.abort)
65
 
        self.assertRaises(atomicfile.AtomicFileAlreadyClosed, f.commit)
 
66
        self.assertRaises(errors.AtomicFileAlreadyClosed, f.abort)
 
67
        self.assertRaises(errors.AtomicFileAlreadyClosed, f.commit)
66
68
 
67
69
        # close is re-entrant safe
68
70
        f.close()
69
71
 
70
72
    def test_text_mode(self):
71
73
        f = atomicfile.AtomicFile('test', mode='wt')
72
 
        f.write(b'foo\n')
 
74
        f.write('foo\n')
73
75
        f.commit()
74
76
 
75
77
        contents = open('test', 'rb').read()
76
78
        if sys.platform == 'win32':
77
 
            self.assertEqual(b'foo\r\n', contents)
 
79
            self.assertEqual('foo\r\n', contents)
78
80
        else:
79
 
            self.assertEqual(b'foo\n', contents)
 
81
            self.assertEqual('foo\n', contents)
80
82
 
81
83
    def can_sys_preserve_mode(self):
82
84
        # PLATFORM DEFICIENCY/ TestSkipped
86
88
        if not self.can_sys_preserve_mode():
87
89
            raise TestSkipped("This test cannot be run on your platform")
88
90
        f = atomicfile.AtomicFile('test', mode='wb', new_mode=mode)
89
 
        f.write(b'foo\n')
 
91
        f.write('foo\n')
90
92
        f.commit()
91
93
        st = os.lstat('test')
92
94
        self.assertEqualMode(mode, stat.S_IMODE(st.st_mode))
93
95
 
94
96
    def test_mode_0666(self):
95
 
        self._test_mode(0o666)
 
97
        self._test_mode(0666)
96
98
 
97
99
    def test_mode_0664(self):
98
 
        self._test_mode(0o664)
99
 
 
100
 
    def test_mode_0660(self):
101
 
        self._test_mode(0o660)
102
 
 
103
 
    def test_mode_0660(self):
104
 
        self._test_mode(0o660)
 
100
        self._test_mode(0664)
 
101
 
 
102
    def test_mode_0660(self):
 
103
        self._test_mode(0660)
 
104
 
 
105
    def test_mode_0660(self):
 
106
        self._test_mode(0660)
105
107
 
106
108
    def test_mode_0640(self):
107
 
        self._test_mode(0o640)
 
109
        self._test_mode(0640)
108
110
 
109
111
    def test_mode_0600(self):
110
 
        self._test_mode(0o600)
 
112
        self._test_mode(0600)
111
113
 
112
114
    def test_mode_0400(self):
113
 
        self._test_mode(0o400)
 
115
        self._test_mode(0400)
114
116
        # Make it read-write again so cleanup doesn't complain
115
 
        os.chmod('test', 0o600)
 
117
        os.chmod('test', 0600)
116
118
 
117
119
    def test_no_mode(self):
118
120
        # The default file permissions should be based on umask
119
121
        umask = osutils.get_umask()
120
122
        f = atomicfile.AtomicFile('test', mode='wb')
121
 
        f.write(b'foo\n')
 
123
        f.write('foo\n')
122
124
        f.commit()
123
125
        st = os.lstat('test')
124
 
        self.assertEqualMode(0o666 & ~umask, stat.S_IMODE(st.st_mode))
125
 
 
126
 
    def test_context_manager_commit(self):
127
 
        with atomicfile.AtomicFile('test') as f:
128
 
            self.assertPathDoesNotExist('test')
129
 
            f.write(b'foo\n')
130
 
 
131
 
        self.assertEqual(['test'], os.listdir('.'))
132
 
        self.check_file_contents('test', b'foo\n')
133
 
 
134
 
    def test_context_manager_abort(self):
135
 
        def abort():
136
 
            with atomicfile.AtomicFile('test') as f:
137
 
                f.write(b'foo\n')
138
 
                raise AssertionError
139
 
        self.assertRaises(AssertionError, abort)
140
 
        self.assertEqual([], os.listdir('.'))
 
126
        self.assertEqualMode(0666 & ~umask, stat.S_IMODE(st.st_mode))