fix(reports): preserve previous exports when a write fails (#154)

* fix(reports): preserve previous exports when a write fails

Signed-off-by: Rudy Celekli <47457359+rudycelekli@users.noreply.github.com>

* fix(reports): preserve normal permissions for new exports

Signed-off-by: Rudy Celekli <47457359+rudycelekli@users.noreply.github.com>

---------

Signed-off-by: Rudy Celekli <47457359+rudycelekli@users.noreply.github.com>
Co-authored-by: n-papaioannou <n.papaioannou@ime.life>

Author: rudycelekli

ifixai/cli/reports.py

@@ -1,5 +1,9 @@
+import os
 import re
+import stat
+from contextlib import suppress
 from pathlib import Path
+from uuid import uuid4
 
 import click
 
@@ -19,9 +23,33 @@ def _slugify(value: str) -> str:
     return cleaned or "unknown"
 
 
+def _write_report_atomic(path: Path, content: str) -> None:
+    mode = stat.S_IMODE(path.stat().st_mode) if path.exists() else None
+    temporary = path.parent / f".ifixai-report-{uuid4().hex}.tmp"
+    # New exports follow normal file-creation permissions, including the umask.
+    # Replacements stay private until the existing target mode is restored.
+    descriptor = os.open(
+        temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL,
+        0o666 if mode is None else 0o600,
+    )
+    os.close(descriptor)
+    try:
+        temporary.write_text(content, encoding="utf-8")
+        if mode is not None:
+            temporary.chmod(mode)
+        os.replace(temporary, path)
+    finally:
+        with suppress(OSError):
+            temporary.unlink(missing_ok=True)
+
+
 def save_reports(
-    result: TestRunResult, output_dir: str, report_format: str, run_nonce: str | None = None,
-    *, run_id: str | None = None,
+    result: TestRunResult,
+    output_dir: str,
+    report_format: str,
+    run_nonce: str | None = None,
+    *,
+    run_id: str | None = None,
 ) -> None:
     out_path = Path(output_dir)
     out_path.mkdir(parents=True, exist_ok=True)
@@ -39,14 +67,14 @@ def save_reports(
 
     if report_format in ("markdown", "both"):
         summary_path = out_path / f"{base_name}-summary.md"
-        summary_path.write_text(generate_summary_report(result), encoding="utf-8")
+        _write_report_atomic(summary_path, generate_summary_report(result))
         click.echo(f"  Summary (start here): {summary_path}")
 
         md_path = out_path / f"{base_name}.md"
-        md_path.write_text(generate_markdown_report(result), encoding="utf-8")
+        _write_report_atomic(md_path, generate_markdown_report(result))
         click.echo(f"  Full report:          {md_path}")
 
     if report_format in ("json", "both"):
         json_path = out_path / f"{base_name}.json"
-        json_path.write_text(generate_json_report(result), encoding="utf-8")
+        _write_report_atomic(json_path, generate_json_report(result))
         click.echo(f"  JSON (machine):       {json_path}")

ifixai/tests/test_report_atomicity_native.py

@@ -0,0 +1,106 @@
+import errno
+from pathlib import Path
+
+import pytest
+
+from ifixai.cli.reports import save_reports
+from ifixai.core.types import TestRunResult
+
+
+@pytest.mark.parametrize(
+    "report_format,suffix", [("json", ".json"), ("markdown", "-summary.md")]
+)
+def test_failed_report_write_preserves_the_previous_artifact(
+    tmp_path, monkeypatch, report_format, suffix
+):
+    result = TestRunResult(system_name="agent", fixture_name="support")
+    target = tmp_path / f"ifixai-agent-support{suffix}"
+    old = b"previous successful report\n"
+    target.write_bytes(old)
+    original = Path.write_text
+    writes = []
+
+    def fail_during_actual_write(path, data, *args, **kwargs):
+        writes.append(path)
+        with path.open("w", encoding="utf-8") as handle:
+            handle.write(data[:10])
+            handle.flush()
+        raise OSError(errno.ENOSPC, "owned injected disk-full failure")
+
+    monkeypatch.setattr(Path, "write_text", fail_during_actual_write)
+    with pytest.raises(OSError) as exc:
+        save_reports(result, str(tmp_path), report_format)
+    assert exc.value.errno == errno.ENOSPC
+    assert writes
+    assert target.read_bytes() == old
+    assert list(tmp_path.iterdir()) == [target]
+    monkeypatch.setattr(Path, "write_text", original)
+
+
+def test_successful_json_report_still_publishes_normally(tmp_path):
+    import json
+
+    save_reports(
+        TestRunResult(system_name="agent", fixture_name="support"),
+        str(tmp_path),
+        "json",
+    )
+    assert (
+        json.loads((tmp_path / "ifixai-agent-support.json").read_text())["metadata"][
+            "system_name"
+        ]
+        == "agent"
+    )
+
+
+def test_successful_replacement_preserves_existing_report_permissions(tmp_path):
+    import os
+    import stat
+
+    if os.name == "nt":
+        pytest.skip("POSIX permission bits")
+    target = tmp_path / "ifixai-agent-support.json"
+    target.write_text("old")
+    target.chmod(0o640)
+    save_reports(
+        TestRunResult(system_name="agent", fixture_name="support"),
+        str(tmp_path),
+        "json",
+    )
+    assert stat.S_IMODE(target.stat().st_mode) == 0o640
+    assert len(list(tmp_path.iterdir())) == 1
+
+
+@pytest.mark.parametrize("mask", ["022", "027", "077"])
+def test_new_reports_follow_normal_file_creation_umask(tmp_path, mask):
+    import os
+    import subprocess
+    import sys
+
+    if os.name == "nt":
+        pytest.skip("POSIX permission bits")
+    # Change the umask only in this owned child, never the pytest process.
+    script = """
+import os
+import stat
+import sys
+from pathlib import Path
+from ifixai.cli.reports import save_reports
+from ifixai.core.types import TestRunResult
+mask = int(sys.argv[2], 8)
+os.umask(mask)
+root = Path(sys.argv[1])
+save_reports(TestRunResult(system_name="agent", fixture_name="support"), str(root), "both")
+reports = list(root.iterdir())
+assert len(reports) == 3
+for report in reports:
+    mode = stat.S_IMODE(report.stat().st_mode)
+    assert mode == 0o666 & ~mask, (report.name, oct(mode), oct(0o666 & ~mask))
+"""
+    result = subprocess.run(
+        [sys.executable, "-c", script, str(tmp_path), mask],
+        env={"PATH": os.environ.get("PATH", ""),
+             "PYTHONPATH": os.environ.get("PYTHONPATH", "")},
+        capture_output=True, text=True, check=False,
+    )
+    assert result.returncode == 0, result.stdout + result.stderr