embeddevio
3dac7e730a
- test_openapi_spec: валидность 3.1, полнота операций по ТЗ, writeOnly-секреты, единый формат Error, соответствие pydantic-моделей схемам. - test_contract_responses: все эндпоинты против схем, ошибки 401/404/409/422/503, WS-кадры, отсутствие секретов в ответах. - test_contract_negative: доказательство AC — подмена реализации на расходящуюся роняет контрактный тест с указанием пути до поля.
111 lines
5.1 KiB
Python
111 lines
5.1 KiB
Python
"""Негативные проверки контрактного теста: отклонение реализации обязано его ронять.
|
|
|
|
Эти тесты — доказательство 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)
|