DataOps-platform/tests/security/test_trusted_delivery_database_boundary_contract.py

"""Regression contract for the WP06 database write boundary."""

from __future__ import annotations

import importlib
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]


def test_wp06_database_boundary_uses_preprovisioned_owner_and_definer_gateway():
    migration = importlib.import_module(
        "migrations.versions.20260811_486_trusted_delivery_database_boundary"
    )
    assert migration.down_revision == "20260811_485"
    source = Path(migration.__file__).read_text()
    for required in (
        "dataops_trusted_delivery_owner",
        "dataops_trusted_delivery_writer",
        "SECURITY DEFINER",
        "SET search_path = pg_catalog, public",
        "REVOKE ALL ON FUNCTION",
        "GRANT EXECUTE",
        "REVOKE INSERT, UPDATE, DELETE, TRUNCATE",
        "trusted_delivery_policy_versions",
        "trusted_delivery_control_evidence",
    ):
        assert required in source
    assert "CREATE ROLE" not in source
    assert "ALTER ROLE" not in source


def test_runtime_bootstrap_does_not_grant_schema_wide_table_dml_and_precreates_owner_roles():
    source = (ROOT / "app/core/edge_gateway/runtime_roles.py").read_text()
    assert "dataops_trusted_delivery_owner" in source
    assert "dataops_trusted_delivery_writer" in source
    assert "ON ALL TABLES IN SCHEMA public" not in source
    assert "tablename NOT LIKE 'trusted_delivery_%'" in source


def test_production_runtime_identity_rejects_trusted_delivery_owner_membership():
    source = (ROOT / "app/config/config.py").read_text()
    assert "trusted_delivery_owner_member" in source
    assert "dataops_trusted_delivery_owner" in source
    assert "dataops_trusted_delivery_writer" in source


def test_wp06_all_runtime_writes_are_routed_through_the_v2_closed_gateway():
    migration = importlib.import_module(
        "migrations.versions.20260811_487_trusted_delivery_runtime_write_gateway"
    )
    assert migration.down_revision == "20260811_486"
    source = Path(migration.__file__).read_text()
    for operation in (
        "complete_delivery", "release_hold", "claim_reclaim", "transition_subscription",
        "subscription_attempt", "activate_control_profile", "approve_capability",
        "record_release_gate", "approve_destruction",
    ):
        assert operation in source
    for repository in (
        "app/core/system/trusted_delivery_repository.py",
        "app/core/system/trusted_delivery_subscription_repository.py",
        "app/core/system/trusted_delivery_controls_repository.py",
    ):
        assert "trusted_delivery_runtime_write" in (ROOT / repository).read_text()


def test_wp06_approval_facts_and_incident_impacts_have_a_second_database_gate():
    migration = importlib.import_module(
        "migrations.versions.20260811_488_trusted_delivery_approval_and_incident_gate"
    )
    assert migration.down_revision == "20260811_487"
    source = Path(migration.__file__).read_text()
    for required in (
        "trusted_delivery_approval_is_valid",
        "trusted_delivery_runtime_incident_impact",
        "activate_control_profile",
        "release_hold",
        "data_incident_impacts",
        "REVOKE ALL ON FUNCTION",
    ):
        assert required in source


def test_wp06_runtime_reclaim_is_one_fenced_database_operation():
    """Runtime reclaim must not compose separate delivery/grant writes."""
    migration = importlib.import_module(
        "migrations.versions.20260811_489_trusted_delivery_runtime_reclaim_gateway"
    )
    assert migration.down_revision == "20260811_488"
    source = Path(migration.__file__).read_text()
    for required in (
        "claim_reclaim",
        "fail_reclaim",
        "complete_reclaim",
        "trusted_delivery_runtime_reclaim_write",
        "lease_fence",
        "active legal hold blocks reclaim",
        "trusted_delivery_receipts",
        "SECURITY DEFINER",
        "REVOKE ALL ON FUNCTION",
    ):
        assert required in source


def test_wp06_control_fact_trigger_selects_the_actor_before_validation():
    migration = importlib.import_module(
        "migrations.versions.20260811_491_trusted_delivery_control_fact_actor_fix"
    )
    assert migration.down_revision == "20260811_490"
    source = Path(migration.__file__).read_text()
    assert "IF TG_TABLE_NAME='trusted_delivery_control_evidence' THEN v_actor:=NEW.actor_uid" in source
    assert "v_actor:=NEW.created_by" in source
    assert "CASE WHEN TG_TABLE_NAME='trusted_delivery_control_evidence' THEN NEW.actor_uid" not in source.split("def downgrade", 1)[0]