"""Tests for MCP restart lifecycle audit events and incidents (#665).""" from __future__ import annotations import json import os import tempfile import unittest from unittest.mock import patch import gitea_audit import restart_audit as ra class TestEventSchema(unittest.TestCase): def test_all_lifecycle_event_types_are_named(self): expected = { "mcp.restart.impact_preview", "mcp.restart.drain_enter", "mcp.restart.drain_exit", "mcp.restart.drain_proof", "mcp.restart.apply_gate", "mcp.restart.break_glass", "mcp.restart.post_restart_reconcile", "mcp.restart.narrower_recovery", "mcp.restart.unguarded_detected", } self.assertEqual(set(ra.RESTART_EVENT_TYPES), expected) def test_build_restart_event_core_fields(self): event = ra.build_restart_event( event_type=ra.EVENT_IMPACT_PREVIEW, outcome="safe", correlation_id="rst-abc123", remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", requesting_session_id="sess-1", restart_class="full_mcp_restart", profile_name="prgs-author", authenticated_username="bot", reasons=["inventory complete"], details={"allow_restart": True}, now="2026-07-25T12:00:00+00:00", ) self.assertEqual(event["event_type"], ra.EVENT_IMPACT_PREVIEW) self.assertEqual(event["action"], ra.EVENT_IMPACT_PREVIEW) self.assertEqual(event["action_type"], "restart_lifecycle") self.assertEqual(event["result"], "safe") self.assertEqual(event["correlation_id"], "rst-abc123") self.assertEqual(event["profile_name"], "prgs-author") self.assertEqual(event["authenticated_username"], "bot") meta = event["request_metadata"] self.assertEqual(meta["event_family"], "mcp.restart") self.assertEqual(meta["correlation_id"], "rst-abc123") self.assertEqual(meta["restart_class"], "full_mcp_restart") self.assertEqual(meta["details"]["allow_restart"], True) def test_unknown_event_type_raises(self): with self.assertRaises(ValueError) as ctx: ra.build_restart_event( event_type="mcp.restart.not_a_real_event", outcome="x", correlation_id="rst-1", ) self.assertIn("unknown restart audit event_type", str(ctx.exception)) def test_reasons_and_details_are_redacted(self): event = ra.build_restart_event( event_type=ra.EVENT_APPLY_GATE, outcome="deny", correlation_id="rst-sec", reasons=["token secret-xyz rejected", "ok"], details={"token": "leak-token", "status": "denied"}, ) self.assertNotIn("secret-xyz", event.get("reason") or "") meta = event["request_metadata"] self.assertEqual(meta["details"]["token"], gitea_audit.REDACTED) self.assertEqual(meta["details"]["status"], "denied") for reason in meta["reasons"]: self.assertNotIn("secret-xyz", reason) def test_new_correlation_id_shape(self): cid = ra.new_correlation_id() self.assertTrue(cid.startswith("rst-")) self.assertEqual(len(cid), len("rst-") + 16) class TestEmitAndRequire(unittest.TestCase): def test_emit_appends_json_line(self): with tempfile.TemporaryDirectory() as d: path = os.path.join(d, "audit.log") event = ra.build_restart_event( event_type=ra.EVENT_DRAIN_PROOF, outcome="pass", correlation_id="rst-write", ) self.assertTrue(ra.emit_restart_event(event, path=path)) with open(path, encoding="utf-8") as fh: lines = fh.read().splitlines() self.assertEqual(len(lines), 1) loaded = json.loads(lines[0]) self.assertEqual(loaded["event_type"], ra.EVENT_DRAIN_PROOF) self.assertEqual(loaded["correlation_id"], "rst-write") def test_emit_never_raises(self): self.assertFalse( ra.emit_restart_event({"action": "x"}, path="/no/such/dir/audit.log") ) def test_require_audit_denies_privileged_when_write_fails_and_enabled(self): deny = ra.require_audit_or_deny( privileged=True, written=False, audit_enabled=True ) self.assertEqual(len(deny), 1) self.assertIn("fail closed", deny[0]) def test_require_audit_allows_when_audit_disabled(self): # Rollout policy: enable audit before enforcing deny-on-audit-fail. deny = ra.require_audit_or_deny( privileged=True, written=False, audit_enabled=False ) self.assertEqual(deny, []) def test_require_audit_noop_for_non_privileged(self): deny = ra.require_audit_or_deny( privileged=False, written=False, audit_enabled=True ) self.assertEqual(deny, []) def test_require_audit_allows_when_written(self): deny = ra.require_audit_or_deny( privileged=True, written=True, audit_enabled=True ) self.assertEqual(deny, []) class TestIncidents(unittest.TestCase): def test_break_glass_descriptor(self): desc = ra.build_incident_descriptor( kind=ra.INCIDENT_BREAK_GLASS, reasons=["break-glass authorized"], correlation_id="rst-bg", requesting_session_id="s1", restart_class="full_mcp_restart", remote="prgs", org="O", repo="R", ) self.assertEqual(desc["kind"], ra.INCIDENT_BREAK_GLASS) self.assertIn("Break-glass", desc["title"]) self.assertIn("mcp-health", desc["labels"]) self.assertEqual(desc["source"], "restart_audit#665") def test_incident_body_redacts_and_includes_correlation(self): desc = ra.build_incident_descriptor( kind=ra.INCIDENT_FAILED_DRAIN, reasons=["token secret-xyz failed proof"], correlation_id="rst-body", remote="prgs", org="O", repo="R", proof_id="proof-1", ) body = ra.incident_body(desc) self.assertIn("rst-body", body) self.assertIn("proof-1", body) self.assertIn("mcp-restart-incident:v1", body) self.assertNotIn("secret-xyz", body) def test_materialize_dry_run(self): desc = ra.build_incident_descriptor( kind=ra.INCIDENT_FAILED_DRAIN, reasons=["denied"], correlation_id="rst-dr", ) result = ra.materialize_incident(desc, create_issue_fn=lambda **k: {}, dry_run=True) self.assertFalse(result["created"]) self.assertTrue(result["dry_run"]) self.assertIn("dry-run", result["reasons"][0]) def test_materialize_without_create_fn(self): desc = ra.build_incident_descriptor( kind=ra.INCIDENT_FAILED_DRAIN, reasons=["denied"], correlation_id="rst-nfn", ) result = ra.materialize_incident(desc, create_issue_fn=None) self.assertFalse(result["created"]) self.assertIn("create_issue_fn not provided", result["reasons"][0]) def test_materialize_creates_issue(self): created = {} def _create(*, title, body, labels, org=None, repo=None, **_kw): created["title"] = title created["body"] = body created["labels"] = labels created["org"] = org created["repo"] = repo return {"number": 999} desc = ra.build_incident_descriptor( kind=ra.INCIDENT_BREAK_GLASS, reasons=["break-glass"], correlation_id="rst-create", org="O", repo="R", ) result = ra.materialize_incident(desc, create_issue_fn=_create) self.assertTrue(result["created"]) self.assertEqual(result["issue_number"], 999) self.assertIn("Break-glass", created["title"]) self.assertIn("rst-create", created["body"]) self.assertEqual(created["org"], "O") def test_materialize_never_raises_on_create_failure(self): def _boom(**_kw): raise RuntimeError("token secret-xyz network") desc = ra.build_incident_descriptor( kind=ra.INCIDENT_FAILED_DRAIN, reasons=["x"], correlation_id="rst-boom", ) result = ra.materialize_incident(desc, create_issue_fn=_boom) self.assertFalse(result["created"]) self.assertIn("failed", result["reasons"][0]) self.assertNotIn("secret-xyz", result["reasons"][0]) class TestIncidentFromApplyGate(unittest.TestCase): def test_break_glass_always_incident(self): desc = ra.incident_from_apply_gate( gate_payload={"allow": True, "reasons": [], "proof_id": None}, break_glass=True, correlation_id="rst-bg2", requesting_session_id="s", restart_class="full_mcp_restart", remote="prgs", org="O", repo="R", ) self.assertIsNotNone(desc) self.assertEqual(desc["kind"], ra.INCIDENT_BREAK_GLASS) def test_failed_drain_from_gate_incident(self): desc = ra.incident_from_apply_gate( gate_payload={ "allow": False, "drain_gate_allow": False, "reasons": ["proof expired"], "incident": { "reasons": ["proof expired"], "proof_id": "p1", }, }, break_glass=False, correlation_id="rst-fd", requesting_session_id="s", restart_class="full_mcp_restart", remote="prgs", org="O", repo="R", ) self.assertIsNotNone(desc) self.assertEqual(desc["kind"], ra.INCIDENT_FAILED_DRAIN) self.assertEqual(desc["proof_id"], "p1") def test_allow_without_break_glass_no_incident(self): desc = ra.incident_from_apply_gate( gate_payload={ "allow": True, "drain_gate_allow": True, "reasons": [], }, break_glass=False, correlation_id="rst-ok", requesting_session_id="s", restart_class="full_mcp_restart", remote="prgs", org="O", repo="R", ) self.assertIsNone(desc) class TestRecordLifecycle(unittest.TestCase): def test_record_emits_and_materializes(self): created = [] def _create(**kwargs): created.append(kwargs) return {"number": 42} with tempfile.TemporaryDirectory() as d: path = os.path.join(d, "audit.log") with patch.dict(os.environ, {"GITEA_AUDIT_LOG": path}, clear=False): incident = ra.build_incident_descriptor( kind=ra.INCIDENT_BREAK_GLASS, reasons=["bg"], correlation_id="rst-lc", org="O", repo="R", ) out = ra.record_restart_lifecycle( event_type=ra.EVENT_BREAK_GLASS, outcome="break_glass", correlation_id="rst-lc", remote="prgs", org="O", repo="R", privileged=True, create_incident=incident, create_issue_fn=_create, audit_path=path, ) self.assertTrue(out["audit_written"]) self.assertEqual(out["deny_reasons"], []) self.assertTrue(out["incident_result"]["created"]) self.assertEqual(out["incident_result"]["issue_number"], 42) self.assertEqual(len(created), 1) def test_privileged_deny_when_audit_write_fails(self): with patch.dict( os.environ, {"GITEA_AUDIT_LOG": "/no/such/dir/a.log"}, clear=False ): with patch("restart_audit.emit_restart_event", return_value=False): with patch("gitea_audit.audit_enabled", return_value=True): out = ra.record_restart_lifecycle( event_type=ra.EVENT_APPLY_GATE, outcome="deny", correlation_id="rst-deny", privileged=True, audit_path="/no/such/dir/a.log", ) self.assertFalse(out["audit_written"]) self.assertEqual(len(out["deny_reasons"]), 1) if __name__ == "__main__": unittest.main()