diff --git a/_pytest/assertion/rewrite.py b/_pytest/assertion/rewrite.py index db36749308b..0499a792f7f 100644 --- a/_pytest/assertion/rewrite.py +++ b/_pytest/assertion/rewrite.py @@ -12,7 +12,9 @@ import sys import types +import atomicwrites import py + from _pytest.assertion import util @@ -140,7 +142,7 @@ def find_module(self, name, path=None): # Probably a SyntaxError in the test. return None if write: - _make_rewritten_pyc(state, source_stat, pyc, co) + _write_pyc(state, co, source_stat, pyc) else: state.trace("found cached rewritten pyc for %r" % (fn,)) self.modules[name] = co, pyc @@ -258,22 +260,21 @@ def _write_pyc(state, co, source_stat, pyc): # sometime to be able to use imp.load_compiled to load them. (See # the comment in load_module above.) try: - fp = open(pyc, "wb") - except IOError: - err = sys.exc_info()[1].errno - state.trace("error writing pyc file at %s: errno=%s" % (pyc, err)) + with atomicwrites.atomic_write(pyc, mode="wb", overwrite=True) as fp: + fp.write(imp.get_magic()) + mtime = int(source_stat.mtime) + size = source_stat.size & 0xFFFFFFFF + fp.write(struct.pack("`_ library. diff --git a/setup.py b/setup.py index 984994a7f63..d6defba6b89 100644 --- a/setup.py +++ b/setup.py @@ -61,6 +61,7 @@ def main(): 'setuptools', 'attrs>=17.4.0', 'more-itertools>=4.0.0', + 'atomicwrites>=1.0', ] # if _PYTEST_SETUP_SKIP_PLUGGY_DEP is set, skip installing pluggy; # used by tox.ini to test with pluggy master diff --git a/testing/test_assertrewrite.py b/testing/test_assertrewrite.py index 2a3dbc2ecc3..4f7c95600c6 100644 --- a/testing/test_assertrewrite.py +++ b/testing/test_assertrewrite.py @@ -839,22 +839,22 @@ def test_meta_path(): def test_write_pyc(self, testdir, tmpdir, monkeypatch): from _pytest.assertion.rewrite import _write_pyc from _pytest.assertion import AssertionState - try: - import __builtin__ as b - except ImportError: - import builtins as b + import atomicwrites + from contextlib import contextmanager config = testdir.parseconfig([]) state = AssertionState(config, "rewrite") source_path = tmpdir.ensure("source.py") pycpath = tmpdir.join("pyc").strpath assert _write_pyc(state, [1], source_path.stat(), pycpath) - def open(*args): + @contextmanager + def atomic_write_failed(fn, mode='r', overwrite=False): e = IOError() e.errno = 10 raise e + yield # noqa - monkeypatch.setattr(b, "open", open) + monkeypatch.setattr(atomicwrites, "atomic_write", atomic_write_failed) assert not _write_pyc(state, [1], source_path.stat(), pycpath) def test_resources_provider_for_loader(self, testdir):