arm-core/api/tests/test_openapi_routes.py
Andrey Epifancev e246531c70 feat(consolidation): кандидаты и ручное сопоставление контрагента с алиасом и автоприменением
GET …/counterparty-candidates (scope group|all, QUERY_TOO_SHORT, recommended по
нормализованному имени строки) и POST …/osv/entries/{entry_id}/counterparty:
проверка строки свода версии, ЮЛ без policy, ALIAS_TAKEN с owners / перенос алиаса,
MANUAL строке и PROPAGATED строкам с тем же strip()-именем без противоречия ИНН,
один OSV_COUNTERPARTY_MAP и LEGAL_ENTITY_CHANGE на каждый алиас. Строки общие для
версий блокируются FOR NO KEY UPDATE ORDER BY id; advisory-блокировка алиаса общая
со справочником. Решения #62–#70 в design.md; 4.3 отмечена.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
2026-10-06 06:46:06 +04:00

190 lines
9.6 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""OpenAPI содержит все маршруты Срезов 1–4 (specs directories-api, directory-import, users-admin,
consolidations-api) без дублей моделей.
Схема строится из объявленных маршрутов без БД и сети. Состав маршрутов — по спекам change `slice-1-directories`;
проверка нужна, чтобы маршрут не пропал при переносе роутеров и чтобы `RegionalOfficeOut` (R-13) был одной моделью.
"""
import importlib
from typing import Any
import pytest
app_module = importlib.import_module("src.app")
D = "/api/v1/directories"
A = "/api/v1/admin"
# путь → методы (нижний регистр, как в OpenAPI)
EXPECTED_ROUTES: dict[str, set[str]] = {
"/api/v1/me": {"get"},
# directories-api: филиалы, ГК, ГСК/ГСЗ, ЮЛ/ИП
f"{D}/regional-offices": {"get"},
f"{D}/company-groups": {"get", "post"},
f"{D}/company-groups/{{group_id}}": {"get", "patch"},
f"{D}/company-groups/{{group_id}}/archive": {"post"},
f"{D}/company-groups/{{group_id}}/unarchive": {"post"},
f"{D}/company-groups/{{group_id}}/history": {"get"},
f"{D}/company-groups/{{company_group_id}}/related-groups": {"get", "post"},
f"{D}/related-groups/{{group_id}}": {"get", "patch", "delete"},
f"{D}/related-groups/{{group_id}}/members": {"post"},
f"{D}/related-groups/{{group_id}}/members/{{legal_entity_id}}": {"delete"},
f"{D}/legal-entities": {"get", "post"},
f"{D}/legal-entities/{{entity_id}}": {"get", "patch"},
f"{D}/legal-entities/{{entity_id}}/move": {"post"},
f"{D}/legal-entities/{{entity_id}}/withdraw": {"post"},
f"{D}/legal-entities/{{entity_id}}/history": {"get"},
f"{D}/legal-entities/{{entity_id}}/aliases": {"post"},
f"{D}/legal-entities/{{entity_id}}/aliases/{{alias_id}}": {"delete"},
# directory-import
f"{D}/import": {"post"},
# users-admin
f"{A}/users": {"get", "post"},
f"{A}/users/{{user_id}}": {"patch"},
f"{A}/users/{{user_id}}/roles": {"put"},
f"{A}/roles": {"get"},
f"{A}/permissions": {"get"},
f"{A}/roles/{{code}}/permissions": {"put"},
# primary-files, documents-api (Срез 2c)
"/api/v1/legal-entities/{legal_entity_id}/files": {"get", "post"},
"/api/v1/files/{file_id}": {"get", "delete"},
"/api/v1/documents/{document_id}": {"get"},
"/api/v1/documents/{document_id}/entries": {"get"},
"/api/v1/documents/{document_id}/confirm": {"post"},
# consolidations-api (Срез 4)
"/api/v1/consolidations": {"post"},
"/api/v1/company-groups/{group_id}/consolidations": {"get"},
"/api/v1/consolidations/versions/{version_id}": {"get"},
"/api/v1/consolidations/versions/{version_id}/lock": {"post", "delete"},
"/api/v1/consolidations/versions/{version_id}/status": {"post"},
"/api/v1/consolidations/versions/{version_id}/members": {"put"},
"/api/v1/consolidations/versions/{version_id}/checklist": {"get"},
"/api/v1/consolidations/versions/{version_id}/osv": {"get"},
"/api/v1/consolidations/versions/{version_id}/osv/summary": {"get"},
"/api/v1/consolidations/versions/{version_id}/counterparty-candidates": {"get"},
"/api/v1/consolidations/versions/{version_id}/osv/entries/{entry_id}/counterparty": {"post"},
}
# Префиксы, под которыми маршрутов сверх EXPECTED_ROUTES быть не должно
CHECKED_PREFIXES = (
D,
A,
"/api/v1/legal-entities",
"/api/v1/files",
"/api/v1/documents",
"/api/v1/consolidations",
"/api/v1/company-groups",
)
# Модели, которые обязаны быть в схеме ровно один раз под коротким именем
SINGLE_MODELS = [
"RegionalOfficeOut",
"CompanyGroupOut",
"LegalEntityOut",
"UserOut",
"RoleOut",
"ImportReportOut",
"FileOut",
"DocumentOut",
"EntryOut",
"VersionCard",
"ConsolidationPeriodOut",
"LockOut",
"StatusIn",
]
@pytest.fixture(scope="module")
def schema() -> dict[str, Any]:
spec: dict[str, Any] = app_module.app.openapi()
return spec
@pytest.mark.parametrize("path", sorted(EXPECTED_ROUTES))
def test_spec_route_is_in_openapi(schema: dict[str, Any], path: str) -> None:
assert path in schema["paths"], f"в OpenAPI нет маршрута {path}"
assert set(schema["paths"][path]) >= EXPECTED_ROUTES[path], (path, sorted(schema["paths"][path]))
def test_no_unexpected_routes_under_directories_and_admin(schema: dict[str, Any]) -> None:
"""Маршрутов сверх спек нет: новые маршруты сначала попадают в спеку и в этот список."""
actual = {path for path in schema["paths"] if path.startswith(CHECKED_PREFIXES)}
expected = {path for path in EXPECTED_ROUTES if path.startswith(CHECKED_PREFIXES)}
assert actual == expected
for path in actual:
assert set(schema["paths"][path]) == EXPECTED_ROUTES[path], path
def test_components_have_no_duplicated_models(schema: dict[str, Any]) -> None:
"""Одна модель — одно имя: нет автоимён вида `src__schemas__x__RegionalOfficeOut` и суффиксов `-Input`/`-Output`.
Имена — ключи словаря, поэтому дубликатов быть не может; реальный страж — проверка на `__`/`-` (FastAPI даёт такие
имена, когда две разные модели совпали по имени или одна модель разошлась на вход/выход).
"""
names = list(schema["components"]["schemas"])
for model in SINGLE_MODELS:
assert model in names, model
assert not [name for name in names if "__" in name or "-" in name], names
def test_consolidation_routes_declare_documented_responses(schema: dict[str, Any]) -> None:
"""Создание — 201 с карточкой версии и ошибки 404/409/422; чтение карточки и списка — 404 (ГК вне видимости)."""
create = schema["paths"]["/api/v1/consolidations"]["post"]["responses"]
assert {"201", "404", "409", "422"} <= set(create)
assert create["201"]["content"]["application/json"]["schema"]["$ref"].endswith("/VersionCard")
card = schema["paths"]["/api/v1/consolidations/versions/{version_id}"]["get"]["responses"]
assert {"200", "404"} <= set(card)
listing = schema["paths"]["/api/v1/company-groups/{group_id}/consolidations"]["get"]["responses"]
assert {"200", "404"} <= set(listing)
checklist = schema["paths"]["/api/v1/consolidations/versions/{version_id}/checklist"]["get"]["responses"]
assert {"200", "404"} <= set(checklist)
assert checklist["200"]["content"]["application/json"]["schema"]["$ref"].endswith("/ChecklistOut")
osv = schema["paths"]["/api/v1/consolidations/versions/{version_id}/osv"]["get"]
assert {"200", "404", "422"} <= set(osv["responses"])
assert osv["responses"]["200"]["content"]["application/json"]["schema"]["$ref"].endswith(
"/Page_ConsolidatedRowOut_"
)
params = {parameter["name"]: parameter for parameter in osv["parameters"]}
assert {
"legal_entity_id",
"account_root",
"counterparty_type",
"unmapped",
"match_kind",
"q",
"limit",
"offset",
} <= set(params)
assert params["match_kind"]["schema"]["anyOf"][0]["type"] == "array" # повторяемый параметр
summary = schema["paths"]["/api/v1/consolidations/versions/{version_id}/osv/summary"]["get"]["responses"]
assert {"200", "404"} <= set(summary)
assert summary["200"]["content"]["application/json"]["schema"]["$ref"].endswith("/OsvSummaryOut")
def test_lock_and_status_routes_declare_documented_responses(schema: dict[str, Any]) -> None:
"""Взятие — 201/200 с `LockOut`, 404/409/422; снятие — 204/404/409; смена статуса — карточка, 404/409/422."""
lock = schema["paths"]["/api/v1/consolidations/versions/{version_id}/lock"]
assert {"200", "201", "404", "409", "422"} <= set(lock["post"]["responses"])
assert lock["post"]["responses"]["201"]["content"]["application/json"]["schema"]["$ref"].endswith("/LockOut")
assert {"204", "404", "409"} <= set(lock["delete"]["responses"])
change = schema["paths"]["/api/v1/consolidations/versions/{version_id}/status"]["post"]["responses"]
assert {"200", "404", "409", "422"} <= set(change)
assert change["200"]["content"]["application/json"]["schema"]["$ref"].endswith("/VersionCard")
def test_counterparty_mapping_routes_declare_documented_responses(schema: dict[str, Any]) -> None:
"""Кандидаты — список `CandidateOut`, 404/422; сопоставление — `MapCounterpartyOut` (строка может быть `null`),
404/409/422."""
candidates = schema["paths"]["/api/v1/consolidations/versions/{version_id}/counterparty-candidates"]["get"]
assert {"200", "404", "422"} <= set(candidates["responses"])
assert candidates["responses"]["200"]["content"]["application/json"]["schema"]["items"]["$ref"].endswith(
"/CandidateOut"
)
assert {"q", "scope", "entry_id"} <= {parameter["name"] for parameter in candidates["parameters"]}
path = "/api/v1/consolidations/versions/{version_id}/osv/entries/{entry_id}/counterparty"
mapping = schema["paths"][path]["post"]["responses"]
assert {"200", "404", "409", "422"} <= set(mapping)
assert mapping["200"]["content"]["application/json"]["schema"]["$ref"].endswith("/MapCounterpartyOut")
row = schema["components"]["schemas"]["MapCounterpartyOut"]["properties"]["row"]
assert {"type": "null"} in row["anyOf"]