nexo312/trading-system-docs
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
trading-system-docs/a4/test_a4.py

670 lines
28 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
Red Queen — A4 Testsuite (deterministisch, isoliert).
Nutzt ausschliesslich temporaere DBs (tempfile.mkdtemp). Kein Netz, kein hermes,
keine echte delegate_task. Der Child-Versand wird ueber eine simulierte
`ChildDispatcher`-Abstraktion (Callable) injiziert — die Library selbst startet
nichts und laeuft nie zyklisch.
Lauf:
python3 test_a4.py
Exit-Code 0 = alle Tests gruen; 1 = mindestens ein Fehler.
Deckt A4 §6-§33 ab: plan validation, dependency validation, runnable selection,
A2/A3-Consistency-Gate, maker/checker dispatch contract, checker pass/fail,
retry allowed/denied, retry4 impossible, circuit open blocks maker, approval gate,
risk overrides size, no runnable stop, mission completion gate, final review gate,
idempotent WP execution, restart/resume state, child failure, reason codes,
secret-safe logging.
"""
from __future__ import annotations
import json
import os
import sys
import tempfile
from pathlib import Path
_HERE = Path(__file__).resolve().parent
_REPO = _HERE.parent
for _p in (str(_REPO), str(_HERE)):
if _p not in sys.path:
sys.path.insert(0, _p)
import rq_orchestrator as o # noqa: E402
from rq_orchestrator import ( # noqa: E402
CRITICAL_TARGETS,
ROLE_CHECKER,
ROLE_MAKER,
ST_APPROVAL_REQUIRED,
ST_CHILD_TECHNICAL_FAILURE,
ST_CHECKER_BLOCKED,
ST_DEBUG_REQUIRED,
ST_MISSION_COMPLETED,
ST_MISSION_NOT_COMPLETED,
ST_NEEDS_CHECKER_DISPATCH,
ST_NEEDS_DISPATCH,
ST_NO_RUNNABLE_WORK,
ST_RETRY4_IMPOSSIBLE,
ST_SECOND_OPINION_REQUIRED,
ST_TERMINAL,
ST_WAITING_PENDING,
Orchestrator,
OrchestratorError,
PlanValidator,
build_checker_contract,
build_maker_contract,
classify_risk,
make_child_evidence,
)
PASS = 0
FAIL = 0
FAILURES = []
def check(name: str, cond: bool, extra: str = ""):
global PASS, FAIL
if cond:
PASS += 1
print(f" [PASS] {name}")
else:
FAIL += 1
FAILURES.append(name)
print(f" [FAIL] {name} {extra}")
def expect_err(name, fn, code, fragment=""):
try:
fn()
except OrchestratorError as e:
ok = e.code == code and (not fragment or fragment in e.message)
check(name, ok, f"got code={e.code} msg={e.message!r}")
return e
except Exception as e: # noqa
check(name, False, f"unexpected {type(e).__name__}: {e}")
return None
check(name, False, "no error raised")
return None
def fresh():
d = tempfile.mkdtemp(prefix="a4test_")
mission_db = str(Path(d) / "missions.db")
safety_db = str(Path(d) / "safety.db")
return d, mission_db, safety_db
def default_plan(wps=None, mission_id="RQ-A4-1"):
if wps is None:
wps = [
{
"wp_id": "W1",
"title": "W1",
"goal": "Implementiere Feature X",
"description": "Feature X implementieren",
"scope": "a4/w1.py",
"target_component": "feature",
"acceptance_criteria": "Tests gruen",
"test_requirements": "pytest",
},
]
return {
"mission_id": mission_id,
"title": "A4-Test-Mission",
"goal": "Test-Mission fuer A4",
"scope": "tests",
"work_packages": wps,
}
def make_orch(dispatcher=None, mission_db=None, safety_db=None, **kw):
if mission_db is None or safety_db is None:
_, mission_db, safety_db = fresh()
return Orchestrator(mission_db, safety_db, dispatcher, **kw)
# --------------------------------------------------------------------------- #
# 1) plan_validation
# --------------------------------------------------------------------------- #
def test_plan_validation():
print("\n== plan validation ==")
orch = make_orch()
valid_plan = default_plan()
res = orch.load_plan(valid_plan)
check("valid plan accepted", res["status"] == o.ST_PLAN_VALIDATED, res)
v = PlanValidator()
# cycle
cyc = default_plan(wps=[
{"wp_id": "A", "goal": "a", "scope": "s", "acceptance_criteria": "ac", "dependencies": ["B"]},
{"wp_id": "B", "goal": "b", "scope": "s", "acceptance_criteria": "ac", "dependencies": ["A"]},
])
r = v.validate(cyc)
check("cycle rejected", not r["valid"] and any(e["code"] == "DEPENDENCY_CYCLE" for e in r["errors"]), r)
# self-dep
sd = default_plan([{"wp_id": "A", "goal": "a", "scope": "s", "acceptance_criteria": "ac",
"dependencies": ["A"]}])
r = v.validate(sd)
check("self-dep rejected", not r["valid"] and any(e["code"] == "SELF_DEPENDENCY" for e in r["errors"]), r)
# duplicate id
dup = default_plan([
{"wp_id": "A", "goal": "a", "scope": "s", "acceptance_criteria": "ac"},
{"wp_id": "A", "goal": "a2", "scope": "s", "acceptance_criteria": "ac"},
])
r = v.validate(dup)
check("duplicate id rejected", not r["valid"] and any(e["code"] == "DUPLICATE_WP_ID" for e in r["errors"]), r)
# unknown target (unknown dependency)
unk = default_plan([{"wp_id": "A", "goal": "a", "scope": "s", "acceptance_criteria": "ac",
"dependencies": ["NOPE"]}])
r = v.validate(unk)
check("unknown dependency rejected", not r["valid"] and any(e["code"] == "UNKNOWN_DEPENDENCY" for e in r["errors"]), r)
# missing AC
noac = default_plan([{"wp_id": "A", "goal": "a", "scope": "s"}])
r = v.validate(noac)
check("missing AC rejected", not r["valid"] and any(e["code"] == "MISSING_AC" for e in r["errors"]), r)
# invalid wp state
badstate = default_plan([{"wp_id": "A", "goal": "a", "scope": "s", "acceptance_criteria": "ac",
"state": "IN_PROGRESS"}])
r = v.validate(badstate)
check("invalid wp state rejected", not r["valid"] and any(e["code"] == "INVALID_WP_STATE" for e in r["errors"]), r)
# foreign wp
fw = default_plan([{"wp_id": "A", "mission_id": "OTHER", "goal": "a", "scope": "s", "acceptance_criteria": "ac"}])
v2 = PlanValidator(mission_id="RQ-A4-1")
r = v2.validate(fw)
check("foreign wp rejected", not r["valid"] and any(e["code"] == "FOREIGN_WP" for e in r["errors"]), r)
# --------------------------------------------------------------------------- #
# 2) dependency_validation
# --------------------------------------------------------------------------- #
def test_dependency_validation():
print("\n== dependency validation ==")
orch, mission_db, safety_db = make_orch_with_paths()
plan = default_plan([
{"wp_id": "A", "goal": "a", "scope": "s", "acceptance_criteria": "ac"},
{"wp_id": "B", "goal": "b", "scope": "s", "acceptance_criteria": "ac", "dependencies": ["A"]},
{"wp_id": "C", "goal": "c", "scope": "s", "acceptance_criteria": "ac", "dependencies": ["B"]},
])
res = orch.load_plan(plan)
mid = res["mission"]["id"]
# B nicht runnable bevor A DONE; C nicht vor B
orch.missions.mission_transition(mid, "RUNNING", evidence="t")
runnable = orch.select_runnable(mid)
check("only A runnable initially", [w["id"] for w in runnable] == ["A"], runnable)
# A DONE -> B runnable
orch.missions.wp_transition("A", "IN_PROGRESS")
orch.missions.wp_transition("A", "CHECKING")
orch.missions.wp_complete("A")
runnable = orch.select_runnable(mid)
check("B runnable after A done", [w["id"] for w in runnable] == ["B"], runnable)
orch.missions.wp_transition("B", "IN_PROGRESS")
orch.missions.wp_transition("B", "CHECKING")
orch.missions.wp_complete("B")
runnable = orch.select_runnable(mid)
check("C runnable after B done", [w["id"] for w in runnable] == ["C"], runnable)
# --------------------------------------------------------------------------- #
# 3) runnable_selection
# --------------------------------------------------------------------------- #
def test_runnable_selection():
print("\n== runnable selection ==")
orch, _, _ = make_orch_with_paths()
plan = default_plan([
{"wp_id": "W1", "goal": "a", "scope": "s", "acceptance_criteria": "ac"},
{"wp_id": "W2", "goal": "b", "scope": "s", "acceptance_criteria": "ac", "dependencies": ["W1"]},
])
res = orch.load_plan(plan)
mid = res["mission"]["id"]
orch.missions.mission_transition(mid, "RUNNING", evidence="t")
# W1 READY + deps done + safety closed -> runnable
runnable = orch.select_runnable(mid)
check("W1 runnable", [w["id"] for w in runnable] == ["W1"], runnable)
# W2 blocked by dep
check("W2 not runnable (dep open)", all(w["id"] != "W2" for w in runnable))
# safety circuit open blocks W1
orch.safety.open_circuit("MISSION", mid, trigger="REQ-1", severity="HIGH", mission_id=mid)
runnable = orch.select_runnable(mid)
check("no runnable with circuit open", runnable == [], runnable)
# --------------------------------------------------------------------------- #
# 4) a2a3_consistency
# --------------------------------------------------------------------------- #
def test_a2a3_consistency():
print("\n== a2/a3 consistency gate ==")
orch, _, _ = make_orch_with_paths()
orch.load_plan(default_plan())
mid = "RQ-A4-1"
orch.missions.mission_transition(mid, "RUNNING", evidence="t")
# Circuit OPEN while Mission RUNNING -> mutation blocked
orch.safety.open_circuit("MISSION", mid, trigger="CRIT-1", severity="CRITICAL", mission_id=mid)
step = orch.run_one_step(mid)
check("mutation blocked on open circuit", step["status"] == o.ST_MUTATION_BLOCKED, step)
# kein WP wurde dispatched
st = orch._wp_state("W1")
check("no mutation, W1 still READY", st == "READY", st)
# corrupted safety state -> fail-closed (ORCHESTRATOR_STATE_CONFLICT)
orch2, _, _ = make_orch_with_paths()
orch2.load_plan(default_plan())
# korrupten Circuit-Zustand injizieren (via open_circuit + manuelles UPDATE)
orch2.safety.open_circuit("MISSION", "RQ-A4-1", trigger="t", severity="HIGH", mission_id="RQ-A4-1")
import sqlite3
with sqlite3.connect(orch2.safety_db) as c:
c.execute("UPDATE circuit_state SET state='GARBAGE' WHERE scope_type='MISSION' AND scope_id='RQ-A4-1'")
step2 = orch2.run_one_step("RQ-A4-1")
check("state conflict on corrupt safety", step2["status"] in (o.ST_STATE_CONFLICT, o.ST_MUTATION_BLOCKED), step2)
# --------------------------------------------------------------------------- #
# 5) maker_dispatch_contract
# --------------------------------------------------------------------------- #
def test_maker_dispatch_contract():
print("\n== maker dispatch contract ==")
orch, _, _ = make_orch_with_paths()
res = orch.load_plan(default_plan())
mid = res["mission"]["id"]
wp = orch._wp_meta("W1")
mission = orch.missions.mission_read(mid)
c = build_maker_contract(wp, mission, "MEDIUM", retry_round=1)
for k in ("MISSION_ID", "WP_ID", "GOAL", "SCOPE", "ALLOWED_FILES", "FORBIDDEN_FILES",
"ACCEPTANCE_CRITERIA", "TEST_REQUIREMENTS", "SAFETY_RULES"):
check(f"maker contract has {k}", k in c, c)
check("maker contract role", c.get("role") == ROLE_MAKER)
check("maker contract task id", c.get("TASK_ID") == f"{mid}:W1:MAKER:R1", c)
# --------------------------------------------------------------------------- #
# 6) checker_dispatch_contract
# --------------------------------------------------------------------------- #
def test_checker_dispatch_contract():
print("\n== checker dispatch contract ==")
orch, *_ = make_orch_with_paths()
res = orch.load_plan(default_plan())
mid = res["mission"]["id"]
wp = orch._wp_meta("W1")
mission = orch.missions.mission_read(mid)
maker_ev = {"test_results": [{"verdict": "PASS"}], "IMPLEMENTATION_SUMMARY": "SECRET ARG",
"strategy": "t"}
c = build_checker_contract(wp, mission, maker_ev, "MEDIUM", retry_round=1)
for k in ("requirement", "acceptance_criteria", "diff_files", "test_results", "safety_rules"):
check(f"checker contract has {k}", k in c)
check("checker no maker argumentation", "IMPLEMENTATION_SUMMARY" not in json.dumps(c),
json.dumps(c))
check("checker role", c["role"] == ROLE_CHECKER)
# --------------------------------------------------------------------------- #
# 7) checker_pass
# --------------------------------------------------------------------------- #
def test_checker_pass():
print("\n== checker pass ==")
d, mdb, sdb = fresh()
orch = Orchestrator(mdb, sdb, _pass_dispatcher())
res = orch.load_plan(default_plan())
mid = res["mission"]["id"]
step = orch.run_one_step(mid)
check("mission completes via checker pass", step["status"] in o.ST_CHECKER_PASS, step)
check("wp done after checker pass", orch._wp_state("W1") == "DONE")
# --------------------------------------------------------------------------- #
# 8) checker_fail
# --------------------------------------------------------------------------- #
def test_checker_fail():
print("\n== checker fail ==")
d, mdb, sdb = fresh()
orch = Orchestrator(mdb, sdb, fail_dispatcher())
res = orch.load_plan(default_plan())
mid = res["mission"]["id"]
step = orch.run_one_step(mid)
# 1. FAIL -> RETRY (messbarer Fortschritt), neuer Maker-Dispatch notiert
check("first fail -> retry allowed", step["status"] in (ST_NEEDS_DISPATCH, ST_CHECKER_BLOCKED,
ST_SECOND_OPINION_REQUIRED, ST_DEBUG_REQUIRED), step)
check("attempt recorded", orch.safety.attempts_count(mid) >= 1)
check("decision retry", step.get("decision") == "RETRY", step)
# --------------------------------------------------------------------------- #
# 9) retry_allowed
# --------------------------------------------------------------------------- #
def test_retry_allowed():
print("\n== retry allowed ==")
d, mdb, sdb = fresh()
orch = Orchestrator(mdb, sdb, fail_dispatcher())
res = orch.load_plan(default_plan())
mid = res["mission"]["id"]
step = orch.run_one_step(mid)
check("1 fail -> retry allowed", step["status"] in (ST_NEEDS_DISPATCH, ST_SECOND_OPINION_REQUIRED,
ST_DEBUG_REQUIRED), step)
check("attempt in ledger", len(orch.safety.attempts(mid, "W1")) >= 1)
# --------------------------------------------------------------------------- #
# 10) retry_denied / 11) retry4_impossible
# --------------------------------------------------------------------------- #
def test_retry_denied_and_4_impossible():
print("\n== retry denied / retry4 impossible ==")
d, mdb, sdb = fresh()
orch = Orchestrator(mdb, sdb, fail_dispatcher())
res = orch.load_plan(default_plan())
mid = res["mission"]["id"]
# Pre-record 3 maker FAIL attempts (unique errors) to exhaust repair budget
for i in range(3):
orch.safety.record_attempt(mid, "W1", actor="maker", result="FAIL",
error=f"e{i}", strategy=f"s{i}", progress="NO")
step = orch.run_one_step(mid)
check("3 fails -> no 4th maker", step["status"] in (o.ST_RETRY4_IMPOSSIBLE, o.ST_SECOND_OPINION_REQUIRED), step)
if step["status"] == o.ST_RETRY4_IMPOSSIBLE:
check("repair count exceeded", step["repair_count"] >= 3, step)
check("decision not RETRY", step.get("decision") != "RETRY", step)
# ensure no maker dispatched after limit
check("no dispatch after limit", step.get("dispatch") is None, step)
# --------------------------------------------------------------------------- #
# 12) circuit_open_blocks_maker
# --------------------------------------------------------------------------- #
def test_circuit_open_blocks_maker():
print("\n== circuit open blocks maker ==")
orch, _, _ = make_orch_with_paths()
orch.load_plan(default_plan())
mid = "RQ-A4-1"
orch.missions.mission_transition(mid, "RUNNING", evidence="t")
orch.safety.open_circuit("MISSION", mid, trigger="PERSISTENCE_CORRUPTED", severity="CRITICAL", mission_id=mid)
step = orch.run_one_step(mid)
check("maker dispatch blocked by open circuit", step["status"] == o.ST_MUTATION_BLOCKED, step)
# --------------------------------------------------------------------------- #
# 13) approval_gate
# --------------------------------------------------------------------------- #
def test_approval_gate():
print("\n== approval gate ==")
orch, *_ = make_orch_with_paths()
plan = default_plan([
{"wp_id": "W1", "goal": "a", "scope": "s", "acceptance_criteria": "ac",
"target_component": "auth", "requires_approval": True},
])
res = orch.load_plan(plan)
mid = res["mission"]["id"]
orch.missions.mission_transition(mid, "RUNNING", evidence="t")
step = orch.run_one_step(mid)
check("approval required", step["status"] == o.ST_APPROVAL_REQUIRED, step)
ap = step.get("approval", {})
for k in ("MISSION", "WP", "REQUESTED_ACTION", "WHY", "IMPACT", "ROLLBACK", "RISK"):
check(f"approval has {k}", k in ap, ap)
check("no mutation (approval)", orch._wp_state("W1") in ("BLOCKED", "READY"), orch._wp_state("W1"))
# --------------------------------------------------------------------------- #
# 14) risk_overrides_size
# --------------------------------------------------------------------------- #
def test_risk_overrides_size():
print("\n== risk overrides size ==")
# SMALL size but critical target -> CRITICAL
r = classify_risk("ssh", "SMALL")
check("ssh overrides SMALL -> CRITICAL", r == "CRITICAL", r)
r = classify_risk("firewall", "SMALL")
check("firewall overrides SMALL -> CRITICAL", r == "CRITICAL", r)
r = classify_risk("auth", "SMALL")
check("auth overrides SMALL -> CRITICAL", r == "CRITICAL", r)
r = classify_risk("docs", "SMALL")
check("docs stays SMALL", r == "SMALL", r)
r = classify_risk("feature", "LARGE")
check("feature LARGE stays LARGE", r == "LARGE", r)
r = classify_risk("feature")
check("feature default MEDIUM", r == "MEDIUM", r)
# --------------------------------------------------------------------------- #
# 15) no_runnable_stop
# --------------------------------------------------------------------------- #
def test_no_runnable_stop():
print("\n== no runnable stop ==")
orch, *_ = make_orch_with_paths()
plan = default_plan([
{"wp_id": "A", "goal": "a", "scope": "s", "acceptance_criteria": "ac"},
{"wp_id": "B", "goal": "b", "scope": "s", "acceptance_criteria": "ac", "dependencies": ["A"]},
])
res = orch.load_plan(plan)
mid = res["mission"]["id"]
# leave A in READY (not running), B blocked by dep
orch.missions.mission_transition(mid, "RUNNING", evidence="t")
step = orch.run_one_step(mid)
# A is runnable -> would dispatch. To test no_runnable, set A to BLOCKED
orch.missions.wp_transition("A", "IN_PROGRESS")
orch.missions.wp_transition("A", "BLOCKED")
step2 = orch.run_one_step(mid)
check("no runnable -> stop", step2["status"] in (o.ST_NO_RUNNABLE_WORK, o.ST_WAITING_PENDING), step2)
# --------------------------------------------------------------------------- #
# 16) mission_completion_gate / 17) final_review_gate
# --------------------------------------------------------------------------- #
def test_mission_completion_gate():
print("\n== mission completion gate ==")
orch, *_ = make_orch_with_paths()
res = orch.load_plan(default_plan())
mid = res["mission"]["id"]
# not all done + no review -> denied
g = orch.mission_completion_gate(mid, final_review_pass=True)
check("gate blocks incomplete mission", not g["valid"], g)
# mark wp done + review
orch.missions.wp_transition("W1", "IN_PROGRESS")
orch.missions.wp_transition("W1", "CHECKING")
orch.missions.wp_complete("W1")
orch.missions.mission_transition(mid, "RUNNING", evidence="t")
# without final review -> denied
g = orch.mission_completion_gate(mid, final_review_pass=False)
check("final review gate blocks", not g["valid"] and any("review" in r for r in g["reasons"]), g)
# with final review -> complete
res = orch.attempt_mission_complete(mid, final_review_pass=True)
check("mission completed", res["status"] == o.ST_MISSION_COMPLETED, res)
check("mission state COMPLETED", orch._mission_state(mid) == "COMPLETED")
# --------------------------------------------------------------------------- #
# 18) idempotent_wp
# --------------------------------------------------------------------------- #
def test_idempotent_wp():
print("\n== idempotent WP ==")
calls = []
def disp(contract):
calls.append(contract["TASK_ID"])
if contract["role"] == ROLE_MAKER:
return {"role": ROLE_MAKER, "TASK_ID": contract["TASK_ID"], "result": "PASS",
"test_results": [{"verdict": "PASS"}]}
return {"role": ROLE_CHECKER, "verdict": "PASS", "progress": "YES"}
d, mdb, sdb = fresh()
orch = Orchestrator(mdb, sdb, disp)
res = orch.load_plan(default_plan())
mid = res["mission"]["id"]
orch.run_one_step(mid)
orch.run_one_step(mid) # repeat after DONE
check("wp DONE idempotent (no 2nd maker)", sum(1 for t in calls if ":MAKER:" in t) == 1, calls)
check("attempts dedup", orch.safety.attempts_count(mid) == 2, orch.safety.attempts_count(mid)) # 1 maker + 1 checker
# --------------------------------------------------------------------------- #
# 19) restart_resume
# --------------------------------------------------------------------------- #
def test_restart_resume():
print("\n== restart/resume ==")
d, mdb, sdb = fresh()
# phase 1
orch1 = Orchestrator(mdb, sdb)
res = orch1.load_plan(default_plan())
mid = res["mission"]["id"]
# mark wp IN_PROGRESS + record attempt (simulate partial)
orch1.missions.wp_transition("W1", "IN_PROGRESS")
orch1.safety.record_attempt(mid, "W1", actor="maker", result="FAIL", error="crash", strategy="s1")
# "crash": new objects on same DBs
orch2 = Orchestrator(mdb, sdb)
check("mission state persisted", orch2._mission_state(mid) == "RUNNING" or orch2._mission_state(mid) == "READY")
check("attempt persisted", orch2.safety.attempts_count(mid) == 1)
check("wp state persisted", orch2._wp_state("W1") == "IN_PROGRESS")
check("no double run (WPA already IN_PROGRESS)", orch2.run_one_step(mid)["status"] in (o.ST_WAITING_PENDING, o.ST_NO_RUNNABLE_WORK), orch2.run_one_step(mid))
# --------------------------------------------------------------------------- #
# 20) child_failure
# --------------------------------------------------------------------------- #
def test_child_failure():
print("\n== child technical failure ==")
def crash(contract):
raise RuntimeError("child crashed")
orch = Orchestrator(fresh()[1], fresh()[2], crash)
res = orch.load_plan(default_plan())
mid = res["mission"]["id"]
orch.missions.mission_transition(mid, "RUNNING", evidence="t")
step = orch.run_one_step(mid)
check("technical failure", step["status"] == o.ST_CHILD_TECHNICAL_FAILURE, step)
check("no implicit pass", step.get("no_implicit_pass") is True, step)
check("no endless spawn", orch.safety.attempts_count(mid) >= 1)
check("wp not DONE", orch._wp_state("W1") != "DONE")
# --------------------------------------------------------------------------- #
# 21) reason_codes
# --------------------------------------------------------------------------- #
def test_reason_codes():
print("\n== reason codes machine-readable ==")
check("RUN_CODES set", isinstance(o.RUN_CODES, frozenset) and len(o.RUN_CODES) >= 15, o.RUN_CODES)
check("status codes consistent", all(c in o.RUN_CODES for c in (
o.ST_PLAN_VALIDATED, o.ST_TERMINAL, o.ST_NO_RUNNABLE_WORK, o.ST_MUTATION_BLOCKED,
o.ST_APPROVAL_REQUIRED, o.ST_MISSION_COMPLETED, o.ST_NEEDS_DISPATCH)))
# deterministic A3 decision reason codes
orch, *_ = make_orch_with_paths()
orch.load_plan(default_plan())
orch.safety.open_circuit("MISSION", "RQ-A4-1", trigger="R", severity="HIGH", mission_id="RQ-A4-1")
dec = orch.safety.evaluate_next_action("RQ-A4-1", "W1")
check("reason code present", dec["REASON_CODE"] in ("CIRCUIT_ALREADY_OPEN", "CIRCUIT_BREAK") and dec["DECISION"] in ("BLOCK",), dec)
# --------------------------------------------------------------------------- #
# 22) secret_safe_logging
# --------------------------------------------------------------------------- #
def test_secret_safe_logging():
print("\n== secret-safe logging ==")
orch, *_ = make_orch_with_paths()
orch.load_plan(default_plan())
# evidence record with a credential-like value -> redacted
ev = o.make_child_evidence("M", "W", ROLE_MAKER, "t", "2024-01-01", "dispatched",
"PENDING", evidence_ref="token=ghp_ABC123secret")
blob = json.dumps(ev)
check("no raw secret in evidence", "ghp_ABC123secret" not in blob, blob)
check("no credential value", "token=ghp" not in blob, blob)
check("REDACTED marker or safe", "REDACTED" in blob or "ghp_ABC" not in blob)
# contract builder redacts secrets in goal
c = build_maker_contract({"id": "W", "goal": "pw=Sup3rSec", "scope": "s",
"acceptance_criteria": "ac", "test_requirements": "", "safety_rules": [],
"allowed_files": [], "forbidden_files": []},
{"id": "M"}, "MEDIUM")
check("maker contract secret redacted", "Sup3rSec" not in json.dumps(c), json.dumps(c))
# --------------------------------------------------------------------------- #
# Helper dispatchers
# --------------------------------------------------------------------------- #
def _pass_dispatcher():
def disp(contract):
if contract.get("role") == ROLE_MAKER:
return {"role": ROLE_MAKER, "TASK_ID": contract["TASK_ID"], "result": "PASS",
"test_results": [{"verdict": "PASS"}], "progress": "YES"}
return {"role": ROLE_CHECKER, "verdict": "PASS", "progress": "YES"}
return disp
def fail_dispatcher():
"""Checker FAIL mit messbarem Fortschritt -> A3 DECISION=RETRY (nicht DEBUG)."""
def disp(contract):
if contract.get("role") == ROLE_MAKER:
return {"role": ROLE_MAKER, "TASK_ID": contract["TASK_ID"], "result": "PASS",
"test_results": [{"verdict": "PASS"}], "progress": "YES"}
return {"role": ROLE_CHECKER, "verdict": "FAIL", "fail_reason": "AC not met",
"strategy": "strategy-X", "progress": "YES"}
return disp
# --------------------------------------------------------------------------- #
# Helpers
# --------------------------------------------------------------------------- #
def make_orch_with_paths():
d = tempfile.mkdtemp(prefix="a4h_")
mdb = str(Path(d) / "missions.db")
sdb = str(Path(d) / "safety.db")
return Orchestrator(mdb, sdb), mdb, sdb
def make_orch(*dispatcher, **kw):
d = tempfile.mkdtemp(prefix="a4o_")
mdb = str(Path(d) / "missions.db")
sdb = str(Path(d) / "safety.db")
disp = dispatcher[0] if dispatcher else None
return Orchestrator(mdb, sdb, disp, **kw)
def fresh_db():
d = tempfile.mkdtemp(prefix="a4f_")
return str(Path(d) / "missions.db"), str(Path(d) / "safety.db")
# --------------------------------------------------------------------------- #
# Test Runner
# --------------------------------------------------------------------------- #
ALL_TESTS = [
test_plan_validation,
test_dependency_validation,
test_runnable_selection,
test_a2a3_consistency,
test_maker_dispatch_contract,
test_checker_dispatch_contract,
test_checker_pass,
test_checker_fail,
test_retry_allowed,
test_retry_denied_and_4_impossible,
test_circuit_open_blocks_maker,
test_approval_gate,
test_risk_overrides_size,
test_no_runnable_stop,
test_mission_completion_gate,
test_idempotent_wp,
test_restart_resume,
test_child_failure,
test_reason_codes,
test_secret_safe_logging,
]
def main():
for fn in ALL_TESTS:
try:
fn()
except Exception as e: # noqa
global FAIL, FAILURES
FAIL += 1
FAILURES.append(fn.__name__)
print(f" [ERROR] {fn.__name__}: {type(e).__name__}: {e}")
print("\n" + "=" * 60)
print(f"PASS={PASS} FAIL={FAIL}")
if FAIL:
print("FAILURES:", FAILURES)
return 1
print("ALL TESTS PASSED")
return 0
if __name__ == "__main__":
sys.exit(main())
Reference in a new issue View git blame Copy permalink