fix(report): clean up temp file when the atomic write fails

This commit is contained in:
poria-lang 2026-08-10 20:39:41 +03:00
parent a2951991dd
commit b950bb6c72
2 changed files with 72 additions and 16 deletions

View file

@ -179,17 +179,24 @@ def write_vulnerabilities(
def _atomic_write_text(path: Path, payload: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with tempfile.NamedTemporaryFile(
mode="w",
encoding="utf-8",
dir=str(path.parent),
prefix=f".{path.name}.",
suffix=".tmp",
delete=False,
) as tmp:
tmp.write(payload)
tmp_path = Path(tmp.name)
tmp_path.replace(path)
tmp_path: Path | None = None
try:
with tempfile.NamedTemporaryFile(
mode="w",
encoding="utf-8",
dir=str(path.parent),
prefix=f".{path.name}.",
suffix=".tmp",
delete=False,
) as tmp:
tmp_path = Path(tmp.name)
tmp.write(payload)
assert tmp_path is not None
tmp_path.replace(path)
except Exception: # leave no temp file behind on any failure
if tmp_path is not None:
tmp_path.unlink(missing_ok=True)
raise
def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PLR0915

View file

@ -5,7 +5,8 @@ from __future__ import annotations
import csv
import json
import tempfile
from typing import TYPE_CHECKING, Any
from pathlib import Path
from typing import Any, Self
import pytest
@ -18,10 +19,6 @@ from strix.report.writer import (
)
if TYPE_CHECKING:
from pathlib import Path
def _sample_report(**overrides: Any) -> dict[str, Any]:
base: dict[str, Any] = {
"id": "vuln-0001",
@ -197,3 +194,55 @@ def test_write_executive_report_failure_preserves_previous_report(
write_executive_report(tmp_path, "new content")
assert target.read_text(encoding="utf-8") == "previous report"
def test_write_executive_report_write_failure_preserves_previous_report(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
target = tmp_path / "penetration_test_report.md"
target.write_text("previous report", encoding="utf-8")
temp_path = tmp_path / ".penetration_test_report.md.stub.tmp"
temp_path.touch()
class _FailingTempFile:
name = str(temp_path)
def __enter__(self) -> Self:
return self
def __exit__(self, *_exc: object) -> None:
return None
def write(self, _payload: str) -> int:
raise OSError("disk full")
monkeypatch.setattr(
tempfile,
"NamedTemporaryFile",
lambda *_args, **_kwargs: _FailingTempFile(),
)
with pytest.raises(OSError, match="disk full"):
write_executive_report(tmp_path, "new content")
assert target.read_text(encoding="utf-8") == "previous report"
assert not temp_path.exists() # the partial temp file is cleaned up
def test_write_executive_report_replace_failure_preserves_previous_report(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
target = tmp_path / "penetration_test_report.md"
target.write_text("previous report", encoding="utf-8")
def _failing_replace(_self: Path, _destination: Path) -> None:
raise OSError("disk full")
monkeypatch.setattr(Path, "replace", _failing_replace)
with pytest.raises(OSError, match="disk full"):
write_executive_report(tmp_path, "new content")
assert target.read_text(encoding="utf-8") == "previous report"
assert not list(tmp_path.glob(".*.tmp")) # no orphaned temp files