host
19b9767478
process_audit_verdict now calls validate_signer_separation before any state save or ledger write (previously the function had no production caller on this path: it only ran inside collect_failed_checks via decide_pushback, which itself has no production caller). Signer is bound to the authenticated caller -- a body signer differing from the caller is rejected -- and executor/coordinator_approver/qa_tester must be declared and distinct from the caller; missing field, self-audit, or forged signer raises ValueError, mapped by _loop_run to HTTP 400 with no side effects. No new route. Honest limits are in the helper docstring: declared-identity enforcement only (task rows store no authoritative executor / coordinator / QA identity; task executor is a backend, not a person), and audit-fix reaudit verdicts are out of scope (the fix poster is not necessarily the re-auditor, so binding signer to the fix poster would be a mis-binding). (cherry picked from commit 56dce5165f4d08f38fe1730b9eea3b78f90cb97d) Reconcile-port: handler_lifecycle.py docstring hunk redirected into cloud/server.py audit-verdict handler (target keeps 6-mixin split; enforcement itself is in audit_loop_wiring.py and applied cleanly).
520 lines
22 KiB
Python
520 lines
22 KiB
Python
"""Server wiring for the #260 loop driver (track 2, module level).
|
|
|
|
Uses a real #241/#247 chain (LedgerStore on a temp file) and a
|
|
real sqlite loop-state connection -- no production tasks, no
|
|
GitHub, no production DB. The one fake is FailingLedgerStore,
|
|
used only to prove ledger-failure fail-closed behaviour.
|
|
"""
|
|
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from cloud import audit_loop_wiring as wiring
|
|
from cloud.audit_wiring import pushback_block_for_task
|
|
from cloud.ledger import LedgerStore
|
|
from cloud.ledger import verify_chain
|
|
except ImportError:
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
from cloud import audit_loop_wiring as wiring
|
|
from cloud.audit_wiring import pushback_block_for_task
|
|
from cloud.ledger import LedgerStore
|
|
from cloud.ledger import verify_chain
|
|
|
|
MODULE = "mod-a"
|
|
VERSION = "v1"
|
|
OLD_REF = {"archive_ref": "task:parent:pre-rebuild"}
|
|
EXTRA_PASS = {
|
|
"security_unconditional": True,
|
|
"regression_exact_artifact": True,
|
|
"demo_runnable": True,
|
|
"known_deviation_due_todo": True,
|
|
"audit_signer_separation": True,
|
|
"qa_failed_links_rerun": True,
|
|
}
|
|
HARD_PASS = {
|
|
"files_changed": ["apps/web/a.py"],
|
|
"owned_dir": "apps/web",
|
|
"protection": "protected",
|
|
"ci_status": "success",
|
|
"ci_checks": [{"name": "ci", "status": "success"}],
|
|
"required_checks": ["ci"],
|
|
"contract_version": "v1",
|
|
"contract_frozen": True,
|
|
"coordinator_approved": True,
|
|
"unresolved_review_comments": 0,
|
|
}
|
|
|
|
|
|
def fail_verdict(text="check x failed", source="audit", **extra):
|
|
verdict = {
|
|
"verdict": "fail",
|
|
"source": source,
|
|
"verdict_text": text,
|
|
"failures": [{"check": "ci_green", "detail": text}],
|
|
"passed_items": ["a", "b", "c"],
|
|
"contract_version": VERSION,
|
|
# A3: signer is the caller (auditor-1 in _verdict); the
|
|
# other three roles must be declared and distinct.
|
|
"executor": "executor-1",
|
|
"coordinator_approver": "coordinator-1",
|
|
"qa_tester": "qa-1",
|
|
}
|
|
verdict.update(extra)
|
|
return verdict
|
|
|
|
|
|
def pass_verdict(**extra):
|
|
verdict = {
|
|
"verdict": "pass",
|
|
"source": "audit",
|
|
"verdict_text": "all green",
|
|
"failures": [],
|
|
"passed_items": ["a", "b", "c"],
|
|
"contract_version": VERSION,
|
|
# A3: see fail_verdict.
|
|
"executor": "executor-1",
|
|
"coordinator_approver": "coordinator-1",
|
|
"qa_tester": "qa-1",
|
|
}
|
|
verdict.update(extra)
|
|
return verdict
|
|
|
|
|
|
class FailingLedgerStore:
|
|
"""Every append raises: proves the wiring never saves state
|
|
after a ledger failure."""
|
|
|
|
def __getattr__(self, name):
|
|
if name.startswith("append_"):
|
|
def _raise(*args, **kwargs):
|
|
raise RuntimeError("ledger down")
|
|
return _raise
|
|
raise AttributeError(name)
|
|
|
|
|
|
class WiringTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.ledger_path = os.path.join(self.tmp.name, "ledger.db")
|
|
self.store = LedgerStore(self.ledger_path)
|
|
self.conn = sqlite3.connect(":memory:")
|
|
self.task = {"id": "task-parent", "project_id": "proj-1",
|
|
"module_id": MODULE, "contract_version": VERSION}
|
|
|
|
def tearDown(self):
|
|
try:
|
|
self.store.close()
|
|
except Exception:
|
|
pass
|
|
self.conn.close()
|
|
self.tmp.cleanup()
|
|
|
|
def _block(self, task_id="task-parent"):
|
|
return pushback_block_for_task(self.store.conn, task_id)
|
|
|
|
def _verdict(self, verdict, **kwargs):
|
|
kwargs.setdefault("user_id", "auditor-1")
|
|
return wiring.process_audit_verdict(
|
|
self.conn, self.store, self.task, verdict, **kwargs)
|
|
|
|
def test_module_mapping_explicit_wins_and_fallback_is_task_id(self):
|
|
self.assertEqual(
|
|
wiring.resolve_module_id(
|
|
{"id": "t1", "module_id": "m-explicit"}), "m-explicit")
|
|
self.assertEqual(
|
|
wiring.resolve_module_id({"id": "t1"}), "t1")
|
|
self.assertEqual(
|
|
wiring.resolve_module_id({"id": "t1", "module": "m2"}), "m2")
|
|
with self.assertRaises(ValueError):
|
|
wiring.resolve_contract_version({"id": "t1"}, None)
|
|
self.assertEqual(
|
|
wiring.resolve_contract_version(
|
|
{"id": "t1"}, {"contract_version": "v9"}), "v9")
|
|
|
|
def test_state_persists_and_reloads_from_table_not_memory(self):
|
|
result = self._verdict(fail_verdict())
|
|
self.assertEqual(result["action_kind"], "pushback")
|
|
self.assertEqual(result["subtask"]["kind"], "fix")
|
|
self.assertTrue(result["merge_blocked"])
|
|
reloaded = wiring.load_loop_state(self.conn, MODULE, VERSION)
|
|
self.assertEqual(reloaded.pushback_round, 1)
|
|
self.assertEqual(reloaded.failure_count, 1)
|
|
self.assertEqual(reloaded.status, "pushback")
|
|
self.assertTrue(self._block()["merge_blocked"])
|
|
# The chain itself verifies: the wiring wrote real rows.
|
|
self.assertTrue(verify_chain(self.store.conn)["ok"])
|
|
|
|
def test_full_chain_fail_fix_reaudit_pass_unblocks(self):
|
|
self._verdict(fail_verdict())
|
|
fix_task = {"id": "task-fix-1", "project_id": "proj-1",
|
|
"module_id": MODULE, "contract_version": VERSION}
|
|
result = wiring.process_fix_completion(
|
|
self.conn, self.store, fix_task,
|
|
{"reaudit_verdict": pass_verdict()},
|
|
automated={"hard_gate": HARD_PASS, "extra_checks": EXTRA_PASS},
|
|
user_id="auditor-1", ledger_task_id="task-parent")
|
|
self.assertEqual(result["action_kind"], "pass")
|
|
self.assertIsNone(result["subtask"])
|
|
self.assertFalse(result["merge_blocked"])
|
|
self.assertTrue(
|
|
result["automated_checks"]["automated_checks_pass"])
|
|
# Parent task dict untouched by the wiring layer: the
|
|
# original task is never rolled back by this module.
|
|
self.assertNotIn("status", self.task)
|
|
self.assertFalse(self._block()["merge_blocked"])
|
|
state = wiring.load_loop_state(self.conn, MODULE, VERSION)
|
|
self.assertEqual(state.status, "passed")
|
|
self.assertEqual(state.failure_count, 0)
|
|
|
|
def test_automated_failure_forces_claimed_pass_to_fail(self):
|
|
self._verdict(fail_verdict())
|
|
fix_task = {"id": "task-fix-1", "project_id": "proj-1",
|
|
"module_id": MODULE, "contract_version": VERSION}
|
|
broken_extra = dict(EXTRA_PASS, demo_runnable=False)
|
|
result = wiring.process_fix_completion(
|
|
self.conn, self.store, fix_task,
|
|
{"reaudit_verdict": pass_verdict()},
|
|
automated={"hard_gate": HARD_PASS,
|
|
"extra_checks": broken_extra},
|
|
user_id="auditor-1", ledger_task_id="task-parent")
|
|
# Automated checks passing is never audit passing, and
|
|
# here they did not pass: the claimed pass is forced fail.
|
|
self.assertEqual(result["action_kind"], "pushback")
|
|
self.assertTrue(result["merge_blocked"])
|
|
self.assertIn("demo_runnable",
|
|
result["automated_checks"]["failed_checks"])
|
|
self.assertTrue(self._block()["merge_blocked"])
|
|
|
|
def test_fix_without_reaudit_verdict_and_diff_only_refused(self):
|
|
self._verdict(fail_verdict())
|
|
fix_task = {"id": "task-fix-1", "project_id": "proj-1",
|
|
"module_id": MODULE, "contract_version": VERSION}
|
|
with self.assertRaises(ValueError):
|
|
wiring.process_fix_completion(
|
|
self.conn, self.store, fix_task, {"fix_diff": {}},
|
|
user_id="auditor-1", ledger_task_id="task-parent")
|
|
with self.assertRaises(ValueError):
|
|
wiring.process_fix_completion(
|
|
self.conn, self.store, fix_task,
|
|
{"diff_only": True,
|
|
"reaudit_verdict": pass_verdict()},
|
|
user_id="auditor-1", ledger_task_id="task-parent")
|
|
# Fix with no prior loop at all is refused, not invented.
|
|
fresh_conn = sqlite3.connect(":memory:")
|
|
try:
|
|
with self.assertRaises(ValueError):
|
|
wiring.process_fix_completion(
|
|
fresh_conn, self.store, fix_task,
|
|
{"reaudit_verdict": pass_verdict()},
|
|
user_id="auditor-1", ledger_task_id="task-parent")
|
|
finally:
|
|
fresh_conn.close()
|
|
|
|
def test_exactly_three_pushes_back_fourth_rebuilds(self):
|
|
for _ in range(3):
|
|
result = self._verdict(fail_verdict())
|
|
self.assertEqual(result["action_kind"], "pushback")
|
|
# Fourth failure without a caller-supplied old impl ref
|
|
# is refused BEFORE any ledger row: state stays at 3.
|
|
with self.assertRaises(ValueError):
|
|
self._verdict(fail_verdict())
|
|
state = wiring.load_loop_state(self.conn, MODULE, VERSION)
|
|
self.assertEqual(state.pushback_round, 3)
|
|
self.assertEqual(state.failure_count, 3)
|
|
result = self._verdict(fail_verdict(), old_impl_ref=OLD_REF)
|
|
self.assertEqual(result["action_kind"], "rebuild")
|
|
self.assertEqual(result["subtask"]["kind"], "rebuild")
|
|
state = wiring.load_loop_state(self.conn, MODULE, VERSION)
|
|
self.assertEqual(state.rebuild_no, 1)
|
|
self.assertEqual(state.status, "rebuilding")
|
|
|
|
def test_targeted_third_round_scope(self):
|
|
self._verdict(fail_verdict())
|
|
self._verdict(fail_verdict())
|
|
result = self._verdict(fail_verdict(
|
|
failures=[{"check": "ci_green", "detail": "one left"}],
|
|
passed_items=["a", "b", "c"],
|
|
broken_passed_items=[]))
|
|
self.assertEqual(result["action_kind"], "pushback")
|
|
self.assertEqual(result["subtask"]["scope"], "targeted")
|
|
self.assertEqual(result["subtask"]["targeted_items"],
|
|
["ci_green"])
|
|
|
|
def test_correction_to_pass_does_not_unblock(self):
|
|
self._verdict(fail_verdict())
|
|
verdict_rows = self.store.conn.execute(
|
|
"SELECT id FROM audit_ledger WHERE action = ?"
|
|
" ORDER BY id ASC", ("audit_verdict",)).fetchall()
|
|
entry_id = verdict_rows[0][0]
|
|
result = wiring.process_correction(
|
|
self.conn, self.store, self.task,
|
|
{"original_verdict": "fail", "corrected_verdict": "pass",
|
|
"reviewer": "reviewer-1", "original_entry_id": entry_id,
|
|
"findings": ["misjudged"], "contract_version": VERSION},
|
|
user_id="reviewer-1", ledger_task_id="task-parent")
|
|
self.assertEqual(result["correction_action"], "schedule_reaudit")
|
|
self.assertTrue(result["merge_blocked"])
|
|
self.assertEqual(result["next_state"].failure_count, 0)
|
|
self.assertEqual(result["next_state"].status, "blocked")
|
|
self.assertEqual(result["subtask"]["kind"], "reaudit")
|
|
# Counting view (0) vs gate view (still blocked) is the
|
|
# #247 intended divergence, pinned here, not papered over.
|
|
self.assertTrue(self._block()["merge_blocked"])
|
|
# Only a fresh passing verdict unblocks.
|
|
self._verdict(pass_verdict())
|
|
self.assertFalse(self._block()["merge_blocked"])
|
|
|
|
def test_correction_to_fail_reblocks_at_once(self):
|
|
self._verdict(pass_verdict())
|
|
self.assertFalse(self._block()["merge_blocked"])
|
|
verdict_rows = self.store.conn.execute(
|
|
"SELECT id FROM audit_ledger WHERE action = ?"
|
|
" ORDER BY id ASC", ("audit_verdict",)).fetchall()
|
|
entry_id = verdict_rows[-1][0]
|
|
result = wiring.process_correction(
|
|
self.conn, self.store, self.task,
|
|
{"original_verdict": "pass", "corrected_verdict": "fail",
|
|
"reviewer": "reviewer-1", "original_entry_id": entry_id,
|
|
"findings": ["missed failure"],
|
|
"contract_version": VERSION},
|
|
user_id="reviewer-1", ledger_task_id="task-parent")
|
|
self.assertEqual(
|
|
result["correction_action"], "retrigger_pushback_and_reblock")
|
|
self.assertTrue(result["merge_blocked"])
|
|
self.assertEqual(result["next_state"].failure_count, 1)
|
|
self.assertTrue(self._block()["merge_blocked"])
|
|
|
|
def test_acceptance_rebuild_pending_confirm_reject_no_budget(self):
|
|
for _ in range(4):
|
|
result = self._verdict(
|
|
fail_verdict(source="acceptance"))
|
|
self.assertEqual(result["action_kind"],
|
|
"pending_user_confirmation")
|
|
self.assertIsNone(result["subtask"])
|
|
self.assertIsNotNone(result["notification"])
|
|
state = wiring.load_loop_state(self.conn, MODULE, VERSION)
|
|
self.assertTrue(state.pending_confirmation)
|
|
self.assertEqual(state.rebuild_no, 0)
|
|
# System/coordinator markers can never confirm.
|
|
with self.assertRaises(ValueError):
|
|
wiring.process_rebuild_confirmation(
|
|
self.conn, self.store, self.task,
|
|
user_confirmed=True, confirmed_by="system",
|
|
user_id="user-1", ledger_task_id="task-parent")
|
|
with self.assertRaises(ValueError):
|
|
wiring.process_rebuild_confirmation(
|
|
self.conn, self.store, self.task,
|
|
user_confirmed=True, confirmed_by="coordinator",
|
|
user_id="user-1", ledger_task_id="task-parent")
|
|
# Rejection escalates and consumes no rebuild budget.
|
|
rejected = wiring.process_rebuild_confirmation(
|
|
self.conn, self.store, self.task,
|
|
user_confirmed=False, confirmed_by="user-1",
|
|
user_id="user-1", ledger_task_id="task-parent")
|
|
self.assertEqual(rejected["action_kind"], "escalate")
|
|
state = wiring.load_loop_state(self.conn, MODULE, VERSION)
|
|
self.assertEqual(state.rebuild_no, 0)
|
|
|
|
def test_acceptance_rebuild_confirm_then_rebuilds(self):
|
|
for _ in range(4):
|
|
self._verdict(fail_verdict(source="acceptance"))
|
|
confirmed = wiring.process_rebuild_confirmation(
|
|
self.conn, self.store, self.task,
|
|
user_confirmed=True, confirmed_by="user-1",
|
|
old_impl_ref=OLD_REF,
|
|
user_id="user-1", ledger_task_id="task-parent")
|
|
self.assertEqual(confirmed["action_kind"], "rebuild")
|
|
self.assertEqual(confirmed["subtask"]["kind"], "rebuild")
|
|
state = wiring.load_loop_state(self.conn, MODULE, VERSION)
|
|
self.assertEqual(state.rebuild_no, 1)
|
|
self.assertEqual(state.status, "rebuilding")
|
|
# Confirming again with no pending confirmation fails.
|
|
with self.assertRaises(ValueError):
|
|
wiring.process_rebuild_confirmation(
|
|
self.conn, self.store, self.task,
|
|
user_confirmed=True, confirmed_by="user-1",
|
|
user_id="user-1", ledger_task_id="task-parent")
|
|
|
|
def test_quota_blocked_consumes_nothing(self):
|
|
self._verdict(fail_verdict())
|
|
blocked = self._verdict(
|
|
fail_verdict(), quota_check=lambda estimate: False)
|
|
self.assertEqual(blocked["action_kind"], "quota_blocked")
|
|
self.assertFalse(blocked["state_saved"])
|
|
state = wiring.load_loop_state(self.conn, MODULE, VERSION)
|
|
self.assertEqual(state.pushback_round, 1)
|
|
self.assertEqual(state.failure_count, 1)
|
|
|
|
def _raising(estimate):
|
|
raise RuntimeError("quota service down")
|
|
blocked2 = self._verdict(
|
|
fail_verdict(), quota_check=_raising)
|
|
self.assertEqual(blocked2["action_kind"], "quota_blocked")
|
|
state = wiring.load_loop_state(self.conn, MODULE, VERSION)
|
|
self.assertEqual(state.pushback_round, 1)
|
|
|
|
def test_quota_adapter_matches_server_rule(self):
|
|
check = wiring.build_quota_check(1000000, 999999)
|
|
self.assertTrue(check(None))
|
|
self.assertFalse(check(0.01))
|
|
self.assertTrue(wiring.build_quota_check(None, 10 ** 9)(100.0))
|
|
|
|
def test_cost_threshold_escalates_rebuild_with_full_history(self):
|
|
for _ in range(3):
|
|
self._verdict(fail_verdict())
|
|
result = self._verdict(
|
|
fail_verdict(cost_estimate=100.0), cost_threshold=10.0,
|
|
old_impl_ref=OLD_REF)
|
|
self.assertEqual(result["action_kind"], "escalate")
|
|
self.assertEqual(result["notification"]["party"], "user")
|
|
self.assertEqual(
|
|
len(result["notification"]["escalation_history"]), 4)
|
|
self.assertFalse(
|
|
result["escalation_grants_user_acceptance"])
|
|
|
|
def test_ledger_failure_fail_closed_state_not_saved(self):
|
|
failing = FailingLedgerStore()
|
|
with self.assertRaises(RuntimeError):
|
|
wiring.process_audit_verdict(
|
|
self.conn, failing, self.task, fail_verdict(),
|
|
user_id="auditor-1")
|
|
self.assertIsNone(
|
|
wiring.load_loop_state(self.conn, MODULE, VERSION))
|
|
|
|
def test_automated_reaudit_evidence_missing_fails(self):
|
|
result = wiring.evaluate_automated_reaudit(None, {})
|
|
self.assertFalse(result["automated_checks_pass"])
|
|
self.assertIn("hard_gate_evidence_missing",
|
|
result["failed_checks"])
|
|
result = wiring.evaluate_automated_reaudit(
|
|
HARD_PASS, EXTRA_PASS)
|
|
self.assertTrue(result["automated_checks_pass"])
|
|
# Merge-body vocabulary is translated, not rejected:
|
|
# branch_protection -> the hard gate's protection.
|
|
body_style = dict(HARD_PASS)
|
|
body_style["branch_protection"] = body_style.pop("protection")
|
|
result = wiring.evaluate_automated_reaudit(
|
|
body_style, EXTRA_PASS)
|
|
self.assertTrue(result["automated_checks_pass"])
|
|
|
|
def test_release_gate_multi_source_and_read_only(self):
|
|
self._verdict(fail_verdict())
|
|
before = self.store.conn.execute(
|
|
"SELECT COUNT(*) FROM audit_ledger").fetchone()[0]
|
|
decision = wiring.evaluate_release_gate(
|
|
"task-parent", self.store.conn,
|
|
task_status="completed", hard_gate={"allowed": False,
|
|
"blocks": ["ci_not_green"]})
|
|
self.assertFalse(decision["allowed"])
|
|
self.assertEqual(decision["http_status"], 409)
|
|
self.assertEqual(decision["blocked_sources"],
|
|
["task_status", "hard_gate", "audit_pushback"])
|
|
self.assertEqual(decision["code"], "release_requires_merged")
|
|
after = self.store.conn.execute(
|
|
"SELECT COUNT(*) FROM audit_ledger").fetchone()[0]
|
|
self.assertEqual(before, after)
|
|
self.assertTrue(verify_chain(self.store.conn)["ok"])
|
|
|
|
def test_release_gate_evidence_missing_and_allowed(self):
|
|
decision = wiring.evaluate_release_gate(
|
|
"task-parent", self.store.conn,
|
|
task_status="merged", hard_gate=None)
|
|
self.assertEqual(decision["code"],
|
|
"release_hard_gate_evidence_missing")
|
|
decision = wiring.evaluate_release_gate(
|
|
"task-parent", self.store.conn,
|
|
task_status="merged",
|
|
hard_gate={"allowed": True, "blocks": []})
|
|
self.assertTrue(decision["allowed"])
|
|
self.assertEqual(decision["code"], "release_allowed")
|
|
|
|
def test_release_gate_chain_read_failure_fail_closed(self):
|
|
closed = sqlite3.connect(":memory:")
|
|
closed.close()
|
|
decision = wiring.evaluate_release_gate(
|
|
"task-parent", closed, task_status="merged",
|
|
hard_gate={"allowed": True, "blocks": []})
|
|
self.assertFalse(decision["allowed"])
|
|
self.assertEqual(
|
|
decision["code"], "release_blocked_state_unavailable")
|
|
|
|
|
|
class SignerSeparationTestCase(unittest.TestCase):
|
|
"""A3: validate_signer_separation 在提交路径上的真实调用方。
|
|
|
|
覆盖：自审（调用方 = 声明的 executor/coordinator/QA 任一）
|
|
被拒且不落链不落状态；缺字段被拒；体里 signer 与调用方
|
|
不符被拒；声明三方均与调用方不同时正常提交。此前该校验
|
|
只活在重审清单函数里（collect_failed_checks，经无生产
|
|
调用方的 decide_pushback），提交路径上从未执行。
|
|
|
|
注意：本类故意不继承 WiringTestCase——继承会使父类全部
|
|
测试方法在本类下再跑一遍（全量计数虚增），故只复用其
|
|
setUp 形状，不复用其用例。
|
|
"""
|
|
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.ledger_path = os.path.join(self.tmp.name, "ledger.db")
|
|
self.store = LedgerStore(self.ledger_path)
|
|
self.conn = sqlite3.connect(":memory:")
|
|
self.task = {"id": "task-parent", "project_id": "proj-1",
|
|
"module_id": MODULE, "contract_version": VERSION}
|
|
|
|
def tearDown(self):
|
|
try:
|
|
self.store.close()
|
|
except Exception:
|
|
pass
|
|
self.conn.close()
|
|
self.tmp.cleanup()
|
|
|
|
def _verdict(self, verdict, **kwargs):
|
|
kwargs.setdefault("user_id", "auditor-1")
|
|
return wiring.process_audit_verdict(
|
|
self.conn, self.store, self.task, verdict, **kwargs)
|
|
|
|
def _chain_rows(self):
|
|
return self.store.conn.execute(
|
|
"SELECT COUNT(*) FROM audit_ledger").fetchone()[0]
|
|
|
|
def test_self_audit_rejected_without_side_effects(self):
|
|
for field in ("executor", "coordinator_approver", "qa_tester"):
|
|
with self.subTest(field=field):
|
|
verdict = fail_verdict(**{field: "auditor-1"})
|
|
with self.assertRaises(ValueError) as caught:
|
|
self._verdict(verdict)
|
|
self.assertIn("signer", str(caught.exception))
|
|
self.assertIsNone(
|
|
wiring.load_loop_state(self.conn, MODULE, VERSION))
|
|
self.assertEqual(self._chain_rows(), 0)
|
|
|
|
def test_missing_identity_field_rejected(self):
|
|
verdict = fail_verdict()
|
|
del verdict["qa_tester"]
|
|
with self.assertRaises(ValueError):
|
|
self._verdict(verdict)
|
|
self.assertIsNone(
|
|
wiring.load_loop_state(self.conn, MODULE, VERSION))
|
|
self.assertEqual(self._chain_rows(), 0)
|
|
|
|
def test_forged_signer_rejected(self):
|
|
verdict = fail_verdict(signer="someone-else")
|
|
with self.assertRaises(ValueError):
|
|
self._verdict(verdict)
|
|
self.assertEqual(self._chain_rows(), 0)
|
|
|
|
def test_distinct_declared_roles_submit(self):
|
|
result = self._verdict(pass_verdict())
|
|
self.assertEqual(result["action_kind"], "pass")
|
|
self.assertFalse(result["merge_blocked"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|