tegwick
bbbede5f47
check_all runs every check stage even when one fails; malformed dates are reported rather than aborting; accepted findings and closure evidence are shown; defer requires a valid future date. Adds SCOPE.md, the scope assessment, the open-findings source review and a unittest suite. Stops tracking __pycache__. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 6903@bnt-lap001 Assistant-Session: 8319e8a8-ffa6-4eb3-b8bf-b29945628f89
157 lines
8.8 KiB
Python
157 lines
8.8 KiB
Python
import contextlib
|
|
import datetime as dt
|
|
import io
|
|
import pathlib
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1] / "tools"))
|
|
import check_all
|
|
import fix_tracker
|
|
import record_check
|
|
import register_check
|
|
import register_lib as lib
|
|
|
|
NOW = dt.datetime(2026, 9, 5, tzinfo=dt.timezone.utc)
|
|
|
|
|
|
class ReviewWorkflowTests(unittest.TestCase):
|
|
def report(self, findings=(), regulatory=()):
|
|
output = io.StringIO()
|
|
with patch.object(lib, "findings", return_value=list(findings)), \
|
|
patch.object(lib, "regulatory", return_value=list(regulatory)), \
|
|
patch.object(lib, "records_with_waits", return_value=[]), \
|
|
patch.object(fix_tracker, "report", return_value=[]), \
|
|
patch.object(register_check, "NOW", NOW), contextlib.redirect_stdout(output):
|
|
self.assertEqual(register_check.main(), 0)
|
|
return output.getvalue()
|
|
|
|
def test_all_stages_run_after_failure(self):
|
|
for failed_stage in range(3):
|
|
results = [subprocess.CompletedProcess([], int(i == failed_stage)) for i in range(3)]
|
|
with patch.object(check_all.subprocess, "run", side_effect=results) as run, \
|
|
contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertEqual(check_all.main(), 1)
|
|
self.assertEqual(run.call_count, 3)
|
|
self.assertTrue(run.call_args_list[-1].args[0][1].endswith("inbox_check.py"))
|
|
|
|
def test_all_success_and_spawn_failure(self):
|
|
with patch.object(check_all.subprocess, "run", return_value=subprocess.CompletedProcess([], 0)), \
|
|
contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertEqual(check_all.main(), 0)
|
|
with patch.object(check_all.subprocess, "run", side_effect=[OSError("unavailable"),
|
|
subprocess.CompletedProcess([], 0), subprocess.CompletedProcess([], 0)]) as run, \
|
|
contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertEqual(check_all.main(), 1)
|
|
self.assertEqual(run.call_count, 3)
|
|
|
|
def test_nested_policy_discovered_reported_and_reviewable(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = pathlib.Path(directory)
|
|
policies = root / "docs/regulatory/policies"
|
|
policies.mkdir(parents=True)
|
|
(policies / "README.md").write_text("Catalogue without front-matter")
|
|
policy = policies / "example.md"
|
|
policy.write_text('---\nid: RISK-POL-9999\nnext_check: "2026-08-20"\n'
|
|
'cadence: instant\n---\n\n# Policy\n')
|
|
with patch.object(lib, "REPO", root):
|
|
records = lib.regulatory()
|
|
self.assertEqual([r["id"] for r in records], ["RISK-POL-9999"])
|
|
self.assertIn("RISK-POL-9999 — due", self.report(regulatory=records))
|
|
with patch.object(lib, "findings", return_value=[]), patch.object(lib, "now", return_value=NOW), \
|
|
contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertEqual(record_check.main(["RISK-POL-9999", "clean", "review evidence"]), 0)
|
|
reviewed = lib.load(policy)
|
|
self.assertEqual(reviewed["cadence"], "1h")
|
|
self.assertIn("review evidence", policy.read_text())
|
|
|
|
def test_invalid_dates_do_not_abort_reports(self):
|
|
finding = dict(id="RISK-F-9999", _path=pathlib.Path("test.md"), status="open",
|
|
next_check="not-a-date", severity="high", cadence="instant")
|
|
report = self.report([finding], [dict(id="RISK-POL-9999", next_check="bad")])
|
|
self.assertIn("next_check is not a date", report)
|
|
self.assertIn("RISK-POL-9999 — invalid date", report)
|
|
self.assertIn("Publication handovers pending", report)
|
|
self.assertIsNone(lib.moment("2026-99-99"))
|
|
self.assertEqual(lib.moment("2026-09-05T00:00:00"), NOW)
|
|
|
|
def test_regulatory_deferral_expires_and_bad_date_stays_visible(self):
|
|
policy = dict(id="RISK-POL-9999", next_check="2026-08-20", deferred_to="2026-09-06")
|
|
self.assertIn("Regulatory records due:\n none", self.report(regulatory=[policy]))
|
|
policy["deferred_to"] = "2026-09-04"
|
|
self.assertIn("RISK-POL-9999 — due", self.report(regulatory=[policy]))
|
|
policy["deferred_to"] = "bad"
|
|
self.assertIn("invalid date: deferred_to", self.report(regulatory=[policy]))
|
|
|
|
def test_disclosure_obligations_survive_closure(self):
|
|
findings = [dict(id="RISK-F-9998", _path=pathlib.Path("a.md"), status="fixed",
|
|
disclosure="embargoed", embargo_condition="owner verifies", embargo_review="2026-09-01"),
|
|
dict(id="RISK-F-9999", _path=pathlib.Path("b.md"), status="fixed",
|
|
disclosure="public", publication="pending-handover")]
|
|
report = self.report(findings)
|
|
self.assertIn("RISK-F-9998 — embargo review due", report)
|
|
self.assertIn("RISK-F-9999 (fixed) — pending policy-nexus", report)
|
|
findings[0].pop("embargo_review")
|
|
self.assertIn("missing or invalid embargo_review", self.report(findings))
|
|
findings[0]["embargo_review"] = "2027-01-01"
|
|
self.assertIn("Embargo reviews due:\n none", self.report(findings))
|
|
|
|
def test_bad_deferral_does_not_write_and_review_clears_old_deferral(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = pathlib.Path(directory) / "record.md"
|
|
original = '---\nid: RISK-F-9999\ncadence: instant\ndeferred_to: "2027-01-01"\n---\n'
|
|
path.write_text(original)
|
|
with patch.object(lib, "findings", side_effect=lambda: [lib.load(path)]), \
|
|
patch.object(lib, "regulatory", return_value=[]), patch.object(lib, "now", return_value=NOW), \
|
|
contextlib.redirect_stdout(io.StringIO()):
|
|
for value in ("bad", "2020-01-01"):
|
|
with self.assertRaises(SystemExit):
|
|
record_check.main(["RISK-F-9999", "defer", value, "operator reason"])
|
|
self.assertEqual(path.read_text(), original)
|
|
record_check.main(["RISK-F-9999", "moved", "new evidence"])
|
|
self.assertNotIn("deferred_to", lib.load(path))
|
|
|
|
def test_archived_completed_fix_is_not_stalled(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = pathlib.Path(directory)
|
|
archive = root / "owner/workplans/archived"
|
|
archive.mkdir(parents=True)
|
|
(archive / "260901-TEST-WP-0001-fix.md").write_text(
|
|
'---\nid: TEST-WP-0001\nstatus: finished\nrepo: owner\n---\n')
|
|
finding = dict(id="RISK-F-9999", status="open", severity="high", fix_tracking="TEST-WP-0001")
|
|
with patch.object(fix_tracker, "ESTATE", root), patch.object(lib, "findings", return_value=[finding]), \
|
|
patch.object(lib, "now", return_value=NOW), \
|
|
patch.object(fix_tracker, "last_commit", return_value=NOW - dt.timedelta(days=100)):
|
|
report = "\n".join(fix_tracker.report())
|
|
self.assertIn("finished [archived]", report)
|
|
self.assertNotIn("UNCHANGED", report)
|
|
|
|
def test_accepted_obligation_requires_a_resolvable_basis_and_review(self):
|
|
finding = dict(id="RISK-F-9999", status="accepted", fix_tracking="unset",
|
|
accepted_by="operator", accepted_until="real data arrives",
|
|
determination="RISK-REG-9999", next_check="2026-08-01")
|
|
with patch.object(lib, "findings", return_value=[finding]), \
|
|
patch.object(lib, "regulatory", return_value=[dict(id="RISK-REG-9999")]):
|
|
report = "\n".join(fix_tracker.report())
|
|
self.assertIn("RECORDED ACCEPTANCE", report)
|
|
self.assertIn("not a fix or renewed acceptance", report)
|
|
self.assertNotIn("NO FIX TRACKING", report)
|
|
finding["next_check"] = "bad"
|
|
self.assertIn("ACCEPTANCE INCOMPLETE: next_check", "\n".join(fix_tracker.report()))
|
|
self.assertIn("NO FIX TRACKING", "\n".join(fix_tracker.report()))
|
|
finding["next_check"] = "2026-08-01"
|
|
with patch.object(lib, "findings", return_value=[finding]), patch.object(lib, "regulatory", return_value=[]):
|
|
self.assertIn("resolvable determination", "\n".join(fix_tracker.report()))
|
|
|
|
def test_completed_source_does_not_hide_closure_evidence(self):
|
|
finding = dict(id="RISK-F-9999", _path=pathlib.Path("test.md"), status="open",
|
|
fix_tracking="TEST-WP-0001 (finished)", closure_condition="runtime acceptance receipt")
|
|
report = self.report([finding])
|
|
self.assertIn("Closure evidence pending:\n RISK-F-9999 — runtime acceptance receipt", report)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|