DataOps-platform/tests/agent/test_wp09_repository_contract.py

from __future__ import annotations

from pathlib import Path

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


def test_wp09_migration_is_chained_after_current_head_has_fenced_budget_and_no_raw_payload_columns():
    source = MIGRATION.read_text(encoding="utf-8")

    assert 'revision = "20260813_496"' in source
    assert 'down_revision = "20260811_495"' in source
    for table in (
        "model_gateway_routes", "agent_runtime_budgets", "agent_runtime_reservations",
        "agent_runtime_capabilities", "agent_runtime_states", "agent_generation_canaries",
        "agent_invocation_audits",
    ):
        assert table in source
    for column in ("lease_fence", "idempotency_key", "input_hash", "evidence_digests"):
        assert column in source
    for forbidden in ("raw_prompt", "raw_input", "raw_output", "secret_value", "api_key"):
        assert forbidden not in source
    assert "DROP TABLE IF EXISTS public.agent_invocation_audits" in source
    assert MIGRATION.read_bytes() == DEPLOYMENT_MIGRATION.read_bytes()


def test_wp09_source_and_deployment_runtime_contracts_are_byte_identical():
    for relative_path in (
        "app/core/llm/runtime_governance.py",
        "app/core/llm/runtime_governance_repository.py",
        "app/core/mcp/governed_invocation.py",
    ):
        source = ROOT / relative_path
        deployment = ROOT / "deployment" / relative_path
        assert source.read_bytes() == deployment.read_bytes()