7f891ddb48
- M-8: CompanyGroupCreate и LegalEntityCreate отвергают неизвестные поля (422, объект не создаётся); тесты. - T11.3: safe_filename отбрасывает и категорию Cf (U+202E, U+200B) до обрезки длины; тесты. - M-2: README «Ошибки API» — пример с ALIAS_TAKEN, INN_EXISTS без дополнительных полей. - M-1: README — требование к ingress по размеру тела для /directories/import. - M-4: README — зависимость формы пользователя от DIRECTORY_VIEW, /me не перечитывается, ASSIGNMENT_CHANGE (№35). Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
296 lines
14 KiB
Python
296 lines
14 KiB
Python
"""API ГК и филиалов (spec directories-api, audit-trail): код, история, no-op PATCH, права, список, архив."""
|
||
|
||
import datetime
|
||
import re
|
||
import uuid
|
||
from collections.abc import Sequence
|
||
from typing import Any
|
||
|
||
import httpx
|
||
import pytest
|
||
from sqlalchemy import func, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from src.models.audit_ex import AuditEvent
|
||
from src.models.directories import CompanyGroup, CompanyGroupHistory
|
||
from src.models.enums import AuditAction, AuditSection
|
||
from tests.conftest import AuthClientFactory
|
||
from tests.factories import make_company_group, make_legal_entity
|
||
|
||
HTTP_OK = 200
|
||
HTTP_CREATED = 201
|
||
HTTP_FORBIDDEN = 403
|
||
HTTP_CONFLICT = 409
|
||
HTTP_UNPROCESSABLE = 422
|
||
CODE_PATTERN = re.compile(r"^GK-\d{6}$")
|
||
KRASNODAR = "Краснодарский региональный филиал"
|
||
HEAD_OFFICE = "Головной офис"
|
||
GROUPS = "/directories/company-groups"
|
||
|
||
|
||
async def _events(session: AsyncSession, *actions: AuditAction) -> Sequence[AuditEvent]:
|
||
stmt = select(AuditEvent).where(AuditEvent.action.in_(actions)).order_by(AuditEvent.occurred_at, AuditEvent.id)
|
||
return (await session.execute(stmt)).scalars().all()
|
||
|
||
|
||
async def _office_id(client: httpx.AsyncClient, name: str) -> str:
|
||
response = await client.get("/directories/regional-offices")
|
||
assert response.status_code == HTTP_OK, response.text
|
||
return next(item["id"] for item in response.json() if item["name"] == name)
|
||
|
||
|
||
async def _create(client: httpx.AsyncClient, **body: Any) -> dict[str, Any]:
|
||
response = await client.post(GROUPS, json=body)
|
||
assert response.status_code == HTTP_CREATED, response.text
|
||
created: dict[str, Any] = response.json()
|
||
return created
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_regional_offices_list(auth_client: AuthClientFactory) -> None:
|
||
client = await auth_client("expert@armka.local")
|
||
response = await client.get("/directories/regional-offices")
|
||
assert response.status_code == HTTP_OK
|
||
offices = {item["name"]: item for item in response.json()}
|
||
assert offices[HEAD_OFFICE]["is_head_office"] is True
|
||
assert offices[KRASNODAR]["is_head_office"] is False
|
||
assert set(offices[KRASNODAR]) == {"id", "name", "is_head_office", "is_active"}
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_create_generates_codes_and_audit(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
client = await auth_client("ka_go@armka.local")
|
||
first = await _create(client, name="ГК Первая")
|
||
second = await _create(client, name="ГК Вторая", crm_id="CRM-1")
|
||
assert CODE_PATTERN.match(first["code"]) and CODE_PATTERN.match(second["code"])
|
||
assert first["code"] != second["code"]
|
||
assert first["valid_from"] and first["legal_entities_count"] == 0
|
||
|
||
events = await _events(seeded, AuditAction.GROUP_CHANGE)
|
||
assert len(events) == 2 # noqa: PLR2004
|
||
assert {e.after_value["code"] for e in events} == {first["code"], second["code"]}
|
||
event = next(e for e in events if e.after_value["code"] == second["code"])
|
||
assert (event.section, event.object_type, str(event.object_id)) == (
|
||
AuditSection.DIRECTORIES,
|
||
"company_group",
|
||
second["id"],
|
||
)
|
||
assert event.after_value == {
|
||
"code": second["code"],
|
||
"name": "ГК Вторая",
|
||
"crm_id": "CRM-1",
|
||
"base_regional_office_id": None,
|
||
}
|
||
assert event.user_id is not None
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_patch_office_writes_history_and_assignment_audit(
|
||
auth_client: AuthClientFactory, seeded: AsyncSession
|
||
) -> None:
|
||
client = await auth_client("ka_go@armka.local")
|
||
rostov = await _office_id(client, "Ростовский региональный филиал")
|
||
krasnodar = await _office_id(client, KRASNODAR)
|
||
group = await _create(client, name="ГК Перевод", base_regional_office_id=rostov)
|
||
|
||
response = await client.patch(f"{GROUPS}/{group['id']}", json={"base_regional_office_id": krasnodar})
|
||
assert response.status_code == HTTP_OK, response.text
|
||
updated = response.json()
|
||
assert updated["base_regional_office_id"] == krasnodar
|
||
assert updated["valid_from"] != group["valid_from"]
|
||
|
||
history = (await seeded.execute(select(CompanyGroupHistory))).scalars().all()
|
||
assert len(history) == 1
|
||
snapshot = history[0]
|
||
assert str(snapshot.base_regional_office_id) == rostov
|
||
assert snapshot.name == "ГК Перевод"
|
||
assert snapshot.valid_to is not None
|
||
assert snapshot.valid_from == datetime.datetime.fromisoformat(group["valid_from"])
|
||
|
||
(event,) = await _events(seeded, AuditAction.ASSIGNMENT_CHANGE)
|
||
assert event.before_value == {"base_regional_office_id": rostov}
|
||
assert event.after_value == {"base_regional_office_id": krasnodar}
|
||
assert event.changed_field == "base_regional_office_id"
|
||
assert (event.section, event.object_type) == (AuditSection.DIRECTORIES, "company_group")
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_patch_name_audit_is_group_change_with_diff(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
client = await auth_client("ka_go@armka.local")
|
||
group = await _create(client, name="А")
|
||
response = await client.patch(f"{GROUPS}/{group['id']}", json={"name": "Б"})
|
||
assert response.status_code == HTTP_OK
|
||
events = await _events(seeded, AuditAction.GROUP_CHANGE)
|
||
changed = events[-1]
|
||
assert (changed.changed_field, changed.before_value, changed.after_value) == (
|
||
"name",
|
||
{"name": "А"},
|
||
{"name": "Б"},
|
||
)
|
||
assert not await _events(seeded, AuditAction.ASSIGNMENT_CHANGE)
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_duplicate_crm_id_is_conflict(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
client = await auth_client("ka_go@armka.local")
|
||
await _create(client, name="ГК-1", crm_id="CRM-DUP")
|
||
response = await client.post(GROUPS, json={"name": "ГК-2", "crm_id": "CRM-DUP"})
|
||
assert response.status_code == HTTP_CONFLICT
|
||
assert response.json()["detail"]["code"] == "CRM_ID_EXISTS"
|
||
assert len(await _events(seeded, AuditAction.GROUP_CHANGE)) == 1 # аудит только у первой
|
||
|
||
other = await _create(client, name="ГК-3")
|
||
clash = await client.patch(f"{GROUPS}/{other['id']}", json={"crm_id": "CRM-DUP"})
|
||
assert clash.status_code == HTTP_CONFLICT
|
||
assert clash.json()["detail"]["code"] == "CRM_ID_EXISTS"
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_noop_patch_writes_no_history_and_no_audit(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
client = await auth_client("ka_go@armka.local")
|
||
group = await _create(client, name="ГК Без изменений", crm_id="CRM-NOOP")
|
||
audit_before = len(await _events(seeded, AuditAction.GROUP_CHANGE, AuditAction.ASSIGNMENT_CHANGE))
|
||
|
||
response = await client.patch(f"{GROUPS}/{group['id']}", json={"name": "ГК Без изменений", "crm_id": "CRM-NOOP"})
|
||
assert response.status_code == HTTP_OK
|
||
assert response.json()["valid_from"] == group["valid_from"]
|
||
assert (await seeded.execute(select(func.count()).select_from(CompanyGroupHistory))).scalar_one() == 0
|
||
assert len(await _events(seeded, AuditAction.GROUP_CHANGE, AuditAction.ASSIGNMENT_CHANGE)) == audit_before
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_view_only_user_cannot_create(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
client = await auth_client("expert@armka.local")
|
||
response = await client.post(GROUPS, json={"name": "ГК Эксперта"})
|
||
assert response.status_code == HTTP_FORBIDDEN
|
||
assert not await _events(seeded, AuditAction.GROUP_CHANGE)
|
||
assert (await client.get(GROUPS)).status_code == HTTP_OK # чтение разрешено
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_list_search_filters_and_pagination(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
client = await auth_client("ka_go@armka.local")
|
||
await make_company_group(seeded, name="Русский Бисквит")
|
||
await make_company_group(seeded, name="Другая группа")
|
||
archived = await _create(client, name="Архивная бисквитная")
|
||
assert (await client.post(f"{GROUPS}/{archived['id']}/archive")).status_code == HTTP_OK
|
||
|
||
by_name = (await client.get(GROUPS, params={"q": "БИСКВИТ"})).json()
|
||
assert {item["name"] for item in by_name["items"]} == {"Русский Бисквит", "Архивная бисквитная"}
|
||
assert by_name["total"] == 2 # noqa: PLR2004
|
||
assert [item["name"] for item in by_name["items"]] == sorted(item["name"] for item in by_name["items"])
|
||
|
||
only_archived = (await client.get(GROUPS, params={"q": "бисквит", "is_archived": "true"})).json()
|
||
assert [item["name"] for item in only_archived["items"]] == ["Архивная бисквитная"]
|
||
active = (await client.get(GROUPS, params={"q": "бисквит", "is_archived": "false"})).json()
|
||
assert [item["name"] for item in active["items"]] == ["Русский Бисквит"]
|
||
|
||
beyond = await client.get(GROUPS, params={"q": "бисквит", "offset": 10})
|
||
assert beyond.status_code == HTTP_OK
|
||
assert beyond.json()["items"] == [] and beyond.json()["total"] == 2 # noqa: PLR2004
|
||
assert beyond.json()["offset"] == 10 # noqa: PLR2004
|
||
|
||
assert (await client.get(GROUPS, params={"limit": 201})).status_code == HTTP_UNPROCESSABLE
|
||
assert (await client.get(GROUPS, params={"offset": -1})).status_code == HTTP_UNPROCESSABLE
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_list_search_by_code_and_crm_id_escapes_like(
|
||
auth_client: AuthClientFactory, seeded: AsyncSession
|
||
) -> None:
|
||
"""R-2: `q` ищет по `name`/`crm_id`/`code`; `%` и `_` в запросе — обычные символы."""
|
||
client = await auth_client("ka_go@armka.local")
|
||
group = await _create(client, name="ГК Поиск", crm_id="CRM-Zeta-77")
|
||
await _create(client, name="ГК Прочая", crm_id="CRM-Other")
|
||
await make_company_group(seeded, name="ГК 100%", code="GK-XPCT")
|
||
|
||
by_code = (await client.get(GROUPS, params={"q": group["code"].lower()})).json()
|
||
assert [item["id"] for item in by_code["items"]] == [group["id"]]
|
||
by_crm = (await client.get(GROUPS, params={"q": "zeta-77"})).json()
|
||
assert [item["id"] for item in by_crm["items"]] == [group["id"]]
|
||
|
||
percent = (await client.get(GROUPS, params={"q": "%"})).json()
|
||
assert [item["name"] for item in percent["items"]] == ["ГК 100%"] # `%` не «любая строка»
|
||
underscore = (await client.get(GROUPS, params={"q": "ГК_П"})).json()
|
||
assert underscore["items"] == [] and underscore["total"] == 0 # `_` не «любой символ»
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_list_counts_legal_entities(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
client = await auth_client("ka_go@armka.local")
|
||
group = await make_company_group(seeded, name="ГК с ЮЛ")
|
||
empty = await make_company_group(seeded, name="ГК пустая")
|
||
await make_legal_entity(seeded, group=group)
|
||
await make_legal_entity(seeded, group=group)
|
||
await make_legal_entity(seeded, group=None)
|
||
|
||
items = {i["id"]: i for i in (await client.get(GROUPS)).json()["items"]}
|
||
assert items[str(group.id)]["legal_entities_count"] == 2 # noqa: PLR2004
|
||
assert items[str(empty.id)]["legal_entities_count"] == 0
|
||
card = (await client.get(f"{GROUPS}/{group.id}")).json()
|
||
assert card["legal_entities_count"] == 2 # noqa: PLR2004
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_archive_unarchive_toggle_and_audit(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
client = await auth_client("ka_go@armka.local")
|
||
group = await _create(client, name="ГК Архив")
|
||
archived = await client.post(f"{GROUPS}/{group['id']}/archive")
|
||
assert archived.status_code == HTTP_OK and archived.json()["is_archived"] is True
|
||
restored = await client.post(f"{GROUPS}/{group['id']}/unarchive")
|
||
assert restored.status_code == HTTP_OK and restored.json()["is_archived"] is False
|
||
|
||
events = [e for e in await _events(seeded, AuditAction.GROUP_CHANGE) if e.changed_field == "is_archived"]
|
||
assert [(e.before_value, e.after_value) for e in events] == [
|
||
({"is_archived": False}, {"is_archived": True}),
|
||
({"is_archived": True}, {"is_archived": False}),
|
||
]
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_history_is_ordered_by_valid_from_desc(auth_client: AuthClientFactory) -> None:
|
||
"""R-3: снимки по убыванию `valid_from`."""
|
||
client = await auth_client("ka_go@armka.local")
|
||
group = await _create(client, name="Имя 0")
|
||
for index in (1, 2, 3):
|
||
response = await client.patch(f"{GROUPS}/{group['id']}", json={"name": f"Имя {index}"})
|
||
assert response.status_code == HTTP_OK
|
||
|
||
history = (await client.get(f"{GROUPS}/{group['id']}/history")).json()
|
||
assert [row["name"] for row in history] == ["Имя 2", "Имя 1", "Имя 0"]
|
||
valid_from = [row["valid_from"] for row in history]
|
||
assert valid_from == sorted(valid_from, reverse=True)
|
||
assert set(history[0]) == {
|
||
"id",
|
||
"crm_id",
|
||
"name",
|
||
"base_regional_office_id",
|
||
"valid_from",
|
||
"valid_to",
|
||
"changed_by",
|
||
}
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_validation_and_unknown_group(auth_client: AuthClientFactory) -> None:
|
||
client = await auth_client("ka_go@armka.local")
|
||
assert (await client.post(GROUPS, json={"name": ""})).status_code == HTTP_UNPROCESSABLE
|
||
group = await _create(client, name="ГК")
|
||
extra = await client.patch(f"{GROUPS}/{group['id']}", json={"code": "GK-000001"})
|
||
assert extra.status_code == HTTP_UNPROCESSABLE # extra="forbid"
|
||
missing = await client.patch(f"{GROUPS}/{uuid.uuid4()}", json={"name": "X"})
|
||
assert missing.status_code == 404 # noqa: PLR2004
|
||
assert missing.json()["detail"]["message"] == "ГК не найдена"
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_create_rejects_unknown_fields(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
"""M-8: `extra="forbid"` — служебное поле (`code`) в теле создания даёт 422 (массив ошибок), ГК не создаётся."""
|
||
client = await auth_client("ka_go@armka.local")
|
||
before = (await seeded.execute(select(func.count()).select_from(CompanyGroup))).scalar_one()
|
||
response = await client.post(GROUPS, json={"name": "ГК Лишнее поле", "code": "GK-999999"})
|
||
assert response.status_code == HTTP_UNPROCESSABLE, response.text
|
||
assert isinstance(response.json()["detail"], list)
|
||
count = (await seeded.execute(select(func.count()).select_from(CompanyGroup))).scalar_one()
|
||
assert count == before
|