yovision/tests/test_sense_control_contract.py

"""Repository-local invariants for the frozen Sense control-plane contract."""

from __future__ import annotations

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


ROOT = Path(__file__).resolve().parents[1]
OPENAPI_PATH = ROOT / "docs" / "contracts" / "sense-control-v1.openapi.json"
QUOTA_SQL_PATH = ROOT / "docs" / "contracts" / "site-quota-v1.sql"
CONTRACT_README_PATH = ROOT / "docs" / "contracts" / "README.md"

HTTP_METHODS = {"get", "post", "put", "patch", "delete", "options", "head", "trace"}
EXPECTED_OPERATIONS = {
    ("get", "/api/v1/sites/{site_id}/devices"),
    ("post", "/api/v1/sites/{site_id}/devices"),
    ("get", "/api/v1/sites/{site_id}/devices/{device_id}"),
    ("patch", "/api/v1/sites/{site_id}/devices/{device_id}"),
    ("put", "/api/v1/sites/{site_id}/devices/{device_id}/desired-state"),
    ("post", "/api/v1/sites/{site_id}/devices:batchDesiredState"),
    ("get", "/api/v1/operations/{operation_id}"),
}

EXPECTED_ENUMS = {
    "Modality": ["video", "radar", "contact", "button", "wearable", "other"],
    "Capability": ["video_capture", "audio_capture", "spatial_rule", "telemetry"],
    "DesiredState": ["disabled", "enabled"],
    "ActualState": ["pending", "online", "offline", "failed"],
}

REQUIRED_ERROR_CODES = {
    "invalid_request",
    "unauthenticated",
    "forbidden",
    "not_found",
    "precondition_required",
    "etag_mismatch",
    "idempotency_conflict",
    "duplicate_serial_number",
    "quota_exceeded",
    "quota_projection_unavailable",
    "quota_projection_invalid",
    "area_policy_denied",
    "area_policy_unavailable",
    "adapter_not_ready",
    "endpoint_credentials_forbidden",
    "batch_too_large",
}


def load_openapi() -> dict[str, Any]:
    return json.loads(OPENAPI_PATH.read_text(encoding="utf-8"))


def resolve_local_ref(document: dict[str, Any], value: dict[str, Any]) -> dict[str, Any]:
    """Resolve one local JSON pointer; the contract only needs direct component refs."""
    ref = value.get("$ref")
    if not ref:
        return value
    if not ref.startswith("#/"):
        raise ValueError(f"external ref is not allowed here: {ref}")
    current: Any = document
    for token in ref[2:].split("/"):
        current = current[token.replace("~1", "/").replace("~0", "~")]
    if not isinstance(current, dict):
        raise ValueError(f"ref does not resolve to an object: {ref}")
    return current


def operation_parameters(
    document: dict[str, Any], operation: dict[str, Any]
) -> dict[str, dict[str, Any]]:
    result: dict[str, dict[str, Any]] = {}
    for value in operation.get("parameters", []):
        parameter = resolve_local_ref(document, value)
        result[parameter.get("name", "")] = parameter
    return result


def validate_openapi(document: dict[str, Any]) -> list[str]:
    errors: list[str] = []
    if document.get("openapi") != "3.1.0":
        errors.append("OpenAPI version must be 3.1.0")
    if document.get("info", {}).get("version") != "1.0.0":
        errors.append("contract version must be 1.0.0")

    top_security = document.get("security")
    if not top_security or not any("bearerAuth" in item for item in top_security):
        errors.append("top-level bearerAuth is required")

    operations: dict[tuple[str, str], dict[str, Any]] = {}
    operation_ids: list[str] = []
    for path, path_item in document.get("paths", {}).items():
        if not path.startswith("/api/v1/"):
            errors.append(f"unversioned path: {path}")
        for method, operation in path_item.items():
            if method not in HTTP_METHODS:
                continue
            key = (method, path)
            operations[key] = operation
            operation_id = operation.get("operationId")
            if not operation_id:
                errors.append(f"missing operationId: {method.upper()} {path}")
            else:
                operation_ids.append(operation_id)
            if not operation.get("responses"):
                errors.append(f"missing responses: {method.upper()} {path}")
            effective_security = operation.get("security", top_security)
            if not effective_security or not any(
                "bearerAuth" in item for item in effective_security
            ):
                errors.append(f"bearerAuth required: {method.upper()} {path}")

    missing_operations = EXPECTED_OPERATIONS - set(operations)
    unexpected_operations = set(operations) - EXPECTED_OPERATIONS
    if missing_operations:
        errors.append(f"missing operations: {sorted(missing_operations)}")
    if unexpected_operations:
        errors.append(f"unexpected operations: {sorted(unexpected_operations)}")
    if len(operation_ids) != len(set(operation_ids)):
        errors.append("operationId values must be unique")

    list_operation = operations.get(("get", "/api/v1/sites/{site_id}/devices"), {})
    list_parameters = operation_parameters(document, list_operation)
    if "cursor" not in list_parameters:
        errors.append("device list requires cursor pagination")
    limit = list_parameters.get("limit", {}).get("schema", {})
    if limit.get("default") != 50 or limit.get("maximum") != 100:
        errors.append("device list limit must default to 50 and have maximum 100")

    header_requirements = {
        ("post", "/api/v1/sites/{site_id}/devices"): "Idempotency-Key",
        ("post", "/api/v1/sites/{site_id}/devices:batchDesiredState"): "Idempotency-Key",
        ("patch", "/api/v1/sites/{site_id}/devices/{device_id}"): "If-Match",
        ("put", "/api/v1/sites/{site_id}/devices/{device_id}/desired-state"): "If-Match",
    }
    for key, header in header_requirements.items():
        parameters = operation_parameters(document, operations.get(key, {}))
        parameter = parameters.get(header, {})
        if parameter.get("in") != "header" or parameter.get("required") is not True:
            errors.append(f"{key[0].upper()} {key[1]} requires header {header}")

    for key in (
        ("patch", "/api/v1/sites/{site_id}/devices/{device_id}"),
        ("put", "/api/v1/sites/{site_id}/devices/{device_id}/desired-state"),
    ):
        responses = operations.get(key, {}).get("responses", {})
        if "412" not in responses or "428" not in responses:
            errors.append(f"{key[0].upper()} {key[1]} must define 412 and 428")

    schemas = document.get("components", {}).get("schemas", {})
    batch_items = (
        schemas.get("BatchDesiredStateRequest", {})
        .get("properties", {})
        .get("items", {})
    )
    if batch_items.get("maxItems") != 128:
        errors.append("batch desired-state request must have maximum 128 items")

    client_schemas = ("DeviceCreate", "DevicePatch", "DesiredStateChange", "BatchDesiredStateRequest")
    for schema_name in client_schemas:
        if "tenant_id" in schemas.get(schema_name, {}).get("properties", {}):
            errors.append(f"tenant_id must not be client-controlled in {schema_name}")

    create_properties = schemas.get("DeviceCreate", {}).get("properties", {})
    if "id" in create_properties:
        errors.append("device id must be generated by the service")
    required_create = set(schemas.get("DeviceCreate", {}).get("required", []))
    if not {"modality", "capabilities"}.issubset(required_create):
        errors.append("device create requires modality and capabilities")

    for schema_name in ("DeviceCreate", "DevicePatch"):
        properties = schemas.get(schema_name, {}).get("properties", {})
        for field in ("endpoint_ref", "credential_ref", "profile_token"):
            if properties.get(field, {}).get("writeOnly") is not True:
                errors.append(f"{schema_name}.{field} must be writeOnly")

    output_properties = schemas.get("Device", {}).get("properties", {})
    tenant = output_properties.get("tenant_id", {})
    if tenant.get("readOnly") is not True:
        errors.append("Device.tenant_id must be readOnly")
    forbidden_output_fields = {
        "endpoint_ref",
        "credential_ref",
        "profile_token",
        "active_profile",
        "stream_uri",
        "password",
        "token",
    }
    leaked_fields = forbidden_output_fields & set(output_properties)
    if leaked_fields:
        errors.append(f"Device leaks sensitive fields: {sorted(leaked_fields)}")

    for schema_name, expected in EXPECTED_ENUMS.items():
        actual = schemas.get(schema_name, {}).get("enum")
        if actual != expected:
            errors.append(f"{schema_name} enum drift: {actual!r}")

    error_codes = set(schemas.get("ErrorCode", {}).get("enum", []))
    missing_error_codes = REQUIRED_ERROR_CODES - error_codes
    if missing_error_codes:
        errors.append(f"missing stable error codes: {sorted(missing_error_codes)}")
    return errors


def validate_quota_sql(sql: str) -> list[str]:
    errors: list[str] = []
    normalized = re.sub(r"\s+", " ", sql.lower()).strip()
    expected_signature = (
        "create view bell.site_quota_v1 ( tenant_id, site_id, "
        "max_video_channels, source_version, source_updated_at ) as"
    )
    if expected_signature not in normalized:
        errors.append("bell.site_quota_v1 column signature is not frozen as expected")
    if "alter view bell.site_quota_v1 owner to bell_app" not in normalized:
        errors.append("Bell role must own the quota view")
    if "grant usage on schema bell to sense_app" not in normalized:
        errors.append("Sense role needs schema USAGE")
    if "grant select on table bell.site_quota_v1 to sense_app" not in normalized:
        errors.append("Sense role needs SELECT on only the quota view")
    if "revoke all privileges on table bell.site_quota_v1 from public" not in normalized:
        errors.append("PUBLIC privileges must be revoked")
    if re.search(
        r"grant\s+(?:all(?:\s+privileges)?|insert|update|delete|truncate|references|trigger)\b",
        normalized,
    ):
        errors.append("quota view grants a forbidden write or broad privilege")
    if re.search(r"grant\s+select\s+on\s+table\s+bell\.sites\b", normalized):
        errors.append("Sense must not receive SELECT on the Bell source table")
    return errors


class SenseControlContractTests(unittest.TestCase):
    def setUp(self) -> None:
        self.document = load_openapi()
        self.quota_sql = QUOTA_SQL_PATH.read_text(encoding="utf-8")

    def test_frozen_openapi_invariants(self) -> None:
        self.assertEqual([], validate_openapi(self.document))

    def test_frozen_quota_projection_invariants(self) -> None:
        self.assertEqual([], validate_quota_sql(self.quota_sql))

    def test_contract_narrative_records_failure_and_compatibility_semantics(self) -> None:
        text = CONTRACT_README_PATH.read_text(encoding="utf-8")
        for marker in (
            "默认 16",
            "1～128",
            "至少保存 24 小时",
            "404 not_found",
            "已有流保持运行",
            "必须发布新版本",
        ):
            self.assertIn(marker, text)

    def test_validator_rejects_missing_authentication(self) -> None:
        changed = copy.deepcopy(self.document)
        changed.pop("security", None)
        self.assertTrue(any("bearerAuth" in item for item in validate_openapi(changed)))

    def test_validator_rejects_client_controlled_tenant(self) -> None:
        changed = copy.deepcopy(self.document)
        changed["components"]["schemas"]["DeviceCreate"]["properties"]["tenant_id"] = {
            "type": "string"
        }
        self.assertTrue(any("tenant_id" in item for item in validate_openapi(changed)))

    def test_validator_rejects_unbounded_list_or_batch(self) -> None:
        changed = copy.deepcopy(self.document)
        changed["components"]["parameters"]["Limit"]["schema"]["maximum"] = 101
        changed["components"]["schemas"]["BatchDesiredStateRequest"]["properties"][
            "items"
        ]["maxItems"] = 129
        errors = validate_openapi(changed)
        self.assertTrue(any("maximum 100" in item for item in errors))
        self.assertTrue(any("maximum 128" in item for item in errors))

    def test_validator_rejects_sensitive_output_or_writable_credential_ref(self) -> None:
        changed = copy.deepcopy(self.document)
        changed["components"]["schemas"]["DevicePatch"]["properties"]["credential_ref"][
            "writeOnly"
        ] = False
        changed["components"]["schemas"]["Device"]["properties"]["stream_uri"] = {
            "type": "string"
        }
        errors = validate_openapi(changed)
        self.assertTrue(any("credential_ref" in item for item in errors))
        self.assertTrue(any("sensitive fields" in item for item in errors))

    def test_validator_rejects_write_privilege(self) -> None:
        changed = self.quota_sql + "\nGRANT ALL PRIVILEGES ON bell.site_quota_v1 TO sense_app;\n"
        self.assertTrue(any("forbidden" in item for item in validate_quota_sql(changed)))


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