content-factory/apps/backend/tests/contracts/test_public_openapi_contract.py

from __future__ import annotations

import sys
import types
import unittest
from importlib.util import find_spec
from pathlib import Path
from typing import Any


BACKEND_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(BACKEND_ROOT))


def _module_available(name: str) -> bool:
    try:
        return find_spec(name) is not None
    except (ModuleNotFoundError, ValueError):
        return False


def _install_optional_dependency_stubs() -> None:
    # Health routes import external clients; this contract test only reads OpenAPI.
    if not _module_available("boto3"):
        boto3 = types.ModuleType("boto3")
        boto3.client = lambda *args, **kwargs: None
        sys.modules["boto3"] = boto3

    if not _module_available("psycopg"):
        psycopg = types.ModuleType("psycopg")
        psycopg.connect = lambda *args, **kwargs: None
        sys.modules["psycopg"] = psycopg

    if not _module_available("redis"):
        redis = types.ModuleType("redis")

        class Redis:
            @staticmethod
            def from_url(*args: Any, **kwargs: Any) -> None:
                return None

        redis.Redis = Redis
        sys.modules["redis"] = redis

    if not _module_available("botocore.config"):
        botocore = types.ModuleType("botocore")
        config = types.ModuleType("botocore.config")

        class Config:
            def __init__(self, *args: Any, **kwargs: Any) -> None:
                pass

        config.Config = Config
        sys.modules["botocore"] = botocore
        sys.modules["botocore.config"] = config


_install_optional_dependency_stubs()

from src.presentation.main import app  # noqa: E402


class BackendPublicOpenApiContractTest(unittest.TestCase):
    def setUp(self) -> None:
        self.openapi = app.openapi()
        self.schemas = self.openapi.get("components", {}).get("schemas", {})

    def test_article_create_request_contract_is_public_and_rejects_invalid_payloads(
        self,
    ) -> None:
        failures: list[str] = []

        article_schema = self.schemas.get("ArticleCreateRequest")
        if article_schema is None:
            failures.append("components.schemas.ArticleCreateRequest is missing")
        else:
            required = set(article_schema.get("required", []))
            for field in ("brief_description", "target_site_id"):
                if field not in required:
                    failures.append(f"ArticleCreateRequest must require {field!r}")

            invalid_article_errors = self._validation_errors(
                schema_name="ArticleCreateRequest",
                payload={},
            )
            if not self._has_required_error(invalid_article_errors, "brief_description"):
                failures.append(
                    "ArticleCreateRequest contract must reject payloads missing "
                    "'brief_description'"
                )
            if not self._has_required_error(invalid_article_errors, "target_site_id"):
                failures.append(
                    "ArticleCreateRequest contract must reject payloads missing "
                    "'target_site_id'"
                )

        request_ref = self._json_request_schema_ref(path="/api/articles", method="post")
        if request_ref != "#/components/schemas/ArticleCreateRequest":
            failures.append(
                "POST /api/articles must expose ArticleCreateRequest as its JSON "
                f"request body; got {request_ref or 'no request body'}"
            )

        role_schema = self.schemas.get("Role")
        if role_schema is None:
            failures.append("components.schemas.Role is missing")
        else:
            role_values = self._enum_values(role_schema)
            if role_values != ["ADMIN", "EDITOR"]:
                failures.append(
                    "Role enum must be exactly ['ADMIN', 'EDITOR']; "
                    f"got {role_values!r}"
                )
            if not self._has_enum_error("Role", "WRITER"):
                failures.append("Role contract must reject invalid value 'WRITER'")

        workflow_status_schema = self.schemas.get("ArticleWorkflowStatus")
        if workflow_status_schema is None:
            failures.append("components.schemas.ArticleWorkflowStatus is missing")
        elif not self._has_enum_error("ArticleWorkflowStatus", "NOT_A_STATUS"):
            failures.append(
                "ArticleWorkflowStatus contract must reject invalid workflow statuses"
            )

        self.assertEqual([], failures)

    def _json_request_schema_ref(self, *, path: str, method: str) -> str | None:
        operation = self.openapi.get("paths", {}).get(path, {}).get(method, {})
        content = operation.get("requestBody", {}).get("content", {})
        schema = content.get("application/json", {}).get("schema", {})
        return schema.get("$ref")

    def _has_required_error(self, errors: list[str], field: str) -> bool:
        return any(error == f"missing required field: {field}" for error in errors)

    def _has_enum_error(self, schema_name: str, value: str) -> bool:
        return f"invalid enum value: {value}" in self._validation_errors(
            schema_name=schema_name,
            payload=value,
        )

    def _validation_errors(self, *, schema_name: str, payload: Any) -> list[str]:
        schema = self.schemas.get(schema_name)
        if schema is None:
            return [f"missing schema: {schema_name}"]

        enum_values = self._enum_values(schema)
        if enum_values is not None:
            return [] if payload in enum_values else [f"invalid enum value: {payload}"]

        if schema.get("type") != "object" or not isinstance(payload, dict):
            return []

        required = schema.get("required", [])
        return [
            f"missing required field: {field}"
            for field in required
            if field not in payload
        ]

    def _enum_values(self, schema: dict[str, Any]) -> list[str] | None:
        enum_values = schema.get("enum")
        if isinstance(enum_values, list):
            return enum_values

        for composite_key in ("allOf", "anyOf", "oneOf"):
            for item in schema.get(composite_key, []):
                resolved = self._resolve_schema(item)
                values = self._enum_values(resolved)
                if values is not None:
                    return values

        return None

    def _resolve_schema(self, schema: dict[str, Any]) -> dict[str, Any]:
        ref = schema.get("$ref")
        if not isinstance(ref, str):
            return schema

        prefix = "#/components/schemas/"
        if not ref.startswith(prefix):
            return schema

        return self.schemas.get(ref.removeprefix(prefix), schema)


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