Many tests failing on Windows

pip_audit/_dependency_source/pyproject.py

@@ -142,8 +142,8 @@ def fix(self, fix_version: ResolvedFixVersion) -> None:
             # Now dump the new edited TOML to the temporary file.
             toml.dump(pyproject_data, tmp)
 
-            # And replace the original `pyproject.toml` file.
-            os.replace(tmp.name, self.filename)
+        # And replace the original `pyproject.toml` file.
+        os.replace(tmp.name, self.filename)
 
 
 class PyProjectSourceError(DependencySourceError):

pip_audit/_dependency_source/requirement.py

@@ -5,7 +5,6 @@
 from __future__ import annotations
 
 import logging
-import os
 import re
 import shutil
 from contextlib import ExitStack
@@ -193,7 +192,7 @@ def fix(self, fix_version: ResolvedFixVersion) -> None:
             # Make temporary copies of the existing requirements files. If anything goes wrong, we
             # want to copy them back into place and undo any partial application of the fix.
             tmp_files: list[IO[str]] = [
-                stack.enter_context(NamedTemporaryFile(mode="w")) for _ in self._filenames
+                stack.enter_context(NamedTemporaryFile(mode="r+")) for _ in self._filenames
             ]
             for filename, tmp_file in zip(self._filenames, tmp_files):
                 with filename.open("r") as f:
@@ -220,7 +219,7 @@ def _fix_file(self, filename: Path, fix_version: ResolvedFixVersion) -> None:
         #
         # This time we're using the `RequirementsFile.parse` API instead of `Requirements.from_file`
         # since we want to access each line sequentially in order to rewrite the file.
-        reqs = list(RequirementsFile.parse(filename=str(filename)))
+        reqs = list(RequirementsFile.parse(filename=filename.as_posix()))
 
         # Check ahead of time for anything invalid in the requirements file since we don't want to
         # encounter this while writing out the file. Check for duplicate requirements and lines that
@@ -278,10 +277,9 @@ def _fix_file(self, filename: Path, fix_version: ResolvedFixVersion) -> None:
     def _recover_files(self, tmp_files: list[IO[str]]) -> None:
         for filename, tmp_file in zip(self._filenames, tmp_files):
             try:
-                os.replace(tmp_file.name, filename)
-                # We need to tinker with the internals to prevent the file wrapper from attempting
-                # to remove the temporary file like in the regular case.
-                tmp_file._closer.delete = False  # type: ignore[attr-defined]
+                tmp_file.seek(0)
+                with filename.open("w") as f:
+                    shutil.copyfileobj(tmp_file, f)
             except Exception as e:
                 # Not much we can do at this point since we're already handling an exception. Just
                 # log the error and try to recover the rest of the files.

test/conftest.py

@@ -76,7 +76,7 @@ def fix(self, _) -> None:
 @pytest.fixture(scope="session")
 def cache_dir():
     cache = tempfile.TemporaryDirectory()
-    yield cache.name
+    yield Path(cache.name)
     cache.cleanup()
 
 

test/dependency_source/test_requirement.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import os
+import sys
 from email.message import EmailMessage
 from pathlib import Path
 from tempfile import NamedTemporaryFile, TemporaryDirectory
@@ -121,13 +122,13 @@ def test_requirement_source_git(req_file):
         [
             (
                 req_file(),
-                "git+https://github.com/unbit/uwsgi.git@1bb9ad77c6d2d310c2d6d1d9ad62de61f725b824",
+                "git+https://github.com/benoitc/gunicorn.git@61ccfd6c38d477a908e0f376757bbb884438053a",
             )
         ]
     )
 
     specs = list(source.collect())
-    assert ResolvedDependency(name="uWSGI", version=Version("2.0.20")) in specs
+    assert ResolvedDependency(name="gunicorn", version=Version("20.1.0")) in specs
 
 
 @pytest.mark.online
@@ -378,7 +379,9 @@ def test_requirement_source_fix_rollback_failure(monkeypatch, req_file):
     def mock_replace(*_args, **_kwargs):
         raise OSError
 
-    monkeypatch.setattr(os, "replace", mock_replace)
+    from tempfile import _TemporaryFileWrapper
+
+    monkeypatch.setattr(_TemporaryFileWrapper, "seek", mock_replace, raising=False)
 
     source = requirement.RequirementSource(req_paths)
     with pytest.raises(DependencyFixError):
@@ -559,6 +562,7 @@ def virtual_env(*args, **kwargs):
     assert ResolvedDependency("Flask", Version("2.0.1")) in specs
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="os.mkfifo does not exists on windows")
 def test_requirement_source_fifo():
     with TemporaryDirectory() as tmp_dir:
         fifo_path = Path(os.path.join(tmp_dir, "fifo"))

test/test_cache.py

@@ -10,14 +10,14 @@
 def test_get_cache_dir(monkeypatch):
     # When we supply a cache directory, always use that
     cache_dir = _get_cache_dir(Path("/tmp/foo/cache_dir"))
-    assert str(cache_dir) == "/tmp/foo/cache_dir"
+    assert cache_dir.as_posix() == "/tmp/foo/cache_dir"
 
-    get_pip_cache = pretend.call_recorder(lambda: "/fake/pip/cache/dir")
+    get_pip_cache = pretend.call_recorder(lambda: Path("/fake/pip/cache/dir"))
     monkeypatch.setattr(cache, "_get_pip_cache", get_pip_cache)
 
     # When `pip cache dir` works, we use it. In this case, it's mocked.
     cache_dir = _get_cache_dir(None, use_pip=True)
-    assert str(cache_dir) == "/fake/pip/cache/dir"
+    assert cache_dir.as_posix() == "/fake/pip/cache/dir"
 
 
 def test_get_pip_cache():
@@ -45,7 +45,7 @@ def test_get_cache_dir_old_pip(monkeypatch):
 
     # When we supply a cache directory, always use that
     cache_dir = _get_cache_dir(Path("/tmp/foo/cache_dir"))
-    assert str(cache_dir) == "/tmp/foo/cache_dir"
+    assert cache_dir.as_posix() == "/tmp/foo/cache_dir"
 
     # In this case, we can't query `pip` to figure out where its HTTP cache is
     # Instead, we use `~/.pip-audit-cache`

test/test_cli.py

@@ -25,9 +25,9 @@ def test_str(self):
 
 
 class TestVulnerabilityServiceChoice:
-    def test_to_service_is_exhaustive(self):
+    def test_to_service_is_exhaustive(self, cache_dir):
         for choice in VulnerabilityServiceChoice:
-            assert choice.to_service(0, pretend.stub()) is not None
+            assert choice.to_service(0, cache_dir) is not None
 
     def test_str(self):
         for choice in VulnerabilityServiceChoice: