"""Проверки самой спецификации: валидность 3.1, полнота по ТЗ, writeOnly-секреты, единый Error.""" from __future__ import annotations from typing import Any, cast import pytest from openapi_spec_validator import validate as validate_openapi from pydantic import BaseModel from awg_api.contract import ( SECRET_FIELD_NAMES, error_codes, iter_operations, response_schema, validator_for, write_only_fields, ) from awg_api.schemas import ( AuthLoginRequest, ConfigUpdateRequest, DnsConfigIn, GeoIpUpdateRequest, ImportRequest, ModeUpdateRequest, PeerConfigIn, RoutingConfigIn, RoutingRuleCreate, RoutingRuleUpdate, TunnelConfigIn, ) # Операции, требуемые ТЗ/issue #19. REQUIRED_OPERATIONS: tuple[tuple[str, str], ...] = ( ("/api/v1/status", "get"), ("/api/v1/config", "get"), ("/api/v1/config", "put"), ("/api/v1/config/validate", "post"), ("/api/v1/routing/rules", "get"), ("/api/v1/routing/rules", "post"), ("/api/v1/routing/rules/{rule_id}", "get"), ("/api/v1/routing/rules/{rule_id}", "put"), ("/api/v1/routing/rules/{rule_id}", "delete"), ("/api/v1/routing/mode", "put"), ("/api/v1/routing/geoip/update", "post"), ("/api/v1/routing/import", "post"), ("/api/v1/metrics/stream", "get"), ("/api/v1/health", "get"), ("/api/v1/ready", "get"), ("/metrics", "get"), ("/api/v1/devices", "get"), ("/api/v1/logs", "get"), ("/api/v1/auth/login", "post"), ("/api/v1/auth/logout", "post"), ("/api/v1/auth/session", "get"), ) REQUEST_MODEL_CASES: tuple[tuple[str, type[BaseModel]], ...] = ( ("TunnelConfig", TunnelConfigIn), ("PeerConfig", PeerConfigIn), ("DnsConfig", DnsConfigIn), ("RoutingConfig", RoutingConfigIn), ("ConfigUpdateRequest", ConfigUpdateRequest), ("RoutingRuleCreate", RoutingRuleCreate), ("RoutingRuleUpdate", RoutingRuleUpdate), ("ModeUpdateRequest", ModeUpdateRequest), ("GeoIpUpdateRequest", GeoIpUpdateRequest), ("ImportRequest", ImportRequest), ("AuthLoginRequest", AuthLoginRequest), ) STATUS_OK_RANGE = range(200, 300) # 101 Switching Protocols — WebSocket-апгрейд, не ошибка и не JSON-ответ. STATUS_PROTOCOL_SWITCH = 101 def _spec_schema(spec: dict[str, Any], name: str) -> dict[str, Any]: return cast(dict[str, Any], spec["components"]["schemas"][name]) def test_spec_is_valid_openapi_31(spec: dict[str, Any]) -> None: assert spec["openapi"].startswith("3.1") validate_openapi(spec) @pytest.mark.parametrize(("path", "method"), REQUIRED_OPERATIONS) def test_required_operation_declared(spec: dict[str, Any], path: str, method: str) -> None: assert method in spec["paths"].get(path, {}), f"{method.upper()} {path} отсутствует в контракте" def test_every_non_2xx_response_uses_error_schema(spec: dict[str, Any]) -> None: """Единый формат ошибок: любой не-2xx ответ описан схемой Error.""" offenders: list[str] = [] for path, method, operation in iter_operations(spec): for status in operation.get("responses", {}): if int(status) in STATUS_OK_RANGE or int(status) == STATUS_PROTOCOL_SWITCH: continue schema = response_schema(spec, path, method, status) if not isinstance(schema, dict) or not str(schema.get("$ref", "")).endswith("/Error"): offenders.append(f"{method.upper()} {path} -> {status}") assert offenders == [], f"ответы без схемы Error: {offenders}" def test_error_codes_are_closed_enum(spec: dict[str, Any]) -> None: codes = error_codes(spec) assert {"validation_failed", "unauthorized", "not_found", "internal_error"} <= codes def test_secret_fields_are_write_only(spec: dict[str, Any]) -> None: declared = write_only_fields(spec) assert SECRET_FIELD_NAMES <= declared, f"секреты без writeOnly: {SECRET_FIELD_NAMES - declared}" def test_write_only_secrets_removed_from_response_schema(spec: dict[str, Any]) -> None: """Схема ответа (после вырезания writeOnly) не содержит приватных ключей.""" schema = response_schema(spec, "/api/v1/config", "get", 200) assert schema is not None response_document = validator_for(schema, spec).schema tunnel = response_document["components"]["schemas"]["TunnelConfig"] peer = response_document["components"]["schemas"]["PeerConfig"] assert "private_key" not in tunnel["properties"] assert "private_key" not in tunnel.get("required", []) assert "preshared_key" not in peer["properties"] def test_private_key_in_response_fails_validation(spec: dict[str, Any]) -> None: schema = response_schema(spec, "/api/v1/config", "get", 200) assert schema is not None validator = validator_for(schema, spec) payload = { "tunnel": { "address": "10.8.0.2/32", "listen_port": 51820, "mtu": 1420, "private_key": "must-not-be-here", }, "peers": [], "dns": {"mode": "internal", "servers": []}, "routing": { "mode": "geoip", "kill_switch": True, "mtu_clamping": True, "ipv6": "direct", "geoip": {"source": "cache", "prefix_count": 1, "updated_at": None}, }, "updated_at": "2026-10-22T00:00:00Z", } assert list(validator.iter_errors(payload)), "приватный ключ в ответе обязан ломать валидацию" @pytest.mark.parametrize(("schema_name", "model"), REQUEST_MODEL_CASES) def test_request_models_match_spec(spec: dict[str, Any], schema_name: str, model: type[BaseModel]) -> None: """Модель запроса и схема контракта совпадают по полям и обязательности.""" spec_properties = set(_spec_schema(spec, schema_name).get("properties", {})) spec_required = set(_spec_schema(spec, schema_name).get("required", [])) model_schema = model.model_json_schema() model_properties = set(model_schema.get("properties", {})) model_required = set(model_schema.get("required", [])) assert model_properties == spec_properties, ( f"{schema_name}: поля расходятся (только в контракте: " f"{spec_properties - model_properties}, только в модели: {model_properties - spec_properties})" ) assert model_required == spec_required, ( f"{schema_name}: обязательность расходится (только в контракте: " f"{spec_required - model_required}, только в модели: {model_required - spec_required})" ) def test_secret_fields_write_only_in_request_models() -> None: tunnel_schema = TunnelConfigIn.model_json_schema() peer_schema = PeerConfigIn.model_json_schema() auth_schema = AuthLoginRequest.model_json_schema() assert tunnel_schema["properties"]["private_key"].get("writeOnly") is True assert peer_schema["properties"]["preshared_key"].get("writeOnly") is True assert auth_schema["properties"]["password"].get("writeOnly") is True def test_spec_paths_are_versioned(spec: dict[str, Any]) -> None: """Все прикладные эндпоинты живут под /api/v1 (кроме служебных /metrics и /ready-хелперов).""" allowed_unversioned = {"/metrics"} offenders = [ path for path, _method, _operation in iter_operations(spec) if not path.startswith("/api/v1/") and path not in allowed_unversioned ] assert offenders == [], f"эндпоинты вне /api/v1: {offenders}"