replace atomicwrites with os.replace

Author: asottile

.pre-commit-config.yaml

@@ -67,7 +67,6 @@ repos:
           - attrs>=19.2.0
           - packaging
           - tomli
-          - types-atomicwrites
           - types-pkg_resources
 -   repo: local
     hooks:

changelog/10114.trivial.rst

@@ -0,0 +1 @@
+Replace `atomicwrites <https://github.com/untitaker/python-atomicwrites>`__ dependency on windows with `os.replace`.

setup.cfg

@@ -46,7 +46,6 @@ install_requires =
     packaging
     pluggy>=0.12,<2.0
     py>=1.8.2
-    atomicwrites>=1.0;sys_platform=="win32"
     colorama;sys_platform=="win32"
     importlib-metadata>=0.12;python_version<"3.8"
     tomli>=1.0.0;python_version<"3.11"

src/_pytest/assertion/rewrite.py

@@ -302,53 +302,29 @@ def _write_pyc_fp(
     fp.write(marshal.dumps(co))
 
 
-if sys.platform == "win32":
-    from atomicwrites import atomic_write
-
-    def _write_pyc(
-        state: "AssertionState",
-        co: types.CodeType,
-        source_stat: os.stat_result,
-        pyc: Path,
-    ) -> bool:
-        try:
-            with atomic_write(os.fspath(pyc), mode="wb", overwrite=True) as fp:
-                _write_pyc_fp(fp, source_stat, co)
-        except OSError as e:
-            state.trace(f"error writing pyc file at {pyc}: {e}")
-            # we ignore any failure to write the cache file
-            # there are many reasons, permission-denied, pycache dir being a
-            # file etc.
-            return False
-        return True
-
-else:
-
-    def _write_pyc(
-        state: "AssertionState",
-        co: types.CodeType,
-        source_stat: os.stat_result,
-        pyc: Path,
-    ) -> bool:
-        proc_pyc = f"{pyc}.{os.getpid()}"
-        try:
-            fp = open(proc_pyc, "wb")
-        except OSError as e:
-            state.trace(f"error writing pyc file at {proc_pyc}: errno={e.errno}")
-            return False
-
-        try:
+def _write_pyc(
+    state: "AssertionState",
+    co: types.CodeType,
+    source_stat: os.stat_result,
+    pyc: Path,
+) -> bool:
+    proc_pyc = f"{pyc}.{os.getpid()}"
+    try:
+        with open(proc_pyc, "wb") as fp:
             _write_pyc_fp(fp, source_stat, co)
-            os.rename(proc_pyc, pyc)
-        except OSError as e:
-            state.trace(f"error writing pyc file at {pyc}: {e}")
-            # we ignore any failure to write the cache file
-            # there are many reasons, permission-denied, pycache dir being a
-            # file etc.
-            return False
-        finally:
-            fp.close()
-        return True
+    except OSError as e:
+        state.trace(f"error writing pyc file at {proc_pyc}: errno={e.errno}")
+        return False
+
+    try:
+        os.replace(proc_pyc, pyc)
+    except OSError as e:
+        state.trace(f"error writing pyc file at {pyc}: {e}")
+        # we ignore any failure to write the cache file
+        # there are many reasons, permission-denied, pycache dir being a
+        # file etc.
+        return False
+    return True
 
 
 def _rewrite_test(fn: Path, config: Config) -> Tuple[os.stat_result, types.CodeType]:

testing/test_assertrewrite.py

@@ -1009,7 +1009,7 @@ def test_meta_path():
         )
         assert pytester.runpytest().ret == 0
 
-    def test_write_pyc(self, pytester: Pytester, tmp_path, monkeypatch) -> None:
+    def test_write_pyc(self, pytester: Pytester, tmp_path) -> None:
         from _pytest.assertion.rewrite import _write_pyc
         from _pytest.assertion import AssertionState
 
@@ -1021,27 +1021,8 @@ def test_write_pyc(self, pytester: Pytester, tmp_path, monkeypatch) -> None:
         co = compile("1", "f.py", "single")
         assert _write_pyc(state, co, os.stat(source_path), pycpath)
 
-        if sys.platform == "win32":
-            from contextlib import contextmanager
-
-            @contextmanager
-            def atomic_write_failed(fn, mode="r", overwrite=False):
-                e = OSError()
-                e.errno = 10
-                raise e
-                yield  # type:ignore[unreachable]
-
-            monkeypatch.setattr(
-                _pytest.assertion.rewrite, "atomic_write", atomic_write_failed
-            )
-        else:
-
-            def raise_oserror(*args):
-                raise OSError()
-
-            monkeypatch.setattr("os.rename", raise_oserror)
-
-        assert not _write_pyc(state, co, os.stat(source_path), pycpath)
+        with mock.patch.object(os, "replace", side_effect=OSError):
+            assert not _write_pyc(state, co, os.stat(source_path), pycpath)
 
     def test_resources_provider_for_loader(self, pytester: Pytester) -> None:
         """