Bound advisory network retries while keeping audits fail closed

tests/test_ci_audit.py

@@ -0,0 +1,81 @@
+"""Network retries must never turn findings or incomplete audits into success."""
+
+import importlib.util
+import json
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+
+@pytest.fixture
+def audit(monkeypatch, tmp_path):
+    spec = importlib.util.spec_from_file_location(
+        "ci_audit", Path(__file__).parents[1] / "scripts/ci-audit.py"
+    )
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    monkeypatch.setattr(module.tempfile, "gettempdir", lambda: str(tmp_path))
+    monkeypatch.setattr(module.time, "sleep", lambda _: None)
+    monkeypatch.delenv("CI_AUDIT_TIMEOUT", raising=False)
+    monkeypatch.delenv("CI_AUDIT_ATTEMPTS", raising=False)
+    return module
+
+
+def result(code=0, stderr="", vulns=None, dependencies=None):
+    report = {
+        "dependencies": dependencies
+        if dependencies is not None
+        else [{"name": "example", "vulns": vulns or []}]
+    }
+    return SimpleNamespace(returncode=code, stderr=stderr, stdout=json.dumps(report))
+
+
+def test_transient_retry_then_complete(audit, monkeypatch):
+    replies = iter([result(1, "ReadTimeout"), result()])
+    calls = []
+
+    def run(command, **kwargs):
+        calls.append(command)
+        return next(replies)
+
+    monkeypatch.setattr(audit.subprocess, "run", run)
+    audit.main()
+    assert len(calls) == 2
+    assert calls[0] == calls[1]
+    assert "--cache-dir" in calls[0] and "--disable-pip" in calls[0]
+
+
+@pytest.mark.parametrize(
+    "reply",
+    [
+        result(1, vulns=[{"id": "CVE-test"}]),
+        result(dependencies=[]),
+        result(dependencies=[{"name": "example", "skip_reason": "unavailable"}]),
+        result(2),
+    ],
+)
+def test_no_retry_for_findings_or_incomplete(audit, monkeypatch, reply):
+    calls = []
+    monkeypatch.setattr(
+        audit.subprocess, "run", lambda *a, **k: calls.append(a) or reply
+    )
+    with pytest.raises(SystemExit):
+        audit.main()
+    assert len(calls) == 1
+
+
+def test_network_exhaustion_blocks_and_redacts(audit, monkeypatch, capsys):
+    calls = []
+    monkeypatch.setattr(
+        audit.subprocess,
+        "run",
+        lambda *a, **k: (
+            calls.append(a)
+            or result(1, "ConnectionError https://user:secret@example.invalid")
+        ),
+    )
+    with pytest.raises(SystemExit):
+        audit.main()
+    assert len(calls) == 3
+    assert "secret" not in capsys.readouterr().out