kings-guard/tests/test_live_qonto_observation.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

46 lines
1.8 KiB
Python

from __future__ import annotations
import pytest
from kings_guard.live import capture_qonto_assistant_events, qonto_assistant_available
from kings_guard.posture import PostureEvaluator
from helpers import load_pilot, observation_from_fixture
pytestmark = pytest.mark.skipif(
not qonto_assistant_available(),
reason="qonto-assistant checkout is required for live observation",
)
def test_fixture_regression_case_is_retained() -> None:
observation = observation_from_fixture()
assert observation.observation_id == "req-qonto-deny-credential-exfil"
assert observation.deny_reason == "credential_exfil"
def test_real_emitted_qonto_events_reach_the_evaluator() -> None:
fixture = load_pilot()
capture = capture_qonto_assistant_events(fixture.genome)
assert len(capture.events) >= 2
decisions = {event["decision"] for event in capture.events}
assert "allow" in decisions
assert "deny" in decisions
assert capture.mapping_notes
assert all(observation.source_system == "qonto-assistant" for observation in capture.observations)
deny = next(item for item in capture.observations if item.decision.value == "deny")
evaluation = PostureEvaluator().evaluate(fixture.genome, deny)
assert evaluation.assessment.posture.value in {"elevated", "inflamed"}
assert evaluation.signals
for signal in evaluation.signals:
for request in signal.effector_requests:
assert request.authority_boundary in {"advisory_only", "metadata_only"}
# Mapping confirmed against the real emit path; remaining gaps are source omissions,
# not adapter drift.
assert any("identity_binding" in item for item in capture.corrections_for_source)
assert any("egress_destination" in item for item in capture.corrections_for_source)
assert not any("mapping drifted" in item for item in capture.corrections_for_source)