DataOps-platform/tests/agent/test_wp09_settlement_contract.py

from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
MIGRATION = ROOT / "migrations/versions/20260813_498_agent_runtime_settlement.py"
DEPLOYMENT = ROOT / "deployment/migrations/versions/20260813_498_agent_runtime_settlement.py"


def test_wp09_settlement_migration_is_fenced_closed_and_not_public():
    source = MIGRATION.read_text(encoding="utf-8")
    assert 'revision = "20260813_498"' in source
    assert 'down_revision = "20260813_497"' in source
    for required in (
        "agent_runtime_settle", "lease_expires_at", "worker_id", "settlement_digest",
        "REVOKE ALL ON FUNCTION public.agent_runtime_authorize(jsonb) FROM PUBLIC",
        "REVOKE ALL ON FUNCTION public.agent_runtime_transition",
        "REVOKE ALL ON FUNCTION public.agent_runtime_canary_transition",
        "REVOKE ALL ON FUNCTION public.agent_runtime_approval_valid",
        "GRANT EXECUTE ON FUNCTION public.agent_runtime_settle(jsonb) TO dataops_agent_runtime",
        "SET search_path=pg_catalog,public", "p_payload-ARRAY", "approval agent scope denied",
        "approval principal scope denied", "lease_expired", "settlement fence conflict",
    ):
        assert required in source
    assert MIGRATION.read_bytes() == DEPLOYMENT.read_bytes()