"""Загрузка OpenAPI-контракта и валидация ответов реализации против схем. Контракт (``docs/openapi.yaml``) — источник истины: тесты ``backend/tests/test_contract_responses.py`` прогоняют каждый эндпоинт реализации и сверяют ответ против схемы спецификации. Любое расхождение — ``ContractViolationError``, то есть красный CI. """ from __future__ import annotations import copy import os from functools import lru_cache from pathlib import Path from typing import Any import yaml from jsonschema import Draft202012Validator, FormatChecker SPEC_RELATIVE_PATH = Path("docs") / "openapi.yaml" SPEC_ENV_VAR = "AWG_OPENAPI_SPEC" JSON_MEDIA_TYPE = "application/json" HTTP_METHODS: tuple[str, ...] = ("get", "put", "post", "delete", "patch", "head", "options") DOC_ROUTE_PATHS: frozenset[str] = frozenset( {"/openapi.json", "/docs", "/redoc", "/docs/oauth2-redirect"} ) SECRET_FIELD_NAMES: frozenset[str] = frozenset({"private_key", "preshared_key", "password"}) class ContractViolationError(AssertionError): """Ответ реализации не соответствует схеме контракта.""" def spec_path() -> Path: """Путь до спецификации (``AWG_OPENAPI_SPEC`` переопределяет поиск от корня репозитория).""" override = os.environ.get(SPEC_ENV_VAR) if override: return Path(override) for parent in Path(__file__).resolve().parents: candidate = parent / SPEC_RELATIVE_PATH if candidate.is_file(): return candidate raise FileNotFoundError(f"спецификация {SPEC_RELATIVE_PATH} не найдена") @lru_cache(maxsize=1) def load_spec() -> dict[str, Any]: """Разобранная спецификация OpenAPI (кэшируется на процесс).""" raw = yaml.safe_load(spec_path().read_text(encoding="utf-8")) if not isinstance(raw, dict): raise ContractViolationError("спецификация должна быть YAML-объектом на верхнем уровне") return raw def iter_operations(spec: dict[str, Any]) -> list[tuple[str, str, dict[str, Any]]]: """Перечень описанных в контракте операций как (путь, метод, операция).""" operations: list[tuple[str, str, dict[str, Any]]] = [] for path, item in spec.get("paths", {}).items(): if not isinstance(item, dict): continue for method in HTTP_METHODS: operation = item.get(method) if isinstance(operation, dict): operations.append((path, method, operation)) return operations def _resolve_ref(spec: dict[str, Any], ref: str) -> Any: node: Any = spec for part in ref.lstrip("#/").split("/"): node = node[part] return node def _strip_write_only(node: Any) -> None: """Удаляет ``writeOnly``-свойства (секреты) из схемы ответа. ``writeOnly`` в JSON Schema 2020-12 означает «только в запросе», поэтому в ответе такие поля запрещены: они вырезаются, включая упоминание в ``required``; при этом ``additionalProperties: false`` делает их присутствие ошибкой валидации. """ if isinstance(node, dict): properties = node.get("properties") if isinstance(properties, dict): dropped = [ name for name, sub in properties.items() if isinstance(sub, dict) and sub.get("writeOnly") is True ] for name in dropped: del properties[name] required = node.get("required") if dropped and isinstance(required, list): node["required"] = [name for name in required if name not in dropped] for value in node.values(): _strip_write_only(value) elif isinstance(node, list): for item in node: _strip_write_only(item) def _document_for(schema: dict[str, Any], spec: dict[str, Any]) -> dict[str, Any]: document = copy.deepcopy(schema) document["components"] = copy.deepcopy(spec.get("components", {})) _strip_write_only(document) return document def validator_for(schema: dict[str, Any], spec: dict[str, Any]) -> Draft202012Validator: """Валидатор фрагмента спецификации: внутренние ссылки ``#/components`` разрешаются.""" return Draft202012Validator(_document_for(schema, spec), format_checker=FormatChecker()) def response_schema( spec: dict[str, Any], path: str, method: str, status: int | str ) -> dict[str, Any] | None: """Схема тела ответа для (path, method, status) или ``None``, если ответ не описан.""" operation = spec.get("paths", {}).get(path, {}).get(method) if not isinstance(operation, dict): return None response = operation.get("responses", {}).get(str(status)) if isinstance(response, dict) and "$ref" in response: response = _resolve_ref(spec, str(response["$ref"])) if not isinstance(response, dict): return None media = response.get("content", {}).get(JSON_MEDIA_TYPE) if not isinstance(media, dict): # Non-JSON ответы (например, /metrics → text/plain): берём единственный медиатип. content = response.get("content", {}) media = next((item for item in content.values() if isinstance(item, dict)), None) if not isinstance(media, dict): return None schema = media.get("schema") return schema if isinstance(schema, dict) else None def validate_response( spec: dict[str, Any], path: str, method: str, status: int, payload: Any, ) -> None: """Проверяет тело ответа против схемы контракта; поднимает ``ContractViolationError``.""" schema = response_schema(spec, path, method, status) if schema is None: raise ContractViolationError( f"{method.upper()} {path} -> {status}: ответ не описан в спецификации" ) errors = sorted( validator_for(schema, spec).iter_errors(payload), key=lambda err: list(err.absolute_path) ) if errors: lines = [f"{method.upper()} {path} -> {status}: несоответствий схеме: {len(errors)}"] for err in errors[:10]: location = "/".join(str(part) for part in err.absolute_path) or "" lines.append(f" - {location}: {err.message}") raise ContractViolationError("\n".join(lines)) def error_codes(spec: dict[str, Any]) -> set[str]: """Множество допустимых кодов ошибок из схемы ``ErrorCode``.""" enum = spec.get("components", {}).get("schemas", {}).get("ErrorCode", {}).get("enum", []) return {str(item) for item in enum} def write_only_fields(node: Any) -> set[str]: """Имена всех свойств, помеченных ``writeOnly`` (в любом месте документа).""" found: set[str] = set() if isinstance(node, dict): for name, sub in node.get("properties", {}).items(): if isinstance(sub, dict) and sub.get("writeOnly") is True: found.add(str(name)) found |= write_only_fields(sub) for key, value in node.items(): if key != "properties": found |= write_only_fields(value) elif isinstance(node, list): for item in node: found |= write_only_fields(item) return found def iter_route_operations(app: Any) -> set[tuple[str, str]]: """Множество ``(путь, МЕТОД)`` для всех HTTP/WebSocket-маршрутов приложения. Обходит вложенные роутеры: FastAPI 0.159 не разворачивает ``include_router`` в ``app.routes``: хранит обёртки, содержащие ``original_router``. """ found: set[tuple[str, str]] = set() def walk(router: Any, seen: set[int]) -> None: if id(router) in seen: return seen.add(id(router)) for route in getattr(router, "routes", []): path = getattr(route, "path", None) if isinstance(path, str): if path in DOC_ROUTE_PATHS: continue methods = getattr(route, "methods", None) if methods is None: found.add((path, "GET")) # WebSocketRoute continue for method in methods: if method.upper() not in {"HEAD", "OPTIONS"}: found.add((path, method.upper())) continue inner = ( getattr(route, "original_router", None) or getattr(route, "router", None) or getattr(route, "app", None) ) if inner is not None: walk(inner, seen) walk(app, set()) return found def assert_routes_match(spec: dict[str, Any], app: Any) -> None: """Двусторонняя сверка маршрутов приложения и контракта. * каждая операция контракта обязана существовать в приложении; * каждый HTTP/WebSocket-маршрут приложения обязан быть описан в контракте. """ declared: set[tuple[str, str]] = { (path, method.upper()) for path, method, _ in iter_operations(spec) } implemented = iter_route_operations(app) missing = sorted(declared - implemented) undocumented = sorted(implemented - declared) problems: list[str] = [] if missing: problems.append("нет в реализации: " + ", ".join(f"{m} {p}" for p, m in missing)) if undocumented: problems.append("нет в контракте: " + ", ".join(f"{m} {p}" for p, m in undocumented)) if problems: raise ContractViolationError( "маршруты приложения и docs/openapi.yaml расходятся:\n " + "\n ".join(problems) )