tegwick
8971a2dc39
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1029d-2c93-7500-a0e0-e3e089668b29
60 lines
2.8 KiB
Python
60 lines
2.8 KiB
Python
"""Validate source claims against existing emitter and owner-supplied contracts."""
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from qonto_assistant.audit import AUDIT_STREAM_ID, AuditLogger
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def test_declaration_preserves_emitter_boundary_and_counters():
|
|
declaration = yaml.safe_load((ROOT / "specs/audit-emission-cadence.yaml").read_text())
|
|
source = declaration["extensions"]["qonto-assistant"]
|
|
assert declaration["stream_id"] == AUDIT_STREAM_ID
|
|
assert source["instance_boundary"] == "process"
|
|
assert source["heartbeat"]["default_interval_seconds"] == 86400
|
|
assert source["reconciliation"]["persistence"] == "process-local"
|
|
events = []
|
|
audit = AuditLogger(sink=events.append)
|
|
audit.emit({"decision": "deny"})
|
|
heartbeat = audit.emit_heartbeat()
|
|
for entry in declaration["sources"]:
|
|
assert entry["heartbeat"]["event_class"] == heartbeat["event_class"]
|
|
assert entry["extensions"]["netkingdom"]["completeness_claimed"] is False
|
|
if "reconciliation" in entry:
|
|
field, event_class = entry["reconciliation"]["compare_local"].split(".", 1)
|
|
assert event_class in heartbeat[field]
|
|
assert heartbeat["source_transition_counts"]["audit.deny"] == 1
|
|
assert heartbeat["assertion"] == "transitions-reported"
|
|
|
|
|
|
def test_declaration_passes_supplied_owner_inventory_against_canonical_schema():
|
|
checker = ROOT.parent / "net-kingdom/tools/emission-cadence-profile/emission_cadence_profile.py"
|
|
schema = ROOT.parent / "info-tech-canon/infospace/schemas/emission-cadence.schema.yaml"
|
|
if not checker.is_file() or not schema.is_file():
|
|
pytest.skip("Owner schema and profile checkout required; no fallback contract")
|
|
pytest.importorskip("jsonschema")
|
|
inventory = yaml.safe_load((ROOT / "specs/audit-event-class-inventory.yaml").read_text())
|
|
command = [sys.executable, str(checker), "--contract-schema", str(schema)]
|
|
for event_class, classification in inventory["classes"].items():
|
|
flag = (
|
|
"--attributive"
|
|
if classification["evidence_class"] == "attributive"
|
|
else "--load-bearing"
|
|
)
|
|
command.extend([flag, event_class])
|
|
if classification.get("rare") and classification["evidence_class"] == "load-bearing":
|
|
command.extend(["--rare-load-bearing", event_class])
|
|
command.append(str(ROOT / "specs/audit-emission-cadence.yaml"))
|
|
completed = subprocess.run(command, capture_output=True, text=True, check=False)
|
|
assert completed.returncode == 0, completed.stdout + completed.stderr
|
|
report = json.loads(completed.stdout)
|
|
assert report["contract_valid"] is True
|
|
assert report["profile_assessed"] is True
|
|
assert report["conformant"] is True
|
|
assert report["assessment_scope"] == "supplied-inventory"
|