1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
|
import errno
import os
from atomicwrites import atomic_write
import pytest
def test_atomic_write(tmpdir):
fname = tmpdir.join('ha')
for i in range(2):
with atomic_write(str(fname), overwrite=True) as f:
f.write('hoho')
with pytest.raises(OSError) as excinfo:
with atomic_write(str(fname), overwrite=False) as f:
f.write('haha')
assert excinfo.value.errno == errno.EEXIST
assert fname.read() == 'hoho'
assert len(tmpdir.listdir()) == 1
def test_teardown(tmpdir):
fname = tmpdir.join('ha')
with pytest.raises(AssertionError):
with atomic_write(str(fname), overwrite=True):
assert False
assert not tmpdir.listdir()
def test_replace_simultaneously_created_file(tmpdir):
fname = tmpdir.join('ha')
with atomic_write(str(fname), overwrite=True) as f:
f.write('hoho')
fname.write('harhar')
assert fname.read() == 'harhar'
assert fname.read() == 'hoho'
assert len(tmpdir.listdir()) == 1
def test_dont_remove_simultaneously_created_file(tmpdir):
fname = tmpdir.join('ha')
with pytest.raises(OSError) as excinfo:
with atomic_write(str(fname), overwrite=False) as f:
f.write('hoho')
fname.write('harhar')
assert fname.read() == 'harhar'
assert excinfo.value.errno == errno.EEXIST
assert fname.read() == 'harhar'
assert len(tmpdir.listdir()) == 1
# Verify that nested exceptions during rollback do not overwrite the initial
# exception that triggered a rollback.
def test_open_reraise(tmpdir):
fname = tmpdir.join('ha')
with pytest.raises(AssertionError):
with atomic_write(str(fname), overwrite=False) as f:
# Mess with f, so rollback will trigger an OSError. We're testing
# that the initial AssertionError triggered below is propagated up
# the stack, not the second exception triggered during rollback.
f.name = "asdf"
# Now trigger our own exception.
assert False, "Intentional failure for testing purposes"
def test_atomic_write_in_pwd(tmpdir):
orig_curdir = os.getcwd()
try:
os.chdir(str(tmpdir))
fname = 'ha'
for i in range(2):
with atomic_write(str(fname), overwrite=True) as f:
f.write('hoho')
with pytest.raises(OSError) as excinfo:
with atomic_write(str(fname), overwrite=False) as f:
f.write('haha')
assert excinfo.value.errno == errno.EEXIST
assert open(fname).read() == 'hoho'
assert len(tmpdir.listdir()) == 1
finally:
os.chdir(orig_curdir)
|
