coulomb/qonto-assistant
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(audit): publish sequenced heartbeat and reconciliation evidence

Browse source 
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a06ec5-7e2b-7743-ac08-719e1b0f42e2
This commit is contained in:
tegwick 2026-09-05 01:39:48 +02:00
parent fce60099c4
commit b9349782f4
32 changed files with 839 additions and 111 deletions
Download patch file Download diff file Expand all files Collapse all files

31
tests/test_audit_parity.py
View file

@ -18,18 +18,35 @@ from qonto_assistant.rate_limits import ConcurrencyLimiter, RateLimiter
from qonto_assistant.service import CapabilityService
FIXTURE_DIR = Path(__file__).resolve().parent / "fixtures" / "qonto"
POLICY_FILE = Path(__file__).resolve().parents[1] / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml"
POLICY_FILE = (
Path(__file__).resolve().parents[1] / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml"
)
# Fields whose values are expected to vary per call (timing/identifiers) or
# by design (protocol). Everything else must match exactly between REST and
# MCP for the same logical call.
NON_COMPARABLE_FIELDS = {"request_id", "timestamp", "latency_ms", "protocol"}
SECRET_FIELD_NAMES = {"api_key", "authorization", "authorization_header", "openbao_token", "secret", "token"}
NON_COMPARABLE_FIELDS = {
"request_id",
"timestamp",
"latency_ms",
"protocol",
"stream_sequence",
}
SECRET_FIELD_NAMES = {
"api_key",
"authorization",
"authorization_header",
"openbao_token",
"secret",
"token",
}
def _service() -> tuple[CapabilityService, list[dict[str, object]]]:
events: list[dict[str, object]] = []
policy = PolicyEngine.from_file(POLICY_FILE, required_scope="finance.qonto.read", enforce_scope=False)
policy = PolicyEngine.from_file(
POLICY_FILE, required_scope="finance.qonto.read", enforce_scope=False
)
service = CapabilityService(
client=FixtureQontoClient(fixture_dir=FIXTURE_DIR),
policy=policy,
@ -56,10 +73,14 @@ def test_allow_path_audit_schema_identical_across_protocols() -> None:
assert rest_event["protocol"] == "rest"
assert mcp_event["protocol"] == "mcp"
for field in set(rest_event) - NON_COMPARABLE_FIELDS:
assert rest_event[field] == mcp_event[field], f"field {field!r} diverged: {rest_event[field]!r} != {mcp_event[field]!r}"
assert rest_event[field] == mcp_event[field], (
f"field {field!r} diverged: {rest_event[field]!r} != {mcp_event[field]!r}"
)
assert rest_event["decision"] == "allow"
assert rest_event["capability"] == "org_summary"
assert rest_event["identity_binding"] == "self_asserted"
assert rest_event["egress_destination"] == "qonto-thirdparty-api"
def test_deny_path_audit_schema_identical_across_protocols() -> None: