7007c51f63
normalize_alias (lower, ё→е, без пунктуации и ОПФ), POST/DELETE
/directories/legal-entities/{id}/aliases, конфликты ALIAS_EXISTS /
ALIAS_TAKEN (с legal_entity_id владельца), EMPTY_ALIAS, аудит
LEGAL_ENTITY_CHANGE по legal_entity_alias.
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
234 lines
11 KiB
Python
234 lines
11 KiB
Python
"""API алиасов ЮЛ (spec directories-api «Алиасы ЮЛ», audit-trail, directory-policy)."""
|
||
|
||
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 LegalEntityAlias, RegionalOffice
|
||
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_NO_CONTENT = 204
|
||
HTTP_FORBIDDEN = 403
|
||
HTTP_NOT_FOUND = 404
|
||
HTTP_CONFLICT = 409
|
||
HTTP_UNPROCESSABLE = 422
|
||
ENTITIES = "/directories/legal-entities"
|
||
OBJECT_TYPE = "legal_entity_alias"
|
||
KRASNODAR = "Краснодарский региональный филиал"
|
||
|
||
|
||
async def _events(session: AsyncSession) -> Sequence[AuditEvent]:
|
||
stmt = (
|
||
select(AuditEvent).where(AuditEvent.object_type == OBJECT_TYPE).order_by(AuditEvent.occurred_at, AuditEvent.id)
|
||
)
|
||
return (await session.execute(stmt)).scalars().all()
|
||
|
||
|
||
async def _alias_count(session: AsyncSession) -> int:
|
||
return (await session.execute(select(func.count()).select_from(LegalEntityAlias))).scalar_one()
|
||
|
||
|
||
async def _add(client: httpx.AsyncClient, entity_id: Any, alias: str) -> httpx.Response:
|
||
return await client.post(f"{ENTITIES}/{entity_id}/aliases", json={"alias": alias})
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_alias_is_normalized_and_repeat_is_conflict(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
"""Сценарий спеки «Нормализация»."""
|
||
entity = await make_legal_entity(seeded, name="ООО Пекарня")
|
||
client = await auth_client("ka_go@armka.local")
|
||
|
||
created = await _add(client, entity.id, "ООО «Пекарня Счастливочная»")
|
||
assert created.status_code == HTTP_CREATED, created.text
|
||
body = created.json()
|
||
assert (body["alias"], body["normalized_alias"]) == ("ООО «Пекарня Счастливочная»", "пекарня счастливочная")
|
||
assert body["created_at"]
|
||
|
||
repeat = await _add(client, entity.id, "Пекарня Счастливочная, ООО")
|
||
assert repeat.status_code == HTTP_CONFLICT
|
||
assert repeat.json()["detail"]["code"] == "ALIAS_EXISTS"
|
||
assert await _alias_count(seeded) == 1
|
||
assert len(await _events(seeded)) == 1 # неудачная попытка аудита не пишет
|
||
|
||
card = (await client.get(f"{ENTITIES}/{entity.id}")).json()
|
||
assert [alias["id"] for alias in card["aliases"]] == [body["id"]]
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_create_alias_audit_and_created_by(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
entity = await make_legal_entity(seeded)
|
||
client = await auth_client("ka_go@armka.local")
|
||
body = (await _add(client, entity.id, 'АО "РУССКИЙ БИСКВИТ"')).json()
|
||
|
||
(event,) = await _events(seeded)
|
||
assert event.action == AuditAction.LEGAL_ENTITY_CHANGE
|
||
assert (event.section, str(event.object_id)) == (AuditSection.DIRECTORIES, body["id"])
|
||
assert event.before_value is None
|
||
assert event.after_value == {
|
||
"legal_entity_id": str(entity.id),
|
||
"alias": 'АО "РУССКИЙ БИСКВИТ"',
|
||
"normalized_alias": "русский бисквит",
|
||
}
|
||
row = (await seeded.execute(select(LegalEntityAlias))).scalar_one()
|
||
assert row.created_by == event.user_id
|
||
assert row.created_by is not None
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_alias_of_another_entity_is_taken_with_owner_id_only(
|
||
auth_client: AuthClientFactory, seeded: AsyncSession
|
||
) -> None:
|
||
owner = await make_legal_entity(seeded, name="ООО Владелец")
|
||
other = await make_legal_entity(seeded, name="ООО Другое")
|
||
client = await auth_client("ka_go@armka.local")
|
||
assert (await _add(client, owner.id, "ООО «Пекарня»")).status_code == HTTP_CREATED
|
||
|
||
response = await _add(client, other.id, "пекарня, ао")
|
||
assert response.status_code == HTTP_CONFLICT
|
||
detail = response.json()["detail"]
|
||
assert detail["code"] == "ALIAS_TAKEN"
|
||
assert detail["legal_entity_id"] == str(owner.id)
|
||
assert set(detail) == {"code", "message", "legal_entity_id"} # ни имени, ни ИНН владельца
|
||
assert "Владелец" not in response.text
|
||
assert owner.inn not in response.text
|
||
assert await _alias_count(seeded) == 1
|
||
assert len(await _events(seeded)) == 1
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_taken_alias_of_invisible_owner_discloses_only_id(
|
||
auth_client: AuthClientFactory, seeded: AsyncSession
|
||
) -> None:
|
||
"""Владелец в чужом филиале: пользователь филиала получает только его `legal_entity_id` (решение по спеке)."""
|
||
krasnodar = (await seeded.execute(select(RegionalOffice).where(RegionalOffice.name == KRASNODAR))).scalar_one()
|
||
foreign_group = await make_company_group(seeded, name="ГК Краснодар", office=krasnodar)
|
||
foreign = await make_legal_entity(seeded, name="ООО Чужое", group=foreign_group)
|
||
go_client = await auth_client("ka_go@armka.local")
|
||
assert (await _add(go_client, foreign.id, "Пекарня")).status_code == HTTP_CREATED
|
||
|
||
rostov = (await seeded.execute(select(RegionalOffice).where(RegionalOffice.name.like("Ростов%")))).scalar_one()
|
||
own_group = await make_company_group(seeded, name="ГК Ростов", office=rostov)
|
||
own = await make_legal_entity(seeded, name="ООО Своё", group=own_group)
|
||
rostov_client = await auth_client("ka_rf@armka.local")
|
||
response = await _add(rostov_client, own.id, "ООО Пекарня")
|
||
assert response.status_code == HTTP_CONFLICT
|
||
assert response.json()["detail"] == {
|
||
"code": "ALIAS_TAKEN",
|
||
"message": response.json()["detail"]["message"],
|
||
"legal_entity_id": str(foreign.id),
|
||
}
|
||
assert "Чужое" not in response.text
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_empty_normalization_is_422(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
entity = await make_legal_entity(seeded)
|
||
client = await auth_client("ka_go@armka.local")
|
||
for alias in ("ООО", "«»", " "):
|
||
response = await _add(client, entity.id, alias)
|
||
assert response.status_code == HTTP_UNPROCESSABLE
|
||
assert response.json()["detail"]["code"] == "EMPTY_ALIAS"
|
||
assert await _alias_count(seeded) == 0
|
||
assert not await _events(seeded)
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_alias_body_validation(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
entity = await make_legal_entity(seeded)
|
||
client = await auth_client("ka_go@armka.local")
|
||
assert (await _add(client, entity.id, "")).status_code == HTTP_UNPROCESSABLE
|
||
assert (await _add(client, entity.id, "а" * 501)).status_code == HTTP_UNPROCESSABLE
|
||
assert (await client.post(f"{ENTITIES}/{entity.id}/aliases", json={})).status_code == HTTP_UNPROCESSABLE
|
||
assert await _alias_count(seeded) == 0
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_delete_alias_writes_audit_with_before_value(
|
||
auth_client: AuthClientFactory, seeded: AsyncSession
|
||
) -> None:
|
||
entity = await make_legal_entity(seeded)
|
||
editor = await auth_client("ka_go@armka.local")
|
||
alias_id = (await _add(editor, entity.id, "ООО «Пекарня»")).json()["id"]
|
||
|
||
# DIRECTORY_DELETE есть только у администратора; редактор справочников получает 403 и алиас остаётся
|
||
forbidden = await editor.delete(f"{ENTITIES}/{entity.id}/aliases/{alias_id}")
|
||
assert forbidden.status_code == HTTP_FORBIDDEN
|
||
assert await _alias_count(seeded) == 1
|
||
|
||
admin = await auth_client("admin@armka.local")
|
||
response = await admin.delete(f"{ENTITIES}/{entity.id}/aliases/{alias_id}")
|
||
assert response.status_code == HTTP_NO_CONTENT
|
||
assert response.content == b""
|
||
assert await _alias_count(seeded) == 0
|
||
|
||
created, deleted = await _events(seeded)
|
||
assert deleted.action == AuditAction.LEGAL_ENTITY_CHANGE
|
||
assert (deleted.section, str(deleted.object_id)) == (AuditSection.DIRECTORIES, alias_id)
|
||
assert deleted.before_value == {
|
||
"legal_entity_id": str(entity.id),
|
||
"alias": "ООО «Пекарня»",
|
||
"normalized_alias": "пекарня",
|
||
}
|
||
assert deleted.after_value is None
|
||
assert created.object_id == deleted.object_id # запись о создании остаётся (append-only)
|
||
|
||
# освободившийся алиас можно добавить снова
|
||
assert (await _add(editor, entity.id, "Пекарня")).status_code == HTTP_CREATED
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_delete_unknown_or_foreign_alias_is_404(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
first = await make_legal_entity(seeded)
|
||
second = await make_legal_entity(seeded)
|
||
editor = await auth_client("ka_go@armka.local")
|
||
alias_id = (await _add(editor, first.id, "Первое")).json()["id"]
|
||
|
||
admin = await auth_client("admin@armka.local")
|
||
for entity_id, target in ((second.id, alias_id), (first.id, uuid.uuid4()), (uuid.uuid4(), alias_id)):
|
||
response = await admin.delete(f"{ENTITIES}/{entity_id}/aliases/{target}")
|
||
assert response.status_code == HTTP_NOT_FOUND
|
||
assert await _alias_count(seeded) == 1
|
||
assert len(await _events(seeded)) == 1
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_invisible_entity_aliases_are_404_like_missing(
|
||
auth_client: AuthClientFactory, seeded: AsyncSession
|
||
) -> None:
|
||
krasnodar = (await seeded.execute(select(RegionalOffice).where(RegionalOffice.name == KRASNODAR))).scalar_one()
|
||
foreign_group = await make_company_group(seeded, name="ГК Краснодар", office=krasnodar)
|
||
foreign = await make_legal_entity(seeded, group=foreign_group)
|
||
go_client = await auth_client("ka_go@armka.local")
|
||
alias_id = (await _add(go_client, foreign.id, "Чужой алиас")).json()["id"]
|
||
|
||
client = await auth_client("ka_rf@armka.local")
|
||
missing = await _add(client, uuid.uuid4(), "Что угодно")
|
||
assert missing.status_code == HTTP_NOT_FOUND
|
||
response = await _add(client, foreign.id, "Что угодно")
|
||
assert response.status_code == HTTP_NOT_FOUND
|
||
assert response.json() == missing.json()
|
||
assert await _alias_count(seeded) == 1
|
||
|
||
manager = await auth_client("client_manager@armka.local") # Краснодар видит ЮЛ, но права DELETE нет
|
||
assert (await manager.delete(f"{ENTITIES}/{foreign.id}/aliases/{alias_id}")).status_code == HTTP_FORBIDDEN
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_expert_cannot_add_alias(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
entity = await make_legal_entity(seeded)
|
||
client = await auth_client("expert@armka.local")
|
||
response = await _add(client, entity.id, "Пекарня")
|
||
assert response.status_code == HTTP_FORBIDDEN
|
||
assert await _alias_count(seeded) == 0
|
||
assert not await _events(seeded)
|