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

  • Committer: Jelmer Vernooij
  • Date: 2018-02-18 21:42:57 UTC
  • mto: This revision was merged to the branch mainline in revision 6859.
  • Revision ID: jelmer@jelmer.uk-20180218214257-jpevutp1wa30tz3v
Update TODO to reference Breezy, not Bazaar.

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/tests/test_atomicfile.py
22
22
 
23
23
from .. import (
24
24
    atomicfile,
 
25
    errors,
25
26
    osutils,
26
27
    )
27
28
from . import TestCaseInTempDir, TestSkipped
32
33
    def test_commit(self):
33
34
        f = atomicfile.AtomicFile('test')
34
35
        self.assertPathDoesNotExist('test')
35
 
        f.write(b'foo\n')
 
36
        f.write('foo\n')
36
37
        f.commit()
37
38
 
38
39
        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)
 
40
        self.check_file_contents('test', 'foo\n')
 
41
        self.assertRaises(errors.AtomicFileAlreadyClosed, f.commit)
 
42
        self.assertRaises(errors.AtomicFileAlreadyClosed, f.abort)
42
43
        # close is re-entrant safe
43
44
        f.close()
44
45
 
45
46
    def test_abort(self):
46
47
        f = atomicfile.AtomicFile('test')
47
 
        f.write(b'foo\n')
 
48
        f.write('foo\n')
48
49
        f.abort()
49
50
        self.assertEqual([], os.listdir('.'))
50
51
 
51
 
        self.assertRaises(atomicfile.AtomicFileAlreadyClosed, f.abort)
52
 
        self.assertRaises(atomicfile.AtomicFileAlreadyClosed, f.commit)
 
52
        self.assertRaises(errors.AtomicFileAlreadyClosed, f.abort)
 
53
        self.assertRaises(errors.AtomicFileAlreadyClosed, f.commit)
53
54
 
54
55
        # close is re-entrant safe
55
56
        f.close()
56
57
 
57
58
    def test_close(self):
58
59
        f = atomicfile.AtomicFile('test')
59
 
        f.write(b'foo\n')
 
60
        f.write('foo\n')
60
61
        # close on an open file is an abort
61
62
        f.close()
62
63
        self.assertEqual([], os.listdir('.'))
63
64
 
64
 
        self.assertRaises(atomicfile.AtomicFileAlreadyClosed, f.abort)
65
 
        self.assertRaises(atomicfile.AtomicFileAlreadyClosed, f.commit)
 
65
        self.assertRaises(errors.AtomicFileAlreadyClosed, f.abort)
 
66
        self.assertRaises(errors.AtomicFileAlreadyClosed, f.commit)
66
67
 
67
68
        # close is re-entrant safe
68
69
        f.close()
69
70
 
70
71
    def test_text_mode(self):
71
72
        f = atomicfile.AtomicFile('test', mode='wt')
72
 
        f.write(b'foo\n')
 
73
        f.write('foo\n')
73
74
        f.commit()
74
75
 
75
76
        contents = open('test', 'rb').read()
76
77
        if sys.platform == 'win32':
77
 
            self.assertEqual(b'foo\r\n', contents)
 
78
            self.assertEqual('foo\r\n', contents)
78
79
        else:
79
 
            self.assertEqual(b'foo\n', contents)
 
80
            self.assertEqual('foo\n', contents)
80
81
 
81
82
    def can_sys_preserve_mode(self):
82
83
        # PLATFORM DEFICIENCY/ TestSkipped
86
87
        if not self.can_sys_preserve_mode():
87
88
            raise TestSkipped("This test cannot be run on your platform")
88
89
        f = atomicfile.AtomicFile('test', mode='wb', new_mode=mode)
89
 
        f.write(b'foo\n')
 
90
        f.write('foo\n')
90
91
        f.commit()
91
92
        st = os.lstat('test')
92
93
        self.assertEqualMode(mode, stat.S_IMODE(st.st_mode))
118
119
        # The default file permissions should be based on umask
119
120
        umask = osutils.get_umask()
120
121
        f = atomicfile.AtomicFile('test', mode='wb')
121
 
        f.write(b'foo\n')
 
122
        f.write('foo\n')
122
123
        f.commit()
123
124
        st = os.lstat('test')
124
125
        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('.'))