"""有界自动审计循环测试：纯状态机 + worker 显式入口。 只测本轨：确定性门循环、推回包完整性、标准不弱化、上限升级、 全量回归、主观不循环、计数损坏 fail-closed、默认关闭。 """ import json import os import sqlite3 import sys import tempfile import unittest sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..")) from cloud import audit_loop from cloud.audit_loop import Finding, LoopState, evaluate from cloud.worker import Worker def det(fid, gate=None): return Finding(fid, "high", "evidence for %s" % fid, "re-run %s" % (gate or fid), "%s passes" % (gate or fid), gate=gate) class PureEngineTests(unittest.TestCase): def test_pass_all_gates(self): out = evaluate(None, "t1", {"tests": True, "capability": True}) self.assertEqual(out["decision"], "pass") self.assertEqual(out["round"], 0) self.assertIsNone(out["packet"]) def test_pushback_packet_complete(self): out = evaluate(None, "t1", {"tests": False}, [det("f1", "tests")]) self.assertEqual(out["decision"], "pushback") self.assertEqual(out["round"], 1) packet = out["packet"] self.assertEqual(packet.findings[0].evidence, "evidence for f1") self.assertEqual(packet.findings[0].reproduction, "re-run tests") self.assertEqual(packet.findings[0].acceptance, "tests passes") self.assertIn("f1", packet.criteria) def test_invalid_finding_is_caller_bug(self): with self.assertRaises(ValueError): Finding("f1", "high", "", "repro", "accept") with self.assertRaises(ValueError): evaluate(None, "t1", {}, []) def test_escalate_at_third_round(self): state = LoopState("t1", round=3, passed_gates=[], criteria={"f1": "tests"}) out = evaluate(state, "t1", {"tests": False}, [det("f1", "tests")]) self.assertEqual(out["decision"], "escalate") self.assertEqual(out["reason"], "max_rounds_exceeded") self.assertEqual(out["round"], 3) def test_weakening_is_escalate(self): # 上一轮 finding f1（无关联门）本轮既不重报、又无 fixed 证据。 state = LoopState("t1", round=1, passed_gates=[], criteria={"f1": None}) out = evaluate(state, "t1", {"other": True}, []) self.assertEqual(out["decision"], "escalate") self.assertEqual(out["reason"], "standards_weakened") self.assertIn("f1", out["weakened"]) def test_fixed_via_gate_pass_is_not_weakening(self): state = LoopState("t1", round=1, passed_gates=[], criteria={"f1": "tests"}) out = evaluate(state, "t1", {"tests": True}, []) self.assertEqual(out["decision"], "pass") def test_regression_blocks_pass(self): state = LoopState("t1", round=1, passed_gates=["tests"], criteria={}) out = evaluate(state, "t1", {"tests": False}, []) self.assertEqual(out["decision"], "pushback") self.assertTrue(any(f.id == "regression:tests" for f in out["packet"].findings)) def test_missing_gate_escalates(self): state = LoopState("t1", round=1, passed_gates=["tests"], criteria={}) out = evaluate(state, "t1", {"other": True}, []) self.assertEqual(out["decision"], "escalate") self.assertEqual(out["reason"], "gate_missing") self.assertIn("tests", out["missing_gates"]) def test_subjective_never_loops(self): sub = Finding("design", "medium", "looks off", "read code", "reviewer approves", kind="subjective") out = evaluate(None, "t1", {"tests": True}, [sub]) self.assertEqual(out["decision"], "needs_review") self.assertIsNone(out["new_state"]) self.assertEqual(out["round"], 0) def test_corrupted_state_rejected(self): with self.assertRaises(ValueError): LoopState.from_dict({"task_id": "t1", "round": -1}) with self.assertRaises(ValueError): LoopState.from_dict({"task_id": "t1", "round": "x"}) class WorkerEntryTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db = os.path.join(self.tmp.name, "test.db") conn = sqlite3.connect(self.db) conn.execute( "CREATE TABLE IF NOT EXISTS task_attempts(" "task_id TEXT, attempt_no INTEGER, kind TEXT," " interrupted_reason TEXT, checkpoint_round INTEGER," " result_text TEXT, error_text TEXT," " created_at DOUBLE PRECISION," " PRIMARY KEY(task_id, attempt_no))") conn.commit() conn.close() self.worker = Worker.__new__(Worker) self.worker.db_path = self.db self.worker.audits = [] self.worker._audit = lambda p, a, d="": self.worker.audits.append( (p, a, d)) def tearDown(self): self.tmp.cleanup() def task(self, loop=True): task = {"id": "task-1", "project_id": "proj-1"} if loop: task["audit_loop"] = True return task def test_off_by_default_touches_nothing(self): out = self.worker.audit_loop_evaluate( self.task(loop=False), {"tests": False}, [det("f1", "tests")]) self.assertEqual(out["decision"], "off") conn = sqlite3.connect(self.db) rows = conn.execute( "SELECT name FROM sqlite_master WHERE type='table' AND" " name='task_audit_loop_state'").fetchall() conn.close() self.assertEqual(rows, []) def test_fail_fix_fail_fix_pass_path(self): # 轮 1：fail -> 推回；轮 2：fail -> 推回；轮 3：pass。 out = self.worker.audit_loop_evaluate( self.task(), {"tests": False}, [det("f1", "tests")]) self.assertEqual(out["decision"], "pushback") self.assertEqual(out["round"], 1) self.assertEqual(out["packet"]["findings"][0]["id"], "f1") out = self.worker.audit_loop_evaluate( self.task(), {"tests": False}, [det("f1", "tests")]) self.assertEqual(out["decision"], "pushback") self.assertEqual(out["round"], 2) out = self.worker.audit_loop_evaluate( self.task(), {"tests": True}, []) self.assertEqual(out["decision"], "pass") def test_three_rounds_then_escalate(self): for expected in (1, 2, 3): out = self.worker.audit_loop_evaluate( self.task(), {"tests": False}, [det("f1", "tests")]) self.assertEqual(out["decision"], "pushback") self.assertEqual(out["round"], expected) out = self.worker.audit_loop_evaluate( self.task(), {"tests": False}, [det("f1", "tests")]) self.assertEqual(out["decision"], "escalate") self.assertEqual(out["reason"], "max_rounds_exceeded") self.assertTrue(any(a == "worker_audit_loop_escalated" for _, a, _ in self.worker.audits)) def test_corrupted_count_never_loops(self): self.worker.audit_loop_evaluate( self.task(), {"tests": False}, [det("f1", "tests")]) conn = sqlite3.connect(self.db) conn.execute( "UPDATE task_audit_loop_state SET round='not-a-number'" " WHERE task_id='task-1'") conn.commit() conn.close() out = self.worker.audit_loop_evaluate( self.task(), {"tests": False}, [det("f1", "tests")]) self.assertEqual(out["decision"], "escalate") self.assertEqual(out["reason"], "count_corrupted") self.assertTrue(any("count_corrupted" in d for _, _, d in self.worker.audits)) def test_state_survives_worker_restart(self): self.worker.audit_loop_evaluate( self.task(), {"tests": False}, [det("f1", "tests")]) fresh = Worker.__new__(Worker) fresh.db_path = self.db fresh.audits = [] fresh._audit = lambda p, a, d="": fresh.audits.append((p, a, d)) out = fresh.audit_loop_evaluate( self.task(), {"tests": False}, [det("f1", "tests")]) self.assertEqual(out["decision"], "pushback") self.assertEqual(out["round"], 2) def test_subjective_via_worker_needs_review(self): sub = Finding("design", "medium", "looks off", "read code", "reviewer approves", kind="subjective") out = self.worker.audit_loop_evaluate( self.task(), {"tests": True}, [sub]) self.assertEqual(out["decision"], "needs_review") if __name__ == "__main__": unittest.main()