kings-guard/tests/test_immune_memory.py
tegwick 9daea96c43 Finish KG-WP-0003: stream completeness and live qonto observation
Classify evidence as load-bearing or attributive, draft the emission-cadence
declaration for Taxonomy, treat silence as a stream finding, keep completeness
separate from record richness, forbid immune memory as a state plane, and make
containment proposals reconstructable to their origin. Observe real
qonto-assistant audit events; deny-class completeness stays unknown until the
source publishes a heartbeat.

Assistant: grok
Assistant-Session: 01a05ef1-9e5a-70f2-b0ff-0b05d6b38ae9
2026-09-02 00:11:57 +02:00

65 lines
2.6 KiB
Python

from __future__ import annotations
import ast
import inspect
from pathlib import Path
from kings_guard.contracts import ImmuneMemoryEntry
from kings_guard.posture import PostureEvaluator
ROOT = Path(__file__).resolve().parents[1]
INTENT = ROOT / "INTENT.md"
CONTRACTS = ROOT / "specs" / "ImmuneContracts.md"
SRC = ROOT / "src" / "kings_guard"
def test_memory_entry_forbids_runtime_dependency_by_other_layers() -> None:
entry = ImmuneMemoryEntry(
memory_id="mem:example",
subject_scope="qonto-assistant",
summary="credential-exfil probe pattern",
derived_from=("req-qonto-deny-credential-exfil",),
recommended_countermeasures=("lock_actor_temporarily",),
)
assert entry.runtime_input_for_other_layers == "forbidden"
assert entry.confidentiality == "non-secret"
def test_posture_evaluator_does_not_accept_memory_as_input() -> None:
for name in ("evaluate", "evaluate_stream", "evaluate_with_stream"):
signature = inspect.signature(getattr(PostureEvaluator, name))
for parameter in signature.parameters.values():
annotation = str(parameter.annotation)
assert "ImmuneMemoryEntry" not in annotation, name
def test_no_src_function_takes_memory_as_runtime_input() -> None:
"""Catch the drift of wiring immune memory into an engine-facing input."""
hits: list[str] = []
for path in SRC.rglob("*.py"):
tree = ast.parse(path.read_text(encoding="utf-8"))
for node in ast.walk(tree):
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
for arg in node.args.args + node.args.kwonlyargs:
annotation = ast.unparse(arg.annotation) if arg.annotation is not None else ""
if "ImmuneMemoryEntry" in annotation and arg.arg not in {"entry", "memory"}:
hits.append(f"{path.name}:{node.name}:{arg.arg}")
if "ImmuneMemoryEntry" in annotation and node.name.startswith("evaluate"):
hits.append(f"{path.name}:{node.name} evaluates from memory")
assert hits == []
def test_intent_stage_five_forbids_memory_as_state_plane() -> None:
text = INTENT.read_text(encoding="utf-8")
collapsed = " ".join(text.split())
assert "Federated memory" in collapsed
assert "without becoming a state plane" in collapsed
assert "no engine, PEP, or workload may read it as" in collapsed
def test_immune_contracts_state_the_state_plane_rule() -> None:
text = CONTRACTS.read_text(encoding="utf-8")
assert "not a state plane" in text.lower()
assert "runtime_input_for_other_layers" in text
assert "Tooling catalog change" in text