DataOps-platform/tests/agent/test_wp09_control_claim_contract.py

from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
MIGRATION = ROOT / "migrations/versions/20260813_500_runtime_control_claims.py"
DEPLOYMENT = ROOT / "deployment/migrations/versions/20260813_500_runtime_control_claims.py"
ROUTE = ROOT / "app/api/knowledge_base/agent_governance_routes.py"


def test_wp09_control_claim_is_credential_bound_and_not_shared_runtime_callable():
    source = MIGRATION.read_text(encoding="utf-8")
    assert 'revision = "20260813_500"' in source
    assert 'down_revision = "20260813_499"' in source
    for required in (
        "agent_runtime_control_claims", "credential_uid", "credential_token_digest",
        "payload_digest", "nonce", "expires_at", "consumed_at", "FOR UPDATE",
        "agent_runtime_control_issue_claim", "agent_runtime_control_claimed",
        "REVOKE ALL ON FUNCTION public.agent_runtime_control(jsonb) FROM PUBLIC,dataops_agent_runtime,dataops_app_runtime",
        "SET search_path=pg_catalog,public",
    ):
        assert required in source
    assert "GRANT EXECUTE ON FUNCTION public.agent_runtime_control_claimed(jsonb,uuid) TO dataops_app_runtime" in source
    assert "GRANT EXECUTE ON FUNCTION public.agent_runtime_control_claimed(jsonb,uuid) TO dataops_agent_runtime" not in source
    assert source.encode() == DEPLOYMENT.read_bytes()
    route = ROUTE.read_text(encoding="utf-8")
    assert 'runtime/control' in route
    assert "validate_credential(agent_uid, token)" in route
    assert "issue_control_claim" in route and "control_claimed" in route