Drop atomicwrites dependency

changelog/10114.trivial.rst

@@ -0,0 +1 @@
+Dropped the dependency to `atomicwrites <https://github.com/untitaker/python-atomicwrites>`__ library.

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

@@ -303,7 +303,7 @@ def _write_pyc_fp(
 
 
 if sys.platform == "win32":
-    from atomicwrites import atomic_write
+    from _pytest.atomic_writes import atomic_write
 
     def _write_pyc(
         state: "AssertionState",

src/_pytest/atomic_writes.py

@@ -0,0 +1,205 @@
+"""
+Module copied over from https://github.com/untitaker/python-atomicwrites, which has become
+unmaintained.
+
+Since then, we have made changes to simplify the code, focusing on pytest's use-case.
+"""
+import contextlib
+import os
+import sys
+import tempfile
+
+
+_proper_fsync = os.fsync
+
+if sys.platform != "win32":
+    import fcntl
+
+    if hasattr(fcntl, "F_FULLFSYNC"):
+
+        def _proper_fsync(fd):
+            # https://lists.apple.com/archives/darwin-dev/2005/Feb/msg00072.html
+            # https://developer.apple.com/library/mac/documentation/Darwin/Reference/ManPages/man2/fsync.2.html
+            # https://github.com/untitaker/python-atomicwrites/issues/6
+            fcntl.fcntl(fd, fcntl.F_FULLFSYNC)
+
+    def _sync_directory(directory):
+        # Ensure that filenames are written to disk
+        fd = os.open(directory, 0)
+        try:
+            _proper_fsync(fd)
+        finally:
+            os.close(fd)
+
+    def _replace_atomic(src, dst):
+        os.rename(src, dst)
+        _sync_directory(os.path.normpath(os.path.dirname(dst)))
+
+    def _move_atomic(src, dst):
+        os.link(src, dst)
+        os.unlink(src)
+
+        src_dir = os.path.normpath(os.path.dirname(src))
+        dst_dir = os.path.normpath(os.path.dirname(dst))
+        _sync_directory(dst_dir)
+        if src_dir != dst_dir:
+            _sync_directory(src_dir)
+
+else:
+    from ctypes import windll, WinError
+
+    _MOVEFILE_REPLACE_EXISTING = 0x1
+    _MOVEFILE_WRITE_THROUGH = 0x8
+    _windows_default_flags = _MOVEFILE_WRITE_THROUGH
+
+    def _handle_errors(rv):
+        if not rv:
+            raise WinError()
+
+    def _replace_atomic(src, dst):
+        _handle_errors(
+            windll.kernel32.MoveFileExW(
+                src,
+                dst,
+                _windows_default_flags | _MOVEFILE_REPLACE_EXISTING,
+            )
+        )
+
+    def _move_atomic(src, dst):
+        _handle_errors(windll.kernel32.MoveFileExW(src, dst, _windows_default_flags))
+
+
+def replace_atomic(src, dst):
+    """
+    Move ``src`` to ``dst``. If ``dst`` exists, it will be silently
+    overwritten.
+
+    Both paths must reside on the same filesystem for the operation to be
+    atomic.
+    """
+    return _replace_atomic(src, dst)
+
+
+def move_atomic(src, dst):
+    """
+    Move ``src`` to ``dst``. There might a timewindow where both filesystem
+    entries exist. If ``dst`` already exists, :py:exc:`FileExistsError` will be
+    raised.
+
+    Both paths must reside on the same filesystem for the operation to be
+    atomic.
+    """
+    return _move_atomic(src, dst)
+
+
+class AtomicWriter:
+    """
+    A helper class for performing atomic writes. Usage::
+
+        with AtomicWriter(path).open() as f:
+            f.write(...)
+
+    :param path: The destination filepath. May or may not exist.
+    :param mode: The filemode for the temporary file. This defaults to `wb` in
+        Python 2 and `w` in Python 3.
+    :param overwrite: If set to false, an error is raised if ``path`` exists.
+        Errors are only raised after the file has been written to.  Either way,
+        the operation is atomic.
+    :param open_kwargs: Keyword-arguments to pass to the underlying
+        :py:func:`open` call. This can be used to set the encoding when opening
+        files in text-mode.
+
+    If you need further control over the exact behavior, you are encouraged to
+    subclass.
+    """
+
+    def __init__(self, path, mode="w", overwrite=False, **open_kwargs):
+        if "a" in mode:
+            raise ValueError(
+                "Appending to an existing file is not supported, because that "
+                "would involve an expensive `copy`-operation to a temporary "
+                "file. Open the file in normal `w`-mode and copy explicitly "
+                "if that's what you're after."
+            )
+        if "x" in mode:
+            raise ValueError("Use the `overwrite`-parameter instead.")
+        if "w" not in mode:
+            raise ValueError("AtomicWriters can only be written to.")
+
+        self._path = os.fspath(path)
+        self._mode = mode
+        self._overwrite = overwrite
+        self._open_kwargs = open_kwargs
+
+    def open(self):
+        """Open the temporary file."""
+        return self._open(self.get_fileobject)
+
+    @contextlib.contextmanager
+    def _open(self, get_fileobject):
+        f = None  # make sure f exists even if get_fileobject() fails
+        try:
+            success = False
+            with get_fileobject(**self._open_kwargs) as f:
+                yield f
+                self.sync(f)
+            self.commit(f)
+            success = True
+        finally:
+            if not success:
+                try:
+                    self.rollback(f)
+                except Exception:
+                    pass
+
+    def get_fileobject(
+        self, suffix="", prefix=tempfile.gettempprefix(), dir=None, **kwargs
+    ):
+        """Return the temporary file to use."""
+        if dir is None:
+            dir = os.path.normpath(os.path.dirname(self._path))
+        descriptor, name = tempfile.mkstemp(suffix=suffix, prefix=prefix, dir=dir)
+        # io.open() will take either the descriptor or the name, but we need
+        # the name later for commit()/replace_atomic() and couldn't find a way
+        # to get the filename from the descriptor.
+        os.close(descriptor)
+        kwargs["mode"] = self._mode
+        kwargs["file"] = name
+        return open(**kwargs)
+
+    def sync(self, f):
+        """
+        Responsible for clearing as many file caches as possible before
+        commit.
+        """
+        f.flush()
+        _proper_fsync(f.fileno())
+
+    def commit(self, f):
+        """Move the temporary file to the target location."""
+        if self._overwrite:
+            replace_atomic(f.name, self._path)
+        else:
+            move_atomic(f.name, self._path)
+
+    def rollback(self, f):
+        """Clean up all temporary resources."""
+        os.unlink(f.name)
+
+
+def atomic_write(path, writer_cls=AtomicWriter, **cls_kwargs):
+    """
+    Simple atomic writes. This wraps :py:class:`AtomicWriter`::
+
+        with atomic_write(path) as f:
+            f.write(...)
+
+    :param path: The target path to write to.
+    :param writer_cls: The writer class to use. This parameter is useful if you
+        subclassed :py:class:`AtomicWriter` to change some behavior and want to
+        use that new subclass.
+
+    Additional keyword arguments are passed to the writer class. See
+    :py:class:`AtomicWriter`.
+    """
+    return writer_cls(path, **cls_kwargs).open()

testing/test_atomic_writes.py

@@ -0,0 +1,98 @@
+"""
+Module copied over from https://github.com/untitaker/python-atomicwrites, which has become
+unmaintained.
+
+Since then, we have made changes to simplify the code, focusing on pytest's use-case.
+"""
+import errno
+import os
+from pathlib import Path
+
+import pytest
+from _pytest.atomic_writes import atomic_write
+
+
+def test_atomic_write(tmp_path: Path) -> None:
+    fname = tmp_path.joinpath("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_text() == "hoho"
+    assert len(list(tmp_path.iterdir())) == 1
+
+
+def test_teardown(tmp_path: Path) -> None:
+    fname = tmp_path.joinpath("ha")
+    with pytest.raises(AssertionError):
+        with atomic_write(str(fname), overwrite=True):
+            assert False
+
+    assert not list(tmp_path.iterdir())
+
+
+def test_replace_simultaneously_created_file(tmp_path: Path) -> None:
+    fname = tmp_path.joinpath("ha")
+    with atomic_write(str(fname), overwrite=True) as f:
+        f.write("hoho")
+        fname.write_text("harhar")
+        assert fname.read_text() == "harhar"
+    assert fname.read_text() == "hoho"
+    assert len(list(tmp_path.iterdir())) == 1
+
+
+def test_dont_remove_simultaneously_created_file(tmp_path: Path) -> None:
+    fname = tmp_path.joinpath("ha")
+    with pytest.raises(OSError) as excinfo:
+        with atomic_write(str(fname), overwrite=False) as f:
+            f.write("hoho")
+            fname.write_text("harhar")
+            assert fname.read_text() == "harhar"
+
+    assert excinfo.value.errno == errno.EEXIST
+    assert fname.read_text() == "harhar"
+    assert len(list(tmp_path.iterdir())) == 1
+
+
+# Verify that nested exceptions during rollback do not overwrite the initial
+# exception that triggered a rollback.
+def test_open_reraise(tmp_path: Path) -> None:
+    fname = tmp_path.joinpath("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(tmp_path: Path) -> None:
+    orig_curdir = os.getcwd()
+    try:
+        os.chdir(str(tmp_path))
+        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
+
+        with open(fname) as f:
+            assert f.read() == "hoho"
+        assert len(list(tmp_path.iterdir())) == 1
+    finally:
+        os.chdir(orig_curdir)