vignette/apps/api/app/test_runtime_schema_ssot.py

"""앱의 런타임 준비 계약과 infra SQL SSOT 사이의 회귀 검사."""

from __future__ import annotations

import re
import unittest
from pathlib import Path

from .runtime_schema import (
    CALIBRATION_TRANSFER_SCHEMA_CONTRACT,
    CONTINUOUS_IMPROVEMENT_SCHEMA_CONTRACT,
    DELIBERATE_PRACTICE_SCHEMA_CONTRACT,
    MEASUREMENT_SCHEMA_CONTRACT,
    MULTIMODAL_ALLIANCE_SCHEMA_CONTRACT,
    NOTIFICATION_SCHEMA_CONTRACT,
    OUTCOME_TRAJECTORY_SCHEMA_CONTRACT,
    REVIEW_SCHEMA_CONTRACT,
    RUPTURE_REPAIR_SCHEMA_CONTRACT,
    SUPERVISION_RESEARCH_SCHEMA_CONTRACT,
)


REPO_ROOT = Path(__file__).resolve().parents[3]
INFRA_SQL = "\n".join(
    path.read_text(encoding="utf-8")
    for path in sorted((REPO_ROOT / "infra" / "db" / "init").glob("*.sql"))
)


class RuntimeSchemaSsotTest(unittest.TestCase):
    def test_runtime_contract_objects_are_owned_by_infra_sql(self) -> None:
        for contract in (
            REVIEW_SCHEMA_CONTRACT,
            NOTIFICATION_SCHEMA_CONTRACT,
            MEASUREMENT_SCHEMA_CONTRACT,
            OUTCOME_TRAJECTORY_SCHEMA_CONTRACT,
            RUPTURE_REPAIR_SCHEMA_CONTRACT,
            DELIBERATE_PRACTICE_SCHEMA_CONTRACT,
            CALIBRATION_TRANSFER_SCHEMA_CONTRACT,
            SUPERVISION_RESEARCH_SCHEMA_CONTRACT,
            CONTINUOUS_IMPROVEMENT_SCHEMA_CONTRACT,
            MULTIMODAL_ALLIANCE_SCHEMA_CONTRACT,
        ):
            for relation in contract.relations:
                schema, table = relation.split(".")
                pattern = rf"CREATE TABLE IF NOT EXISTS\s+{re.escape(schema)}\.{re.escape(table)}\b"
                self.assertRegex(
                    INFRA_SQL, pattern, msg=f"infra SQL missing relation: {relation}"
                )

            for qualified_name in contract.columns:
                schema, table, column = qualified_name.split(".")
                table_mentions = [
                    block
                    for block in re.split(
                        r"(?=CREATE TABLE IF NOT EXISTS|ALTER TABLE)", INFRA_SQL
                    )
                    if re.search(rf"\b{re.escape(schema)}\.{re.escape(table)}\b", block)
                ]
                self.assertTrue(
                    any(
                        re.search(rf"\b{re.escape(column)}\b", block)
                        for block in table_mentions
                    ),
                    msg=f"infra SQL missing column: {qualified_name}",
                )

            for qualified_name in contract.policies:
                schema, table, policy = qualified_name.split(".")
                pattern = (
                    rf"CREATE POLICY\s+{re.escape(policy)}\s+ON\s+"
                    rf"{re.escape(schema)}\.{re.escape(table)}\b"
                )
                self.assertRegex(
                    INFRA_SQL,
                    pattern,
                    msg=f"infra SQL missing policy: {qualified_name}",
                )

            for qualified_name in contract.triggers:
                schema, table, trigger = qualified_name.split(".")
                pattern = (
                    rf"CREATE TRIGGER\s+{re.escape(trigger)}\s+"
                    rf"(?:BEFORE|AFTER|INSTEAD OF)[\s\S]*?ON\s+"
                    rf"{re.escape(schema)}\.{re.escape(table)}\b"
                )
                self.assertRegex(
                    INFRA_SQL,
                    pattern,
                    msg=f"infra SQL missing trigger: {qualified_name}",
                )

            for qualified_name in contract.indexes:
                schema, table, index = qualified_name.split(".")
                pattern = (
                    rf"CREATE (?:UNIQUE )?INDEX IF NOT EXISTS\s+{re.escape(index)}"
                    rf"\s+ON\s+{re.escape(schema)}\.{re.escape(table)}\b"
                )
                self.assertRegex(
                    INFRA_SQL,
                    pattern,
                    msg=f"infra SQL missing index: {qualified_name}",
                )

    def test_alliance_pulse_contract_keeps_lock_and_history_guards(self) -> None:
        required_fragments = (
            "CONSTRAINT measurement_event_pulse_session_fkey",
            "CONSTRAINT self_assessment_pulse_session_fkey",
            "CONSTRAINT ck_alliance_pulse_lock_before_reveal",
            "CONSTRAINT ck_alliance_pulse_time_order",
            "CONSTRAINT ck_alliance_pulse_terminal_state",
            "CREATE TABLE IF NOT EXISTS audit.alliance_pulse_status_event",
            "SECURITY DEFINER",
            "CREATE POLICY p_alliance_pulse_status_event_select",
        )
        for fragment in required_fragments:
            with self.subTest(fragment=fragment):
                self.assertIn(fragment, INFRA_SQL)

        self.assertNotIn(
            "CREATE POLICY p_alliance_pulse_status_event_insert",
            INFRA_SQL,
            msg="status audit rows must only be produced by the pulse trigger",
        )


if __name__ == "__main__":
    unittest.main()