test(api): контрактные тесты «схема ↔ реализация» (Refs #19)
- test_openapi_spec: валидность 3.1, полнота операций по ТЗ, writeOnly-секреты, единый формат Error, соответствие pydantic-моделей схемам. - test_contract_responses: все эндпоинты против схем, ошибки 401/404/409/422/503, WS-кадры, отсутствие секретов в ответах. - test_contract_negative: доказательство AC — подмена реализации на расходящуюся роняет контрактный тест с указанием пути до поля.
This commit is contained in:
@@ -0,0 +1,48 @@
|
||||
"""Общие фикстуры контрактных тестов."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Iterator
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from awg_api.app import create_app
|
||||
from awg_api.contract import load_spec
|
||||
from awg_api.state import MockState
|
||||
|
||||
TEST_TOKEN = "test-token-contract"
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def spec() -> dict[str, Any]:
|
||||
"""Спецификация контракта (docs/openapi.yaml)."""
|
||||
return load_spec()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def state() -> MockState:
|
||||
return MockState()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def app(state: MockState) -> FastAPI:
|
||||
return create_app(state=state, token=TEST_TOKEN)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(app: FastAPI) -> Iterator[TestClient]:
|
||||
with TestClient(app) as test_client:
|
||||
yield test_client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_headers() -> dict[str, str]:
|
||||
return {"Authorization": f"Bearer {TEST_TOKEN}"}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def token() -> str:
|
||||
return TEST_TOKEN
|
||||
@@ -0,0 +1,110 @@
|
||||
"""Негативные проверки контрактного теста: отклонение реализации обязано его ронять.
|
||||
|
||||
Эти тесты — доказательство AC «GIVEN реализованный эндпоинт с отклонением от схемы,
|
||||
WHEN запущен контрактный тест, THEN CI падает с указанием расхождения»: они подменяют
|
||||
реализацию на заведомо расходящуюся и проверяют, что тот же самый валидатор её ловит.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI, Request
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from awg_api.app import create_app
|
||||
from awg_api.contract import ContractViolationError, assert_routes_match, validate_response
|
||||
from awg_api.state import PRIVATE_KEY_SENTINEL, MockState
|
||||
|
||||
|
||||
class DeviatingStatusState(MockState):
|
||||
"""Реализация, «забывшая» контракт: неверный тип и лишнее поле."""
|
||||
|
||||
def get_status(self) -> dict[str, Any]:
|
||||
payload = super().get_status()
|
||||
payload["tunnel"]["rx_bytes"] = "много" # намеренно неверный тип вместо integer
|
||||
payload["extra_field"] = True # additionalProperties: false
|
||||
return payload
|
||||
|
||||
|
||||
class LeakingConfigState(MockState):
|
||||
"""Реализация, возвращающая приватный ключ в ответе."""
|
||||
|
||||
def get_config(self) -> dict[str, Any]:
|
||||
payload = super().get_config()
|
||||
payload["tunnel"]["private_key"] = PRIVATE_KEY_SENTINEL
|
||||
return payload
|
||||
|
||||
|
||||
class DroppingFieldState(MockState):
|
||||
"""Реализация, потерявшая обязательное поле."""
|
||||
|
||||
def get_config(self) -> dict[str, Any]:
|
||||
payload = super().get_config()
|
||||
del payload["routing"]["kill_switch"]
|
||||
return payload
|
||||
|
||||
|
||||
def _client_for(state: MockState, token: str) -> TestClient:
|
||||
return TestClient(create_app(state=state, token=token))
|
||||
|
||||
|
||||
def test_valid_implementation_passes(spec: dict[str, Any], client: TestClient, auth_headers: dict[str, str]) -> None:
|
||||
"""Контроль ложных срабатываний: корректная реализация проходит без замечаний."""
|
||||
response = client.get("/api/v1/status", headers=auth_headers)
|
||||
validate_response(spec, "/api/v1/status", "get", 200, response.json())
|
||||
|
||||
|
||||
def test_wrong_type_and_extra_field_are_caught(spec: dict[str, Any], token: str) -> None:
|
||||
with _client_for(DeviatingStatusState(), token) as client:
|
||||
response = client.get("/api/v1/status", headers={"Authorization": f"Bearer {token}"})
|
||||
assert response.status_code == 200
|
||||
with pytest.raises(ContractViolationError) as excinfo:
|
||||
validate_response(spec, "/api/v1/status", "get", 200, response.json())
|
||||
message = str(excinfo.value)
|
||||
assert "tunnel/rx_bytes" in message
|
||||
assert "extra_field" in message
|
||||
|
||||
|
||||
def test_private_key_in_response_is_caught(spec: dict[str, Any], token: str) -> None:
|
||||
with _client_for(LeakingConfigState(), token) as client:
|
||||
response = client.get("/api/v1/config", headers={"Authorization": f"Bearer {token}"})
|
||||
assert PRIVATE_KEY_SENTINEL in response.text, "подготовка теста: ключ действительно утёк"
|
||||
with pytest.raises(ContractViolationError) as excinfo:
|
||||
validate_response(spec, "/api/v1/config", "get", 200, response.json())
|
||||
assert "private_key" in str(excinfo.value)
|
||||
|
||||
|
||||
def test_missing_required_field_is_caught(spec: dict[str, Any], token: str) -> None:
|
||||
with _client_for(DroppingFieldState(), token) as client:
|
||||
response = client.get("/api/v1/config", headers={"Authorization": f"Bearer {token}"})
|
||||
with pytest.raises(ContractViolationError) as excinfo:
|
||||
validate_response(spec, "/api/v1/config", "get", 200, response.json())
|
||||
assert "kill_switch" in str(excinfo.value)
|
||||
|
||||
|
||||
def test_undocumented_route_is_caught(spec: dict[str, Any], token: str) -> None:
|
||||
app = create_app(token=token)
|
||||
|
||||
@app.get("/api/v1/undocumented")
|
||||
async def undocumented(request: Request) -> dict[str, bool]:
|
||||
return {"ok": True}
|
||||
|
||||
with pytest.raises(ContractViolationError) as excinfo:
|
||||
assert_routes_match(spec, app)
|
||||
assert "нет в контракте" in str(excinfo.value)
|
||||
assert "/api/v1/undocumented" in str(excinfo.value)
|
||||
|
||||
|
||||
def test_missing_route_is_caught(spec: dict[str, Any]) -> None:
|
||||
with pytest.raises(ContractViolationError) as excinfo:
|
||||
assert_routes_match(spec, FastAPI())
|
||||
assert "нет в реализации" in str(excinfo.value)
|
||||
|
||||
|
||||
def test_undeclared_status_code_is_caught(spec: dict[str, Any], client: TestClient, auth_headers: dict[str, str]) -> None:
|
||||
response = client.get("/api/v1/status", headers=auth_headers)
|
||||
with pytest.raises(ContractViolationError) as excinfo:
|
||||
validate_response(spec, "/api/v1/status", "get", 202, response.json())
|
||||
assert "не описан в спецификации" in str(excinfo.value)
|
||||
@@ -0,0 +1,321 @@
|
||||
"""Контрактный тест «схема ↔ реализация»: каждый эндпоинт сверяется со схемой из docs/openapi.yaml.
|
||||
|
||||
Любое расхождение (лишнее поле, неверный тип, отсутствующая обязательная ветка,
|
||||
утечка writeOnly-секрета) — падение с указанием пути до поля.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from starlette.websockets import WebSocketDisconnect
|
||||
|
||||
from awg_api.app import create_app
|
||||
from awg_api.contract import (
|
||||
ContractViolationError,
|
||||
assert_routes_match,
|
||||
response_schema,
|
||||
validator_for,
|
||||
validate_response,
|
||||
)
|
||||
from awg_api.state import PRIVATE_KEY_SENTINEL, PRESHARED_KEY_SENTINEL, MockState
|
||||
|
||||
RULE_ID_PATH = "/api/v1/routing/rules/{rule_id}"
|
||||
LEAKED_KEY = "NEW-PRIVATE-KEY-MUST-NOT-LEAK"
|
||||
SECRET_SENTINELS = (PRIVATE_KEY_SENTINEL, PRESHARED_KEY_SENTINEL, LEAKED_KEY)
|
||||
SECRET_FIELD_NAMES = ("private_key", "preshared_key", "password")
|
||||
|
||||
TUNNEL_PAYLOAD: dict[str, Any] = {
|
||||
"address": "10.8.0.9/32",
|
||||
"listen_port": 51821,
|
||||
"mtu": 1420,
|
||||
"private_key": LEAKED_KEY,
|
||||
"dns": ["10.8.0.1"],
|
||||
}
|
||||
ROUTING_PAYLOAD: dict[str, Any] = {
|
||||
"mode": "all-vpn",
|
||||
"kill_switch": True,
|
||||
"mtu_clamping": True,
|
||||
"ipv6": "block",
|
||||
"geoip": {"source": "cache", "prefix_count": 11412, "updated_at": None},
|
||||
}
|
||||
RULE_PAYLOAD: dict[str, Any] = {
|
||||
"name": "Правило контрактного теста",
|
||||
"kind": "domain",
|
||||
"value": "contract.example.net",
|
||||
"action": "vpn",
|
||||
}
|
||||
IMPORT_PAYLOAD: dict[str, Any] = {
|
||||
"format": "amneziawg-conf",
|
||||
"content": "[Peer]\nPublicKey = abc\nPresharedKey = def\n",
|
||||
}
|
||||
|
||||
# (метод, путь, тело) — успешный путь для каждой операции контракта.
|
||||
HAPPY_PATHS: tuple[tuple[str, str, dict[str, Any] | None], ...] = (
|
||||
("get", "/api/v1/health", None),
|
||||
("get", "/api/v1/ready", None),
|
||||
("get", "/metrics", None),
|
||||
("get", "/api/v1/status", None),
|
||||
("get", "/api/v1/config", None),
|
||||
("put", "/api/v1/config", {"tunnel": TUNNEL_PAYLOAD}),
|
||||
("post", "/api/v1/config/validate", {"tunnel": TUNNEL_PAYLOAD}),
|
||||
("get", "/api/v1/routing/rules", None),
|
||||
("post", "/api/v1/routing/rules", RULE_PAYLOAD),
|
||||
("put", "/api/v1/routing/mode", {"mode": "dns"}),
|
||||
("post", "/api/v1/routing/geoip/update", {"force": True}),
|
||||
("post", "/api/v1/routing/import", IMPORT_PAYLOAD),
|
||||
("get", "/api/v1/devices", None),
|
||||
("get", "/api/v1/logs", None),
|
||||
("get", "/api/v1/auth/session", None),
|
||||
("post", "/api/v1/auth/logout", None),
|
||||
("post", "/api/v1/auth/login", {"username": "admin", "password": "awg-dash-dev-password"}),
|
||||
)
|
||||
|
||||
EXPECTED_STATUS: dict[tuple[str, str], int] = {("post", "/api/v1/routing/rules"): 201}
|
||||
|
||||
|
||||
def _check(spec: dict[str, Any], method: str, path: str, response: Any) -> None:
|
||||
"""Валидирует ответ против схемы контракта (JSON — как объект, иначе — как текст)."""
|
||||
content_type = response.headers.get("content-type", "")
|
||||
payload = response.json() if content_type.startswith("application/json") else response.text
|
||||
validate_response(spec, path, method, response.status_code, payload)
|
||||
|
||||
|
||||
def test_every_spec_operation_is_implemented(spec: dict[str, Any], app: Any) -> None:
|
||||
assert_routes_match(spec, app)
|
||||
|
||||
|
||||
def test_openapi_json_serves_the_contract(client: TestClient, spec: dict[str, Any]) -> None:
|
||||
response = client.get("/openapi.json")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == spec
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("method", "path", "body"), HAPPY_PATHS)
|
||||
def test_success_response_matches_spec(
|
||||
spec: dict[str, Any],
|
||||
client: TestClient,
|
||||
auth_headers: dict[str, str],
|
||||
method: str,
|
||||
path: str,
|
||||
body: dict[str, Any] | None,
|
||||
) -> None:
|
||||
response = client.request(method.upper(), path, json=body, headers=auth_headers)
|
||||
expected = EXPECTED_STATUS.get((method, path), 200)
|
||||
assert response.status_code == expected, f"{method.upper()} {path}: {response.status_code} {response.text}"
|
||||
_check(spec, method, path, response)
|
||||
|
||||
|
||||
def test_rule_crud_cycle_matches_spec(spec: dict[str, Any], client: TestClient, auth_headers: dict[str, str]) -> None:
|
||||
created = client.post("/api/v1/routing/rules", json=RULE_PAYLOAD, headers=auth_headers)
|
||||
assert created.status_code == 201, created.text
|
||||
_check(spec, "post", "/api/v1/routing/rules", created)
|
||||
rule_id = created.json()["id"]
|
||||
|
||||
fetched = client.get(f"/api/v1/routing/rules/{rule_id}", headers=auth_headers)
|
||||
assert fetched.status_code == 200
|
||||
_check(spec, "get", RULE_ID_PATH, fetched)
|
||||
|
||||
updated = client.put(f"/api/v1/routing/rules/{rule_id}", json={"enabled": False}, headers=auth_headers)
|
||||
assert updated.status_code == 200
|
||||
assert updated.json()["enabled"] is False
|
||||
_check(spec, "put", RULE_ID_PATH, updated)
|
||||
|
||||
deleted = client.delete(f"/api/v1/routing/rules/{rule_id}", headers=auth_headers)
|
||||
assert deleted.status_code == 204
|
||||
assert deleted.content == b""
|
||||
|
||||
|
||||
def test_rules_list_filters_match_spec(spec: dict[str, Any], client: TestClient, auth_headers: dict[str, str]) -> None:
|
||||
response = client.get("/api/v1/routing/rules", params={"enabled": "true", "kind": "cidr"}, headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
payload = response.json()
|
||||
assert all(rule["kind"] == "cidr" and rule["enabled"] for rule in payload["items"])
|
||||
_check(spec, "get", "/api/v1/routing/rules", response)
|
||||
|
||||
|
||||
def test_logs_pagination_matches_spec(spec: dict[str, Any], client: TestClient, auth_headers: dict[str, str]) -> None:
|
||||
first = client.get("/api/v1/logs", params={"limit": 2}, headers=auth_headers)
|
||||
assert first.status_code == 200
|
||||
payload = first.json()
|
||||
assert len(payload["items"]) == 2
|
||||
assert payload["next_cursor"] is not None
|
||||
_check(spec, "get", "/api/v1/logs", first)
|
||||
|
||||
second = client.get(
|
||||
"/api/v1/logs",
|
||||
params={"limit": 2, "cursor": payload["next_cursor"], "level": "info"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert second.status_code == 200
|
||||
assert all(entry["level"] == "info" for entry in second.json()["items"])
|
||||
_check(spec, "get", "/api/v1/logs", second)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("method", "path", "body", "expected_status", "expected_code"),
|
||||
(
|
||||
("get", "/api/v1/status", None, 401, "unauthorized"),
|
||||
("get", "/api/v1/config", None, 401, "unauthorized"),
|
||||
("get", "/api/v1/routing/rules", None, 401, "unauthorized"),
|
||||
("get", "/api/v1/devices", None, 401, "unauthorized"),
|
||||
("get", "/api/v1/logs", None, 401, "unauthorized"),
|
||||
("post", "/api/v1/routing/rules", {"name": "x", "kind": "domain", "value": "x", "action": "vpn"}, 401, "unauthorized"),
|
||||
(
|
||||
"put",
|
||||
"/api/v1/config",
|
||||
{"tunnel": {**TUNNEL_PAYLOAD, "unknown_field": 1}},
|
||||
422,
|
||||
"validation_failed",
|
||||
),
|
||||
("post", "/api/v1/config/validate", {"tunnel": {"address": "10.8.0.2/32", "listen_port": 99999}}, 422, "validation_failed"),
|
||||
("post", "/api/v1/routing/rules", {"name": "", "kind": "nope", "value": "x"}, 422, "validation_failed"),
|
||||
("put", "/api/v1/routing/mode", {"mode": "turbo"}, 422, "validation_failed"),
|
||||
("post", "/api/v1/auth/login", {"username": "admin", "password": "wrong"}, 401, "unauthorized"),
|
||||
("post", "/api/v1/auth/login", {"username": "", "password": ""}, 422, "validation_failed"),
|
||||
("get", "/api/v1/logs", None, 422, "validation_failed"),
|
||||
),
|
||||
)
|
||||
def test_error_response_matches_error_schema(
|
||||
spec: dict[str, Any],
|
||||
client: TestClient,
|
||||
auth_headers: dict[str, str],
|
||||
method: str,
|
||||
path: str,
|
||||
body: dict[str, Any] | None,
|
||||
expected_status: int,
|
||||
expected_code: str,
|
||||
) -> None:
|
||||
headers = dict(auth_headers)
|
||||
params: dict[str, Any] = {}
|
||||
if expected_status == 401 and path != "/api/v1/auth/login":
|
||||
headers = {}
|
||||
if path == "/api/v1/logs" and expected_status == 422:
|
||||
params = {"cursor": "not-a-number"}
|
||||
response = client.request(method.upper(), path, json=body, headers=headers, params=params)
|
||||
assert response.status_code == expected_status, response.text
|
||||
payload = response.json()
|
||||
assert payload["code"] == expected_code
|
||||
assert set(payload) == {"code", "message", "details"}
|
||||
_check(spec, method, path, response)
|
||||
|
||||
|
||||
def test_not_found_rule_matches_error_schema(
|
||||
spec: dict[str, Any], client: TestClient, auth_headers: dict[str, str]
|
||||
) -> None:
|
||||
missing = "00000000-0000-4000-8000-000000000000"
|
||||
url = f"/api/v1/routing/rules/{missing}"
|
||||
responses = [
|
||||
("get", client.get(url, headers=auth_headers)),
|
||||
("put", client.put(url, json={"enabled": True}, headers=auth_headers)),
|
||||
("delete", client.delete(url, headers=auth_headers)),
|
||||
]
|
||||
for method, response in responses:
|
||||
assert response.status_code == 404, response.text
|
||||
assert response.json()["code"] == "not_found"
|
||||
_check(spec, method, RULE_ID_PATH, response)
|
||||
|
||||
|
||||
def test_conflict_on_duplicate_rule_matches_error_schema(
|
||||
spec: dict[str, Any],
|
||||
client: TestClient,
|
||||
auth_headers: dict[str, str],
|
||||
) -> None:
|
||||
first = client.post("/api/v1/routing/rules", json=RULE_PAYLOAD, headers=auth_headers)
|
||||
assert first.status_code == 201
|
||||
duplicate = client.post("/api/v1/routing/rules", json=RULE_PAYLOAD, headers=auth_headers)
|
||||
assert duplicate.status_code == 409
|
||||
assert duplicate.json()["code"] == "conflict"
|
||||
_check(spec, "post", "/api/v1/routing/rules", duplicate)
|
||||
|
||||
|
||||
def test_ready_503_matches_error_schema(spec: dict[str, Any], token: str) -> None:
|
||||
app = create_app(state=MockState(ready=False), token=token)
|
||||
with TestClient(app) as client:
|
||||
response = client.get("/api/v1/ready")
|
||||
assert response.status_code == 503
|
||||
assert response.json()["code"] == "agent_unavailable"
|
||||
_check(spec, "get", "/api/v1/ready", response)
|
||||
health = client.get("/api/v1/health")
|
||||
assert health.status_code == 200
|
||||
assert health.json()["status"] == "degraded"
|
||||
_check(spec, "get", "/api/v1/health", health)
|
||||
|
||||
|
||||
def test_validate_endpoint_reports_issues_without_applying(
|
||||
spec: dict[str, Any],
|
||||
client: TestClient,
|
||||
auth_headers: dict[str, str],
|
||||
) -> None:
|
||||
response = client.post(
|
||||
"/api/v1/config/validate",
|
||||
json={"tunnel": {"address": "10.8.0.2", "listen_port": 51820, "mtu": 1500, "private_key": "k"}},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
payload = response.json()
|
||||
assert payload["valid"] is False
|
||||
assert {issue["code"] for issue in payload["errors"]} == {"invalid_cidr"}
|
||||
assert {issue["severity"] for issue in payload["warnings"]} == {"warning"}
|
||||
_check(spec, "post", "/api/v1/config/validate", response)
|
||||
|
||||
|
||||
def test_metrics_endpoint_returns_prometheus_text(
|
||||
spec: dict[str, Any],
|
||||
client: TestClient,
|
||||
) -> None:
|
||||
response = client.get("/metrics")
|
||||
assert response.status_code == 200
|
||||
assert response.headers["content-type"].startswith("text/plain")
|
||||
assert "awg_dash_tunnel_up 1" in response.text
|
||||
_check(spec, "get", "/metrics", response)
|
||||
|
||||
|
||||
def test_responses_contain_no_secret_material(
|
||||
spec: dict[str, Any],
|
||||
client: TestClient,
|
||||
auth_headers: dict[str, str],
|
||||
) -> None:
|
||||
"""AC: в ответах нет приватных ключей — ни значений, ни имён полей."""
|
||||
responses = [
|
||||
client.get("/api/v1/config", headers=auth_headers),
|
||||
client.put("/api/v1/config", json={"tunnel": TUNNEL_PAYLOAD}, headers=auth_headers),
|
||||
client.get("/api/v1/status", headers=auth_headers),
|
||||
client.get("/api/v1/devices", headers=auth_headers),
|
||||
client.get("/api/v1/logs", headers=auth_headers),
|
||||
client.get("/api/v1/auth/session", headers=auth_headers),
|
||||
client.post("/api/v1/auth/logout", headers=auth_headers),
|
||||
]
|
||||
for response in responses:
|
||||
body = response.text
|
||||
for sentinel in SECRET_SENTINELS:
|
||||
assert sentinel not in body, f"утечка секрета {sentinel} в {response.request.url}"
|
||||
for field in SECRET_FIELD_NAMES:
|
||||
assert field not in body, f"поле {field} присутствует в ответе {response.request.url}"
|
||||
|
||||
|
||||
def test_websocket_frames_match_spec(spec: dict[str, Any], client: TestClient, token: str) -> None:
|
||||
frame_schema = spec["components"]["schemas"]["WsFrame"]
|
||||
metrics_schema = spec["components"]["schemas"]["MetricsSample"]
|
||||
frame_validator = validator_for(frame_schema, spec)
|
||||
data_validator = validator_for(metrics_schema, spec)
|
||||
with client.websocket_connect(f"/api/v1/metrics/stream?token={token}&interval_ms=250") as websocket:
|
||||
frame = websocket.receive_json()
|
||||
assert list(frame_validator.iter_errors(frame)) == []
|
||||
assert frame["type"] == "metrics"
|
||||
assert list(data_validator.iter_errors(frame["data"])) == []
|
||||
|
||||
|
||||
def test_websocket_requires_token(client: TestClient) -> None:
|
||||
with pytest.raises(WebSocketDisconnect), client.websocket_connect("/api/v1/metrics/stream") as websocket:
|
||||
websocket.receive_json()
|
||||
|
||||
|
||||
def test_response_schema_lookup_fails_for_unknown_response(spec: dict[str, Any], client: TestClient) -> None:
|
||||
"""Неописанный в контракте ответ — ошибка контракта, не молчаливый пропуск."""
|
||||
response = client.get("/api/v1/health")
|
||||
with pytest.raises(ContractViolationError):
|
||||
validate_response(spec, "/api/v1/health", "get", 418, response.json())
|
||||
assert response_schema(spec, "/api/v1/health", "get", 418) is None
|
||||
@@ -0,0 +1,186 @@
|
||||
"""Проверки самой спецификации: валидность 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)."""
|
||||
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}"
|
||||
Reference in New Issue
Block a user