yovision/tests/test_area_audit_contract.py

"""Static contract and safety checks for T-010 Area admission and audit Outbox."""

from __future__ import annotations

import json
import re
import unittest
from pathlib import Path
from typing import Any


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


def text(path: Path) -> str:
    return path.read_text(encoding="utf-8")


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


def property_names(value: Any) -> set[str]:
    names: set[str] = set()
    if isinstance(value, dict):
        properties = value.get("properties")
        if isinstance(properties, dict):
            names.update(str(name) for name in properties)
        for child in value.values():
            names.update(property_names(child))
    elif isinstance(value, list):
        for child in value:
            names.update(property_names(child))
    return names


def area_privilege_findings(value: str) -> list[str]:
    source = normalized(value)
    findings: list[str] = []
    if "grant select on table bell.area_policy_v1 to sense_app" not in source:
        findings.append("missing Area-view SELECT")
    if "revoke all on table bell.areas from sense_app" not in source:
        findings.append("missing Area-source revoke")
    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\.",
        source,
    ):
        findings.append("Bell write privilege granted")
    if re.search(r"grant\s+select\s+on(?:\s+table)?\s+bell\.areas\s+to\s+sense_app", source):
        findings.append("Bell Area source is readable")
    return findings


class AreaAuditContractTests(unittest.TestCase):
    def test_area_projection_contract_has_exact_signature_and_policy_enum(self) -> None:
        contract = normalized(text(CONTRACT_ROOT / "area-policy-v1.sql"))
        migration = normalized(text(MIGRATION_ROOT / "005_area_policy.sql"))
        signature = (
            "create or replace view bell.area_policy_v1 ( tenant_id, site_id, area_id, "
            "capture_policy, source_version, source_updated_at ) as"
        )
        for value in (contract, migration):
            self.assertIn(signature, value)
            self.assertIn("alter view bell.area_policy_v1 owner to bell_app", value)
        self.assertIn(
            "capture_policy in ('video_allowed', 'non_imaging_only')", migration
        )
        self.assertIn("new.version := old.version + 1", migration)

    def test_area_projection_is_not_a_second_writable_truth(self) -> None:
        migration = normalized(text(MIGRATION_ROOT / "005_area_policy.sql"))
        self.assertNotRegex(migration, r"create table(?: if not exists)? sense\.areas\b")
        self.assertIn("create table if not exists sense.area_policy_projection_state", migration)
        self.assertIn("area_policy_source_version bigint", migration)
        self.assertIn("area_id text", migration)

    def test_area_privileges_are_select_only_and_negative_cases_fail(self) -> None:
        original = text(MIGRATION_ROOT / "007_privileges_area_audit.sql")
        self.assertEqual([], area_privilege_findings(original))
        self.assertTrue(
            area_privilege_findings(
                original + "\nGRANT UPDATE ON bell.area_policy_v1 TO sense_app;\n"
            )
        )
        self.assertTrue(
            area_privilege_findings(original + "\nGRANT SELECT ON bell.areas TO sense_app;\n")
        )

    def test_local_audit_schema_is_strict_and_contains_no_secret_fields(self) -> None:
        schema = json.loads(text(CONTRACT_ROOT / "sense-device-audit-v1.schema.json"))
        self.assertEqual("https://json-schema.org/draft/2020-12/schema", schema["$schema"])
        self.assertFalse(schema["additionalProperties"])
        self.assertEqual(
            {"device.created", "device.desired_state.accepted"},
            set(schema["properties"]["event_type"]["enum"]),
        )
        self.assertEqual(
            {"user", "service", "system"},
            set(schema["properties"]["actor"]["properties"]["type"]["enum"]),
        )
        forbidden = {
            "endpoint_ref",
            "credential_ref",
            "profile_token",
            "path_name",
            "password",
            "stream_uri",
            "mediamtx_config",
        }
        self.assertEqual(set(), forbidden & property_names(schema))

    def test_outbox_migration_has_delivery_state_and_integrity_constraints(self) -> None:
        migration = normalized(text(MIGRATION_ROOT / "006_device_operation_outbox.sql"))
        for marker in (
            "create table if not exists sense.device_operation_outbox",
            "payload jsonb not null",
            "attempt_count integer not null default 0",
            "next_attempt_at timestamptz",
            "delivered_at timestamptz",
            "jsonb_typeof(payload) = 'object'",
            "insert into sense.schema_migrations(version) values (3)",
        ):
            self.assertIn(marker, migration)

    def test_repository_writes_audit_before_commit_without_sensitive_payload(self) -> None:
        source = text(ROOT / "Sense" / "internal" / "store" / "postgres.go")
        create_start = source.index("func (s *Postgres) CreateDevice")
        create_end = source.index("func checkPostgresAreaPolicy", create_start)
        create_body = source[create_start:create_end]
        state_start = source.index("func (s *Postgres) SetDesiredState")
        state_end = source.index("func (s *Postgres) GetDevice", state_start)
        state_body = source[state_start:state_end]
        self.assertLess(create_body.index("insertPostgresAudit"), create_body.index("tx.Commit"))
        self.assertIn("insertPostgresAudit", state_body)
        self.assertLess(state_body.rindex("insertPostgresAudit"), state_body.rindex("tx.Commit"))
        payload_fragments = "\n".join(
            re.findall(r"Payload:\s*map\[string\]any\{(.*?)\n\s*\}", source, re.S)
        )
        for forbidden in ("EndpointRef", "CredentialRef", "PathName"):
            self.assertNotIn(forbidden, payload_fragments)

    def test_public_api_retains_stable_area_failure_codes(self) -> None:
        spec = json.loads(text(CONTRACT_ROOT / "sense-control-v1.openapi.json"))
        error_codes = set(spec["components"]["schemas"]["ErrorCode"]["enum"])
        self.assertIn("area_policy_unavailable", error_codes)
        self.assertIn("area_policy_denied", error_codes)


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