blob: b3128c5b4f16043e8c76e8927708fd70b5cf1f22 (
plain) (
blame)
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
90
91
|
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):
aw = atomic_write(str(fname), overwrite=False)
with aw:
# Mess with internals, so commit will trigger a ValueError. We're
# testing that the initial AssertionError triggered below is
# propagated up the stack, not the second exception triggered
# during commit.
aw.rollback = lambda: 1 / 0
# 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)
|
