32
34
def test_commit(self):
33
35
f = atomicfile.AtomicFile('test')
34
36
self.assertPathDoesNotExist('test')
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
45
47
def test_abort(self):
46
48
f = atomicfile.AtomicFile('test')
49
51
self.assertEqual([], os.listdir('.'))
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)
54
56
# close is re-entrant safe
57
59
def test_close(self):
58
60
f = atomicfile.AtomicFile('test')
60
62
# close on an open file is an abort
62
64
self.assertEqual([], os.listdir('.'))
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)
67
69
# close is re-entrant safe
70
72
def test_text_mode(self):
71
73
f = atomicfile.AtomicFile('test', mode='wt')
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)
79
self.assertEqual(b'foo\n', contents)
81
self.assertEqual('foo\n', contents)
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)
91
93
st = os.lstat('test')
92
94
self.assertEqualMode(mode, stat.S_IMODE(st.st_mode))
94
96
def test_mode_0666(self):
95
self._test_mode(0o666)
97
99
def test_mode_0664(self):
98
self._test_mode(0o664)
100
def test_mode_0660(self):
101
self._test_mode(0o660)
103
def test_mode_0660(self):
104
self._test_mode(0o660)
100
self._test_mode(0664)
102
def test_mode_0660(self):
103
self._test_mode(0660)
105
def test_mode_0660(self):
106
self._test_mode(0660)
106
108
def test_mode_0640(self):
107
self._test_mode(0o640)
109
self._test_mode(0640)
109
111
def test_mode_0600(self):
110
self._test_mode(0o600)
112
self._test_mode(0600)
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)
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')
123
125
st = os.lstat('test')
124
self.assertEqualMode(0o666 & ~umask, stat.S_IMODE(st.st_mode))
126
def test_context_manager_commit(self):
127
with atomicfile.AtomicFile('test') as f:
128
self.assertPathDoesNotExist('test')
131
self.assertEqual(['test'], os.listdir('.'))
132
self.check_file_contents('test', b'foo\n')
134
def test_context_manager_abort(self):
136
with atomicfile.AtomicFile('test') as f:
139
self.assertRaises(AssertionError, abort)
140
self.assertEqual([], os.listdir('.'))
126
self.assertEqualMode(0666 & ~umask, stat.S_IMODE(st.st_mode))