DataOps-platform/tests/test_wp12_metering_showback_migration_contract.py

"""Static security and delivery contract for the WP12 database revision."""

from __future__ import annotations

from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
MIGRATION = ROOT / "migrations/versions/20260818_546_metering_showback.py"
DEPLOYMENT_MIGRATION = ROOT / "deployment/migrations/versions/20260818_546_metering_showback.py"
CONTROL_MIGRATION = ROOT / "migrations/versions/20260818_547_metering_allocation_budget_gateway.py"
DEPLOYMENT_CONTROL_MIGRATION = ROOT / "deployment/migrations/versions/20260818_547_metering_allocation_budget_gateway.py"
REPLAY_MIGRATION = ROOT / "migrations/versions/20260818_548_metering_allocation_replay.py"
DEPLOYMENT_REPLAY_MIGRATION = ROOT / "deployment/migrations/versions/20260818_548_metering_allocation_replay.py"
INTEGRITY_MIGRATION = ROOT / "migrations/versions/20260818_549_metering_showback_integrity.py"
DEPLOYMENT_INTEGRITY_MIGRATION = ROOT / "deployment/migrations/versions/20260818_549_metering_showback_integrity.py"
UNIFIED_REPLAY_MIGRATION = ROOT / "migrations/versions/20260818_550_metering_unified_replay_acl.py"
DEPLOYMENT_UNIFIED_REPLAY_MIGRATION = ROOT / "deployment/migrations/versions/20260818_550_metering_unified_replay_acl.py"
ATOMIC_ALLOCATION_MIGRATION = ROOT / "migrations/versions/20260818_551_metering_allocation_atomicity.py"
DEPLOYMENT_ATOMIC_ALLOCATION_MIGRATION = ROOT / "deployment/migrations/versions/20260818_551_metering_allocation_atomicity.py"
OWNER_FENCE_MIGRATION = ROOT / "migrations/versions/20260818_552_metering_definer_owner_fence.py"
DEPLOYMENT_OWNER_FENCE_MIGRATION = ROOT / "deployment/migrations/versions/20260818_552_metering_definer_owner_fence.py"


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

    assert 'revision = "20260818_546"' in source
    assert 'down_revision = "20260818_545"' in source
    assert "SECURITY DEFINER SET search_path=pg_catalog,public" in source
    assert "metering_showback_runtime_write" in source
    assert "metering_showback_runtime_read" in source
    assert "chargeback_disabled" in source
    assert "REVOKE ALL ON TABLE public.metering_events FROM PUBLIC,dataops_app,dataops_app_runtime" in source


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

    assert "downgrade refused: WP12 metering facts are nonempty" in source
    assert DEPLOYMENT_MIGRATION.read_bytes() == MIGRATION.read_bytes()


def test_wp12_control_gateway_is_closed_and_fails_downgrade_with_persisted_facts():
    source = CONTROL_MIGRATION.read_text(encoding="utf-8")

    assert 'revision = "20260818_547"' in source
    assert 'down_revision = "20260818_546"' in source
    assert "metering_showback_control_write" in source
    assert "SECURITY DEFINER SET search_path=pg_catalog,public" in source
    assert "metering_allocation_weight_invalid" in source
    assert "provider','disabled" in source
    assert "downgrade refused: WP12 allocation or budget facts are nonempty" in source
    assert DEPLOYMENT_CONTROL_MIGRATION.read_bytes() == source.encode("utf-8")


def test_wp12_persisted_allocation_replay_is_runtime_gateway_only_and_mirrored():
    source = REPLAY_MIGRATION.read_text(encoding="utf-8")

    assert 'revision = "20260818_548"' in source and 'down_revision = "20260818_547"' in source
    assert "metering_showback_allocation_replay" in source
    assert "SECURITY DEFINER SET search_path=pg_catalog,public" in source
    assert "metering_allocation_rule_not_found" in source
    assert DEPLOYMENT_REPLAY_MIGRATION.read_bytes() == source.encode("utf-8")


def test_wp12_integrity_followup_closes_rollup_correction_and_duplicate_target_gaps():
    source = INTEGRITY_MIGRATION.read_text(encoding="utf-8")

    assert 'revision = "20260818_549"' in source and 'down_revision = "20260818_548"' in source
    assert "metering_showback_rollup" in source
    assert "metering_correction_scope_invalid" in source
    assert "metering_allocation_target_duplicate" in source
    assert "metering_allocation_window_overlap" in source
    assert DEPLOYMENT_INTEGRITY_MIGRATION.read_bytes() == source.encode("utf-8")


def test_wp12_unified_replay_fences_legacy_acl_and_preflights_existing_facts():
    source = UNIFIED_REPLAY_MIGRATION.read_text(encoding="utf-8")

    assert 'revision = "20260818_550"' in source and 'down_revision = "20260818_549"' in source
    assert "metering_allocation_replay_coverage_invalid" in source
    assert "upgrade refused: WP12 integrity preflight failed" in source
    assert "metering_showback_runtime_read_legacy" in source
    assert DEPLOYMENT_UNIFIED_REPLAY_MIGRATION.read_bytes() == source.encode("utf-8")


def test_wp12_atomic_allocation_followup_fences_concurrency_and_total_weight():
    source = ATOMIC_ALLOCATION_MIGRATION.read_text(encoding="utf-8")

    assert 'revision = "20260818_551"' in source and 'down_revision = "20260818_550"' in source
    assert "metering_allocation_rules_scope_window_excl" in source
    assert "pg_advisory_xact_lock" in source
    assert "metering_showback_allocation_total_guard" in source
    assert "metering_allocation_conflict" in source
    assert DEPLOYMENT_ATOMIC_ALLOCATION_MIGRATION.read_bytes() == source.encode("utf-8")


def test_wp12_definer_owner_followup_uses_nologin_owner_and_revokes_legacy_paths():
    source = OWNER_FENCE_MIGRATION.read_text(encoding="utf-8")

    assert 'revision = "20260818_552"' in source and 'down_revision = "20260818_551"' in source
    assert "rolcanlogin" in source
    assert "ALTER FUNCTION" in source and "OWNER TO dataops_tenant_foundation_owner" in source
    assert "metering_showback_allocation_total_guard" in source
    assert "REVOKE ALL ON FUNCTION" in source
    assert DEPLOYMENT_OWNER_FENCE_MIGRATION.read_bytes() == source.encode("utf-8")