322 lines
14 KiB
Python
322 lines
14 KiB
Python
"""Контрактный тест «схема ↔ реализация»: каждый эндпоинт сверяется со схемой из 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
|