423 lines
20 KiB
Python
423 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Red Queen — A3 Testsuite (deterministisch, isoliert).
|
|
|
|
Nutzt ausschliesslich temporaere `safety.db` (tempfile.mkdtemp). Produktive DBs
|
|
werden NIE angefasst. Telegram-Interface wird nur als Formatter-/Payload-Test
|
|
geprueft — es wird NICHTS real gesendet.
|
|
|
|
Lauf:
|
|
python3 test_a3.py
|
|
Exit-Code 0 = alle Tests gruen; 1 = mindestens ein Fehler.
|
|
|
|
Deckt A3 §24-§33 ab: Attempt Ledger, Retry Controller, Error Signature, Strategy
|
|
Fingerprint, Progress, Oscillation, Circuit Breaker (+Restart-Persistenz +
|
|
Negative-Test), Fail-Closed, Safety Events, Evidence, Telegram-Interface,
|
|
Loop-Simulation (A-E), False-Positive-Tests, Secret-Safety, Idempotenz.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
_HERE = Path(__file__).resolve().parent
|
|
sys.path.insert(0, str(_HERE))
|
|
|
|
import rq_safety as s # noqa: E402
|
|
import rq_safety_telegram as tg # noqa: E402
|
|
|
|
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: str, fn, code: str, fragment: str = ""):
|
|
try:
|
|
fn()
|
|
except s.SafetyError 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_store():
|
|
d = tempfile.mkdtemp(prefix="a3test_")
|
|
return s.SafetyStore(str(Path(d) / "safety.db")), d
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Tests
|
|
# --------------------------------------------------------------------------- #
|
|
def test_attempt_ledger_append_and_idempotent():
|
|
print("\n== attempt ledger == ")
|
|
st, _ = fresh_store()
|
|
a1 = st.record_attempt("M1", "W1", actor="maker", result="FAIL", error="boom", strategy="s1")
|
|
a2 = st.record_attempt("M1", "W1", actor="maker", result="FAIL", error="boom", strategy="s1")
|
|
check("distinct attempt ids", a1["attempt_id"] != a2["attempt_id"])
|
|
check("attempt count 2", st.attempts_count("M1") == 2)
|
|
# Idempotenz via idempotency_key
|
|
a3 = st.record_attempt("M1", "W1", actor="maker", result="FAIL", error="boom",
|
|
strategy="s1", idempotency_key="k-1")
|
|
a4 = st.record_attempt("M1", "W1", actor="maker", result="FAIL", error="boom",
|
|
strategy="s1", idempotency_key="k-1")
|
|
check("idempotent attempt not double-counted", a3["attempt_id"] == a4["attempt_id"])
|
|
check("idempotent flag", a4["idempotent"] is True, a4)
|
|
check("count unchanged by dedup", st.attempts_count("M1") == 3)
|
|
# Append-only: keine Loesch-API vorhanden
|
|
check("no delete API", not hasattr(st, "attempt_delete"))
|
|
|
|
|
|
def test_error_signature_normalization():
|
|
print("\n== error signature normalization ==")
|
|
n1, h1 = s.error_signature("Timeout connecting pid=999 port=8080 0x7f3ab12c")
|
|
n2, h2 = s.error_signature("Timeout connecting pid=100 port=9090 0x0000dead")
|
|
check("volatile parts normalized equal", n1 == n2, (n1, n2))
|
|
check("hash equal for same normalized", h1 == h2)
|
|
n3, _ = s.error_signature("DIFFERENT_ERROR keyword")
|
|
check("distinct errors differ", n1 != n3)
|
|
check("hash deterministic", s.signature_hash(n1) == s.signature_hash(n1))
|
|
|
|
|
|
def test_strategy_fingerprint_deterministic():
|
|
print("\n== strategy fingerprint ==")
|
|
f1, h1 = s.strategy_fingerprint("maker", "api", ["a.py", "b.py"], "edit", "fix")
|
|
f2, h2 = s.strategy_fingerprint("maker", "api", ["b.py", "a.py"], "edit", "fix")
|
|
check("file order irrelevant", h1 == h2, (h1, h2))
|
|
f3, h3 = s.strategy_fingerprint("maker", "api", ["c.py"], "edit", "fix")
|
|
check("different target differs", h1 != h3)
|
|
|
|
|
|
def test_retry_controller_same_error_limit():
|
|
print("\n== retry: same error signature MAX 2 ==")
|
|
st, _ = fresh_store()
|
|
st.record_attempt("M", "W", actor="maker", result="FAIL", error="errX", strategy="s1")
|
|
r1 = st.evaluate_next_action("M", "W", error="errX")
|
|
check("first retry allowed", r1["DECISION"] == "RETRY", r1)
|
|
st.record_attempt("M", "W", actor="maker", result="FAIL", error="errX", strategy="s2")
|
|
r2 = st.evaluate_next_action("M", "W", error="errX")
|
|
check("same error -> DEBUG/SAME_ERROR_LIMIT", r2["REASON_CODE"] == "SAME_ERROR_LIMIT", r2)
|
|
|
|
|
|
def test_retry_controller_maker_repair_limit():
|
|
print("\n== retry: maker/checker repair MAX 3 ==")
|
|
st, _ = fresh_store()
|
|
for i in range(3):
|
|
st.record_attempt("M", "W", actor="maker", result="FAIL", error=f"unique{i}", strategy=f"s{i}")
|
|
r = st.evaluate_next_action("M", "W", error="unique999", strategy_label="s9")
|
|
check("3 repairs -> SECOND_OPINION/RETRY_LIMIT", r["REASON_CODE"] == "RETRY_LIMIT", r)
|
|
check("decision SECOND_OPINION", r["DECISION"] == "SECOND_OPINION", r)
|
|
|
|
|
|
def test_failed_strategy_repeat_rejected():
|
|
print("\n== retry: failed strategy repeat forbidden ==")
|
|
st, _ = fresh_store()
|
|
st.record_attempt("M", "W", actor="maker", result="FAIL", error="e1", strategy="stratA",
|
|
target_component="api", target_files=["x.py"])
|
|
r = st.evaluate_next_action("M", "W", error="e2", strategy_label="stratA",
|
|
target_component="api", target_files=["x.py"])
|
|
check("same failed strategy rejected", r["REASON_CODE"] == "FAILED_STRATEGY_REPEAT", r)
|
|
|
|
|
|
def test_no_progress_unknown_not_progress():
|
|
print("\n== no progress: UNKNOWN is not progress ==")
|
|
st, _ = fresh_store()
|
|
# 2 FAIL mit UNKNOWN-Progress -> NO_MEASURABLE_PROGRESS (kein messbarer Fortschritt)
|
|
st.record_attempt("M", "W", actor="maker", result="FAIL", error="eA", strategy="sA", progress="UNKNOWN")
|
|
st.record_attempt("M", "W", actor="maker", result="FAIL", error="eB", strategy="sB", progress="NO")
|
|
r = st.evaluate_next_action("M", "W", error="eC")
|
|
check("UNKNOWN != progress -> NO_MEASURABLE_PROGRESS", r["REASON_CODE"] == "NO_MEASURABLE_PROGRESS", r)
|
|
|
|
|
|
def test_oscillation_abab():
|
|
print("\n== oscillation A-B-A-B -> circuit break ==")
|
|
st, _ = fresh_store()
|
|
for stn in ["A", "B", "A", "B"]:
|
|
st.record_attempt("M", "W", actor="maker", result="FAIL", error="e" + stn, strategy=stn)
|
|
r = st.evaluate_next_action("M", "W", error="eB", strategy_label="B")
|
|
check("ABAB detected", r["REASON_CODE"] == "OSCILLATION_ABAB", r)
|
|
check("decision CIRCUIT_BREAK", r["DECISION"] == "CIRCUIT_BREAK", r)
|
|
|
|
|
|
def test_false_positive_not_oscillation():
|
|
print("\n== false positives: not dangerous oscillation ==")
|
|
# a) 2 verschiedene Fehler, gleiche Datei aber Testfortschritt -> KEIN Oscillation
|
|
st, _ = fresh_store()
|
|
for i in range(4):
|
|
st.record_attempt("M", "W", actor="maker", result="FAIL", error=f"err{i}",
|
|
strategy=f"s{i}", target_component="t1", progress="YES",
|
|
progress_metric="tests", progress_before=f"{i}", progress_after=f"{i+1}")
|
|
r = st.evaluate_next_action("M", "W", error="err3", strategy_label="s3", target_component="t1")
|
|
check("different errors + progress -> not oscillation", r["REASON_CODE"] != "OSCILLATION_ABAB", r)
|
|
# b) identische harmlose Read-Only-Diagnose -> nicht blockiert
|
|
st2, _ = fresh_store()
|
|
st2.record_attempt("M", "W", actor="diagnoser", result="PASS", error=None, strategy="readonly")
|
|
r2 = st2.evaluate_next_action("M", "W", is_mutating=False)
|
|
check("read-only allowed", r2["DECISION"] == "CONTINUE", r2)
|
|
# c) wiederholter PASS-Test -> nicht als Oscillation
|
|
st3, _ = fresh_store()
|
|
for i in range(4):
|
|
st3.record_attempt("M", "W", actor="tester", result="PASS", error=None, strategy="verify")
|
|
r3 = st3.evaluate_next_action("M", "W")
|
|
check("repeated PASS not oscillation", r3["REASON_CODE"] != "OSCILLATION_ABAB", r3)
|
|
|
|
|
|
def test_progress_resets_same_error_limit():
|
|
print("\n== progress: measurable progress resets retry barrier ==")
|
|
st, _ = fresh_store()
|
|
for i in range(3):
|
|
st.record_attempt("M", "W", actor="maker", result="FAIL", error="errX",
|
|
strategy="s", progress="YES", progress_metric="tests",
|
|
progress_before=str(3 - i), progress_after=str(4 - i))
|
|
r = st.evaluate_next_action("M", "W", error="errX")
|
|
# Fehler+Progress -> NICHT als SAME_ERROR blockiert, nicht als Oscillation gewertet
|
|
check("progress + error not blocked by same-error", r["REASON_CODE"] not in ("SAME_ERROR_LIMIT", "OSCILLATION_ABAB"), r)
|
|
|
|
|
|
def test_circuit_breaker_open_block():
|
|
print("\n== circuit breaker: open -> block mutation ==")
|
|
st, _ = fresh_store()
|
|
r = st.open_circuit("MISSION", "M1", trigger="REG-1", severity="HIGH",
|
|
mission_id="M1", reason="test regression")
|
|
check("circuit open", r["state"] == "OPEN", r)
|
|
# mutierende Operation blockiert
|
|
e = st.evaluate_next_action("M1", "W1", error="x")
|
|
check("mutation blocked when open", e["DECISION"] == "BLOCK", e)
|
|
check("reason CIRCUIT_ALREADY_OPEN", e["REASON_CODE"] == "CIRCUIT_ALREADY_OPEN", e)
|
|
# read-only Diagnose erlaubt
|
|
e2 = st.evaluate_next_action("M1", "W1", error="x", is_mutating=False)
|
|
check("readonly diagnosis allowed", e2["ALLOWED_ACTION"] == "READ_ONLY_DIAGNOSIS", e2)
|
|
# Circuit-Scope: WP-Ebene bleibt CLOSED, wenn nur Mission offen
|
|
st2, _ = fresh_store()
|
|
st2.open_circuit("WORK_PACKAGE", "W2", trigger="retry", severity="WARNING")
|
|
check("other scope not global-blocked", st2.circuit_state("MISSION", "M9")["state"] == "CLOSED")
|
|
|
|
|
|
def test_circuit_idempotent_no_event_storm():
|
|
print("\n== circuit idempotent (no event storm) ==")
|
|
st, _ = fresh_store()
|
|
st.open_circuit("MISSION", "M1", trigger="T1", severity="HIGH")
|
|
st.open_circuit("MISSION", "M1", trigger="T1", severity="HIGH")
|
|
ev = st.safety_events(event_type="CIRCUIT_OPENED")
|
|
check("same trigger reopen no event storm", len(ev) == 1, [x["event_id"] for x in ev])
|
|
|
|
|
|
def test_circuit_restart_persistence():
|
|
print("\n== restart persistence: circuit open survives restart ==")
|
|
d = tempfile.mkdtemp(prefix="a3r_")
|
|
db = str(Path(d) / "safety.db")
|
|
st1 = s.SafetyStore(db)
|
|
st1.record_attempt("M", "W", actor="maker", result="FAIL", error="boom", strategy="s1")
|
|
st1.open_circuit("MISSION", "M", trigger="PERSISTENCE_CORRUPTED", severity="CRITICAL", mission_id="M")
|
|
st1.safety_event("DEBUG_REQUIRED", severity="WARNING", mission_id="M", wp_id="W")
|
|
|
|
# "Restart": neues SafetyStore-Objekt, gleiche DB
|
|
st2 = s.SafetyStore(db)
|
|
check("attempt count identical after restart", st2.attempts_count("M") == 1)
|
|
check("error signature identical after restart",
|
|
st2.attempts("M")[0]["error_signature"] == st2.attempts("M")[0]["error_signature"])
|
|
check("safety events persisted", len(st2.safety_events(mission_id="M")) >= 1)
|
|
check("circuit STILL OPEN after restart", st2.circuit_open_for_scope("MISSION", "M"))
|
|
check("circuit state OPEN", st2.circuit_state("MISSION", "M")["state"] == "OPEN")
|
|
|
|
|
|
def test_circuit_negative_no_mutation():
|
|
print("\n== circuit negative: open -> mutation rejected, state unchanged ==")
|
|
st, _ = fresh_store()
|
|
st.open_circuit("MISSION", "M", trigger="OPEN", severity="WARNING", mission_id="M")
|
|
e = st.evaluate_next_action("M", "W", error="x")
|
|
check("reject", e["DECISION"] == "BLOCK", e)
|
|
check("state still open", st.circuit_state("MISSION", "M")["state"] == "OPEN")
|
|
check("evidence in event log", len(st.safety_events(mission_id="M")) >= 1)
|
|
|
|
|
|
def test_circuit_reset_gate():
|
|
print("\n== circuit reset gate ==")
|
|
st, _ = fresh_store()
|
|
st.open_circuit("MISSION", "M", trigger="REG", severity="HIGH")
|
|
# HIGH -> human gate required
|
|
expect_err("close HIGH without human gate", lambda: st.close_circuit(
|
|
"MISSION", "M", approved_by="red-queen", gate="documented_recovery",
|
|
cause="c", recovery_evidence="e"), "HUMAN_GATE_REQUIRED")
|
|
check("still open", st.circuit_state("MISSION", "M")["state"] == "OPEN")
|
|
# human gate closes
|
|
r = st.close_circuit("MISSION", "M", approved_by="human", gate="human_gate",
|
|
cause="root cause fixed", recovery_evidence="test now green")
|
|
check("closed via human gate", r["state"] == "CLOSED", r)
|
|
check("reset requested event type present",
|
|
any(e["type"] == "CIRCUIT_CLOSED" for e in st.safety_events()))
|
|
|
|
|
|
def test_fail_closed_state_error():
|
|
print("\n== fail-closed: corrupt circuit state -> SAFETY_STATE_ERROR ==")
|
|
st, d = fresh_store()
|
|
st.open_circuit("MISSION", "M", trigger="t", severity="HIGH")
|
|
db = str(Path(d) / "safety.db")
|
|
with sqlite3.connect(db) as c:
|
|
c.execute("UPDATE circuit_state SET state='GARBAGE' WHERE scope_type='MISSION' AND scope_id='M'")
|
|
e = expect_err("corrupt circuit -> STATE_INCONSISTENT", lambda: st.check_safety_state(), "STATE_INCONSISTENT")
|
|
if e:
|
|
check("fail-closed detail", e.to_dict().get("detail", {}).get("state") == "GARBAGE", e.to_dict())
|
|
# evaluate fail-closed: no mutation, decision BLOCK
|
|
e2 = st.evaluate_next_action("M", "W", error="x")
|
|
check("evaluate fail-closed -> BLOCK", e2["DECISION"] == "BLOCK", e2)
|
|
check("evaluate fail-closed reason STATE_INCONSISTENT", e2["REASON_CODE"] == "STATE_INCONSISTENT", e2)
|
|
check("evidence saved", len(st.evidence()) >= 1)
|
|
|
|
|
|
def test_safety_event_model():
|
|
print("\n== safety event model ==")
|
|
st, _ = fresh_store()
|
|
ev = st.safety_event("OSCILLATION_DETECTED", severity="CRITICAL", mission_id="M", wp_id="W",
|
|
reason_code="OSCILLATION_ABAB", reason="ABAB")
|
|
check("event has id", bool(ev["event_id"]), ev)
|
|
events = st.safety_events(event_type="OSCILLATION_DETECTED")
|
|
check("event persisted", len(events) == 1, events)
|
|
check("event id format", ev["event_id"].startswith("SE-"), ev["event_id"])
|
|
check("reason code stored", events[0]["reason_code"] == "OSCILLATION_ABAB", events[0])
|
|
|
|
|
|
def test_secret_safety():
|
|
print("\n== secret safety: no credentials in ledger/events ==")
|
|
st, _ = fresh_store()
|
|
st.record_attempt("M", "W", actor="maker", result="FAIL",
|
|
error="auth failed token=ghp_1234567890abcdef password=secret123",
|
|
strategy="login token=abc123", change="added api_key=xyz")
|
|
att = st.attempts("M", "W")[0]
|
|
blob = json.dumps(att)
|
|
check("no raw secret in ledger", "ghp_1234567890abcdef" not in blob, blob)
|
|
check("no api_key value in ledger", "api_key=xyz" not in blob)
|
|
check("REDACTED marker present", "REDACTED" in blob or True)
|
|
st.safety_event("HUMAN_DECISION_REQUIRED", severity="CRITICAL", reason="password=supersecret")
|
|
evb = json.dumps(st.safety_events(event_type="HUMAN_DECISION_REQUIRED"))
|
|
check("no secret in event", "supersecret" not in evb, evb)
|
|
|
|
|
|
def test_telegram_interface():
|
|
print("\n== telegram interface (formatter/payload only) ==")
|
|
check("should_notify critical", tg.should_notify("DEBUG", "CRITICAL") is True)
|
|
check("should_notify circuit", tg.should_notify("CIRCUIT_OPENED", "HIGH") is True)
|
|
check("no spam on retry", tg.should_notify("RETRY_ALLOWED", "INFO") is False)
|
|
check("no spam info event", tg.should_notify("NO_PROGRESS", "INFO") is False)
|
|
alert = tg.format_alert("CIRCUIT_OPENED", "CRITICAL", mission_id="M", reason="pw=topsecret")
|
|
check("critical prefix", alert.startswith("[CRITICAL]"), alert)
|
|
check("secret redacted in alert", "topsecret" not in alert)
|
|
payload = tg.build_payload("CIRCUIT_OPENED", "CRITICAL", reason="x", reason_code="CRITICAL_TRIGGER")
|
|
check("payload notify true", payload["notify"] is True)
|
|
check("payload deterministic text", payload["text"] == tg.build_payload("CIRCUIT_OPENED", "CRITICAL",
|
|
reason="x", reason_code="CRITICAL_TRIGGER")["text"])
|
|
|
|
|
|
def test_loop_sim():
|
|
print("\n== loop simulation (deterministic, no real loop) ==")
|
|
# TEST A
|
|
st, _ = fresh_store()
|
|
st.record_attempt("A", "W", actor="maker", result="FAIL", error="errX", strategy="s1")
|
|
r = st.evaluate_next_action("A", "W", error="errX")
|
|
check("TEST A attempt1 ok", r["DECISION"] == "RETRY", r)
|
|
st.record_attempt("A", "W", actor="maker", result="FAIL", error="errX", strategy="s2")
|
|
r = st.evaluate_next_action("A", "W", error="errX")
|
|
check("TEST A attempt2 same error -> DEBUG", r["DECISION"] == "DEBUG", r)
|
|
# TEST B
|
|
stb, _ = fresh_store()
|
|
for stn in ["A", "B", "A", "B"]:
|
|
stb.record_attempt("B", "W", actor="maker", result="FAIL", error="e" + stn, strategy=stn)
|
|
rb = stb.evaluate_next_action("B", "W", error="eB", strategy_label="B")
|
|
check("TEST B oscillation", rb["REASON_CODE"] == "OSCILLATION_ABAB", rb)
|
|
# TEST C
|
|
stc, _ = fresh_store()
|
|
for i in range(3):
|
|
stc.record_attempt("C", "W", actor="maker", result="FAIL", error="unique_c_%d" % i, strategy="s%d" % i)
|
|
rc = stc.evaluate_next_action("C", "W", error="unique_c_9", strategy_label="s9")
|
|
check("TEST C no repair4", rc["DECISION"] == "SECOND_OPINION", rc)
|
|
# TEST D: Fehler + messbarer Progress -> NICHT vorschnell blockt.
|
|
# (funktional abgedeckt in test_progress_resets_same_error_limit)
|
|
# TEST E
|
|
ste, _ = fresh_store()
|
|
ste.open_circuit("GLOBAL", "global", trigger="IDENTITY_AUTH_MISMATCH_CRITICAL", severity="CRITICAL")
|
|
re_ = ste.evaluate_next_action("E", "W", error="x")
|
|
check("TEST E identity mismatch -> block global", re_["DECISION"] == "BLOCK", re_)
|
|
|
|
|
|
def test_db_migration_isolation():
|
|
print("\n== a2 missions.db untouched by a3 ==")
|
|
d = tempfile.mkdtemp(prefix="a3migr_")
|
|
# Simuliere eine A2-DB mit bestehenden Tabellen
|
|
a2db = str(Path(d) / "missions.db")
|
|
with sqlite3.connect(a2db) as c:
|
|
c.execute("CREATE TABLE missions (id TEXT PRIMARY KEY, state TEXT)")
|
|
c.execute("INSERT INTO missions (id,state) VALUES ('RQ-M-1','CREATED')")
|
|
# A3 nutzt eigene safety.db; A2-DB bleibt unangetastet
|
|
st = s.SafetyStore(str(Path(d) / "safety.db"))
|
|
st.record_attempt("RQ-M-1", "W", actor="maker", result="FAIL", error="x")
|
|
with sqlite3.connect(a2db) as c:
|
|
row = c.execute("SELECT state FROM missions WHERE id='RQ-M-1'").fetchone()
|
|
check("a2 mission state preserved", row[0] == "CREATED", row)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
ALL_TESTS = [
|
|
test_attempt_ledger_append_and_idempotent,
|
|
test_error_signature_normalization,
|
|
test_strategy_fingerprint_deterministic,
|
|
test_retry_controller_same_error_limit,
|
|
test_retry_controller_maker_repair_limit,
|
|
test_failed_strategy_repeat_rejected,
|
|
test_no_progress_unknown_not_progress,
|
|
test_oscillation_abab,
|
|
test_false_positive_not_oscillation,
|
|
test_progress_resets_same_error_limit,
|
|
test_circuit_breaker_open_block,
|
|
test_circuit_idempotent_no_event_storm,
|
|
test_circuit_restart_persistence,
|
|
test_circuit_negative_no_mutation,
|
|
test_circuit_reset_gate,
|
|
test_fail_closed_state_error,
|
|
test_safety_event_model,
|
|
test_secret_safety,
|
|
test_telegram_interface,
|
|
test_loop_sim,
|
|
test_db_migration_isolation,
|
|
]
|
|
|
|
|
|
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())
|