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)
|