yovision/tests/test_postgres_contract.py

"""Static safety checks for T-009 PostgreSQL migrations and local harness."""

from __future__ import annotations

import re
import unittest
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
MIGRATION_ROOT = ROOT / "deploy" / "postgres"


def migration_text(name: str) -> str:
    return (MIGRATION_ROOT / name).read_text(encoding="utf-8")


def normalized(text: str) -> str:
    return re.sub(r"\s+", " ", text.lower()).strip()


def privilege_findings(text: str) -> list[str]:
    value = normalized(text)
    findings: list[str] = []
    if "grant select on table bell.site_quota_v1 to sense_app" not in value:
        findings.append("Sense SELECT grant on quota view is missing")
    if "grant select on table bell.area_policy_v1 to sense_app" not in value:
        findings.append("Sense SELECT grant on Area view is missing")
    if "revoke all on table bell.sites from sense_app" not in value:
        findings.append("Bell source table revoke is missing")
    if "revoke all on table bell.areas from sense_app" not in value:
        findings.append("Bell Area source table revoke is missing")
    if re.search(
        r"grant\s+(?:all(?:\s+privileges)?|insert|update|delete|truncate|references|trigger)"
        r"(?:\s*,\s*(?:insert|update|delete|truncate|references|trigger))*"
        r"\s+on(?:\s+table)?\s+bell\.",
        value,
    ):
        findings.append("Sense or PUBLIC receives a broad/write Bell privilege")
    if re.search(r"grant\s+select\s+on(?:\s+table)?\s+bell\.sites\s+to\s+sense_app", value):
        findings.append("Sense receives SELECT on Bell source table")
    return findings


class PostgresContractTests(unittest.TestCase):
    def test_migration_sequence_is_complete_and_stable(self) -> None:
        names = [path.name for path in sorted(MIGRATION_ROOT.glob("[0-9][0-9][0-9]_*.sql"))]
        self.assertEqual(
            [
                "001_roles.sql",
                "002_bell.sql",
                "003_sense.sql",
                "004_privileges.sql",
                "005_area_policy.sql",
                "006_device_operation_outbox.sql",
                "007_privileges_area_audit.sql",
                "008_control_api.sql",
                "009_privileges_control_api.sql",
                "010_reconcile_safety.sql",
                "011_privileges_reconcile_safety.sql",
                "012_bell_events.sql",
                "013_privileges_bell_events.sql",
                "014_audit_relay.sql",
                "015_privileges_audit_relay.sql",
                "016_event_ingress.sql",
                "017_privileges_event_ingress.sql",
            ],
            names,
        )

    def test_roles_are_non_login_and_unprivileged(self) -> None:
        text = normalized(migration_text("001_roles.sql"))
        for role in ("bell_app", "sense_app"):
            self.assertRegex(
                text,
                rf"create role {role} nologin nosuperuser nocreatedb nocreaterole noreplication",
            )
        self.assertNotRegex(text, r"\bpassword\b")

    def test_bell_runtime_role_is_separate_from_migration_owner(self) -> None:
        text = normalized(migration_text("012_bell_events.sql"))
        self.assertIn(
            "create role bell_runtime nologin nosuperuser nocreatedb nocreaterole noreplication",
            text,
        )
        self.assertNotIn("grant bell_app to bell_runtime", text)
        self.assertIn("pg_has_role('bell_runtime', 'bell_app', 'member')", text)

    def test_bell_quota_contract_and_version_are_enforced(self) -> None:
        text = normalized(migration_text("002_bell.sql"))
        expected_signature = (
            "create or replace view bell.site_quota_v1 ( tenant_id, site_id, "
            "max_video_channels, source_version, source_updated_at ) as"
        )
        self.assertIn(expected_signature, text)
        self.assertIn("max_video_channels smallint not null default 16", text)
        self.assertIn("check (max_video_channels between 1 and 128)", text)
        self.assertIn("new.version := old.version + 1", text)
        self.assertIn("alter view bell.site_quota_v1 owner to bell_app", text)

    def test_sense_schema_has_no_writable_site_truth(self) -> None:
        text = normalized(migration_text("003_sense.sql"))
        self.assertNotRegex(text, r"create table(?: if not exists)? sense\.sites\b")
        self.assertIn("create table if not exists sense.devices", text)
        self.assertIn("quota_source_version bigint", text)
        self.assertIn("create table if not exists sense.site_quota_projection_state", text)

    def test_cross_schema_privileges_are_minimal(self) -> None:
        privileges = migration_text("004_privileges.sql") + migration_text(
            "007_privileges_area_audit.sql"
        )
        self.assertEqual([], privilege_findings(privileges))

    def test_privilege_validator_rejects_broad_or_source_access(self) -> None:
        original = migration_text("004_privileges.sql") + migration_text(
            "007_privileges_area_audit.sql"
        )
        changed = original + "\nGRANT ALL PRIVILEGES ON bell.sites TO sense_app;\n"
        self.assertTrue(privilege_findings(changed))
        changed = original + "\nGRANT SELECT ON bell.sites TO sense_app;\n"
        self.assertTrue(privilege_findings(changed))

    def test_local_harness_isolated_from_existing_cluster(self) -> None:
        text = (ROOT / "scripts" / "test_postgres.ps1").read_text(encoding="utf-8")
        for marker in (
            "initdb.exe",
            "pg_ctl.exe",
            "127.0.0.1",
            "yovision-t012-pg-",
            "YOVISION_TEST_POSTGRES_DSN",
            "Get-NetTCPConnection",
        ):
            self.assertIn(marker, text)
        self.assertNotIn("D:\\pgsql17\\data", text)

    def test_audit_relay_uses_fencing_and_separate_schema_ownership(self) -> None:
        migration = normalized(migration_text("014_audit_relay.sql"))
        privileges = normalized(migration_text("015_privileges_audit_relay.sql"))
        for marker in (
            "relay_lease_owner",
            "relay_lease_token",
            "relay_lease_until",
            "create table if not exists bell.audit_events",
            "create table if not exists bell.audit_relay_receipts",
            "expires_at >= received_at + interval '10 minutes'",
        ):
            self.assertIn(marker, migration)
        self.assertIn("grant select, insert on table bell.audit_events to bell_runtime", privileges)
        self.assertIn("revoke all on table bell.audit_events, bell.audit_relay_receipts from sense_app", privileges)

    def test_event_ingress_has_permanent_source_receipts_and_minimal_privileges(self) -> None:
        migration = normalized(migration_text("016_event_ingress.sql"))
        privileges = normalized(migration_text("017_privileges_event_ingress.sql"))
        for marker in (
            "create table if not exists bell.event_ingress_bindings",
            "create table if not exists bell.event_ingress_receipts",
            "primary key (producer_id, source_event_id)",
            "create table if not exists bell.event_ingress_nonces",
            "expires_at >= received_at + interval '10 minutes'",
            "insert into bell.schema_migrations(version) values (5)",
        ):
            self.assertIn(marker, migration)
        self.assertIn(
            "grant select, insert on table bell.event_ingress_receipts to bell_runtime",
            privileges,
        )
        self.assertIn(
            "grant select, insert, delete on table bell.event_ingress_nonces to bell_runtime",
            privileges,
        )
        self.assertNotIn("endpoint_ref", privileges.split("revoke all on table sense.devices")[1])


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