net-kingdom/tools/coordination-frame/tests/test_render.py

import hashlib
import importlib.util
import json
import shutil
from pathlib import Path

import pytest
import yaml

SCRIPT = Path(__file__).resolve().parents[1] / "render.py"
SPEC = importlib.util.spec_from_file_location("frame_renderer", SCRIPT)
renderer = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(renderer)
WORKSPACE = SCRIPT.parents[3]


@pytest.fixture
def workspace(tmp_path):
    required = (
        "coordination-engine/config/coordination-frame.json",
        "coordination-engine/src/coordination_engine/timing.py",
        "approval-engine/cadence.yaml",
        "approval-engine/deploy/approval-engine.yaml",
        "qonto-assistant/specs/audit-emission-cadence.yaml",
        "rapp-qonto/runtime/knative/service.yaml",
        "kings-guard/src/kings_guard/fixtures/qonto_assistant_source_cadence.json",
    )
    if any(not (WORKSPACE / name).is_file() for name in required):
        pytest.skip("sibling owner sources required; no fallback frame or contract")
    for name in (
        "coordination-engine/config/coordination-frame.json",
        "coordination-engine/src/coordination_engine/timing.py",
        "approval-engine/cadence.yaml",
        "approval-engine/deploy/approval-engine.yaml",
        "qonto-assistant/specs/audit-emission-cadence.yaml",
        "rapp-qonto/runtime/knative/service.yaml",
        "kings-guard/src/kings_guard/fixtures/qonto_assistant_source_cadence.json",
    ):
        path = tmp_path / name
        path.parent.mkdir(parents=True, exist_ok=True)
        shutil.copyfile(WORKSPACE / name, path)
    return tmp_path


@pytest.mark.parametrize("frame_seconds", [600, 3600, 86400])
def test_one_frame_edit_rescales_sources_runtime_and_receiver(workspace, frame_seconds):
    config = workspace / renderer.FRAME_REF
    config.write_text(
        json.dumps({"schema_version": "coordination-frame/v1", "frame_seconds": frame_seconds})
    )
    _, outputs = renderer.compile_bindings(workspace)
    approval = json.loads(outputs[workspace / "approval-engine/cadence.yaml"])
    qonto = yaml.safe_load(outputs[workspace / "qonto-assistant/specs/audit-emission-cadence.yaml"])
    for entry in approval["sources"]:
        if "heartbeat" in entry:
            assert entry["heartbeat"]["interval_seconds"] == frame_seconds
            extension = entry["heartbeat"]["extensions"]["approval-engine"]
            assert extension["requested_observer_max_gap_seconds"] == 2 * frame_seconds
            assert extension["effective_deadline_seconds"] == 3 * frame_seconds
    for entry in qonto["sources"]:
        assert entry["heartbeat"]["interval"] == f"PT{frame_seconds}S"
    observer = json.loads(
        outputs[
            workspace / "kings-guard/src/kings_guard/fixtures/qonto_assistant_source_cadence.json"
        ]
    )
    assert observer["sources"][0]["heartbeat"]["interval_seconds"] == frame_seconds

    receiver = json.loads(
        outputs[
            workspace / "net-kingdom/generated/coordination-frame/approval-engine-receiver.json"
        ]
    )
    assert set(receiver["heartbeat_classes"].values()) == {2 * frame_seconds}
    assert receiver["effective_deadline_seconds"] == 3 * frame_seconds
    assert receiver["may_read"] is False
    assert receiver["status"] == "proposed-not-applied"
    for name in (
        "approval-engine/approval_engine/coordination.py",
        "qonto-assistant/src/qonto_assistant/coordination.py",
    ):
        assert f"FRAME_SECONDS = {frame_seconds}\n" in outputs[workspace / name]
    deploy = yaml.safe_load_all(outputs[workspace / "approval-engine/deploy/approval-engine.yaml"])
    workload = next(item for item in deploy if item["kind"] == "StatefulSet")
    expected_digest = hashlib.sha256(config.read_bytes()).hexdigest()
    assert (
        workload["spec"]["template"]["metadata"]["annotations"][
            "coordination.coulomb.social/frame-sha256"
        ]
        == expected_digest
    )
    args = workload["spec"]["template"]["spec"]["containers"][0]["args"]
    assert args[args.index("--heartbeat-seconds") + 1] == str(frame_seconds)
    service = yaml.safe_load(outputs[workspace / "rapp-qonto/runtime/knative/service.yaml"])
    assert (
        service["spec"]["template"]["metadata"]["annotations"][
            "coordination.coulomb.social/frame-sha256"
        ]
        == expected_digest
    )
    env = service["spec"]["template"]["spec"]["containers"][0]["env"]
    assert next(
        item["value"] for item in env if item["name"] == "QONTO_AUDIT_HEARTBEAT_INTERVAL_SECONDS"
    ) == str(frame_seconds)


def test_render_is_read_only_and_write_is_idempotent_with_drift_detection(workspace, capsys):
    before = {path: path.read_bytes() for path in workspace.rglob("*") if path.is_file()}
    assert renderer.main(["--workspace", str(workspace)]) == 0
    assert all(path.read_bytes() == data for path, data in before.items())
    assert renderer.main(["--workspace", str(workspace), "--check"]) == 1
    assert renderer.main(["--workspace", str(workspace), "--write"]) == 0
    assert renderer.main(["--workspace", str(workspace), "--check"]) == 0
    receiver = workspace / "net-kingdom/generated/coordination-frame/approval-engine-receiver.json"
    data = json.loads(receiver.read_text())
    data["heartbeat_classes"]["revocation"] = 999999
    receiver.write_text(json.dumps(data))
    assert renderer.main(["--workspace", str(workspace), "--check"]) == 1


def test_invalid_frame_does_not_partially_rewrite_sources(workspace):
    config = workspace / renderer.FRAME_REF
    config.write_text('{"schema_version":"coordination-frame/v1","frame_seconds":0}')
    before = {path: path.read_bytes() for path in workspace.rglob("*") if path.is_file()}
    with pytest.raises(SystemExit):
        renderer.main(["--workspace", str(workspace), "--write"])
    assert all(path.read_bytes() == data for path, data in before.items())