embeddevio
839f773010
- contract.py: загрузка docs/openapi.yaml, вырезание writeOnly-секретов из схем ответов, валидация тела ответа (jsonschema Draft2020-12 + FormatChecker), двусторонняя сверка маршрутов приложения и контракта. - state.py: мок-состояние туннеля/правил/устройств/журнала; секреты хранятся отдельно от словарей-ответов.
243 lines
10 KiB
Python
243 lines
10 KiB
Python
"""Загрузка 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 "<root>"
|
|
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)
|
|
)
|