"""API ГСК/ГСЗ и их состава (spec directories-api «ГСК и ГСЗ — группы и состав», directory-policy, audit-trail). Сценарии спеки: ГСК+ГСЗ одновременно, вторая ГСК, повторное включение, удаление непустой группы; Review Focus 4 — закрытый ЮЛ (`close_date`, без ГК) → 422 `DIFFERENT_COMPANY_GROUP`. """ 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 RegionalOffice, RelatedGroup, RelatedGroupMember from src.models.enums import AuditAction, AuditSection, RelatedGroupKind from tests.conftest import AuthClientFactory from tests.factories import make_company_group, make_legal_entity, make_membership, make_related_group HTTP_OK = 200 HTTP_CREATED = 201 HTTP_NO_CONTENT = 204 HTTP_FORBIDDEN = 403 HTTP_NOT_FOUND = 404 HTTP_CONFLICT = 409 HTTP_UNPROCESSABLE = 422 GROUPS = "/directories/company-groups" RELATED = "/directories/related-groups" ENTITIES = "/directories/legal-entities" KRASNODAR = "Краснодарский региональный филиал" GO = "ka_go@armka.local" ADMIN = "admin@armka.local" KA_RF = "ka_rf@armka.local" EXPERT = "expert@armka.local" async def _events(session: AsyncSession, object_type: str | None = None) -> Sequence[AuditEvent]: stmt = select(AuditEvent).where(AuditEvent.action == AuditAction.MEMBERSHIP_CHANGE) if object_type is not None: stmt = stmt.where(AuditEvent.object_type == object_type) return (await session.execute(stmt.order_by(AuditEvent.occurred_at, AuditEvent.id))).scalars().all() async def _member_rows(session: AsyncSession, **filters: Any) -> list[RelatedGroupMember]: stmt = select(RelatedGroupMember).filter_by(**filters).order_by(RelatedGroupMember.valid_from) return list((await session.execute(stmt)).scalars().all()) async def _office(session: AsyncSession, name_like: str) -> RegionalOffice: return (await session.execute(select(RegionalOffice).where(RegionalOffice.name.like(name_like)))).scalar_one() async def _create( client: httpx.AsyncClient, group_id: Any, kind: str = "gsk", name: str = "ГСК А", **extra: Any ) -> httpx.Response: return await client.post(f"{GROUPS}/{group_id}/related-groups", json={"kind": kind, "name": name, **extra}) async def _add(client: httpx.AsyncClient, related_id: Any, entity_id: Any) -> httpx.Response: return await client.post(f"{RELATED}/{related_id}/members", json={"legal_entity_id": str(entity_id)}) @pytest.mark.integration async def test_create_gsk_and_gsz_codes_and_audit(auth_client: AuthClientFactory, seeded: AsyncSession) -> None: group = await make_company_group(seeded, name="ГК Бисквит") client = await auth_client(GO) gsk = await _create(client, group.id, "gsk", " ГСК Бисквит ", created_date="2025-01-15") gsz = await _create(client, group.id, "gsz", "ГСЗ Бисквит") assert gsk.status_code == HTTP_CREATED, gsk.text assert gsz.status_code == HTTP_CREATED, gsz.text gsk_body, gsz_body = gsk.json(), gsz.json() assert re.fullmatch(r"GSK-\d{6}", gsk_body["code"]) assert re.fullmatch(r"GSZ-\d{6}", gsz_body["code"]) assert gsk_body == { "id": gsk_body["id"], "code": gsk_body["code"], "company_group_id": str(group.id), "kind": "gsk", "name": "ГСК Бисквит", "created_date": "2025-01-15", "crm_id": None, "members_count": 0, } assert gsz_body["created_date"] is None events = await _events(seeded, "related_group") assert [str(event.object_id) for event in events] == [gsk_body["id"], gsz_body["id"]] first = events[0] assert (first.section, first.before_value) == (AuditSection.DIRECTORIES, None) assert first.after_value == { "code": gsk_body["code"], "company_group_id": str(group.id), "kind": "gsk", "name": "ГСК Бисквит", "created_date": "2025-01-15", } assert first.user_id is not None @pytest.mark.integration async def test_create_validation(auth_client: AuthClientFactory, seeded: AsyncSession) -> None: group = await make_company_group(seeded) client = await auth_client(GO) assert (await _create(client, group.id, "gsk", " ")).status_code == HTTP_UNPROCESSABLE assert (await _create(client, group.id, "gk", "Имя")).status_code == HTTP_UNPROCESSABLE missing = await _create(client, uuid.uuid4()) assert missing.status_code == HTTP_NOT_FOUND assert missing.json()["detail"]["message"] == "ГК не найдена" assert (await seeded.execute(select(func.count()).select_from(RelatedGroup))).scalar_one() == 0 assert not await _events(seeded) @pytest.mark.integration async def test_member_of_gsk_and_gsz_at_the_same_time(auth_client: AuthClientFactory, seeded: AsyncSession) -> None: """Сценарий спеки «Одновременно в ГСК и ГСЗ»; `kind` строки = `kind` группы (в БД).""" group = await make_company_group(seeded) entity = await make_legal_entity(seeded, name="АО Русский бисквит", group=group) client = await auth_client(GO) gsk = (await _create(client, group.id, "gsk", "ГСК")).json() gsz = (await _create(client, group.id, "gsz", "ГСЗ")).json() first = await _add(client, gsk["id"], entity.id) second = await _add(client, gsz["id"], entity.id) assert first.status_code == HTTP_CREATED, first.text assert second.status_code == HTTP_CREATED, second.text assert first.json() == { "legal_entity_id": str(entity.id), "inn": entity.inn, "name": "АО Русский бисквит", "valid_from": first.json()["valid_from"], } rows = await _member_rows(seeded, legal_entity_id=entity.id, is_current=True) assert sorted((str(row.related_group_id), row.kind) for row in rows) == sorted( [(gsk["id"], RelatedGroupKind.gsk), (gsz["id"], RelatedGroupKind.gsz)] ) for row in rows: related = await seeded.get(RelatedGroup, row.related_group_id) assert related is not None assert row.kind == related.kind assert row.valid_to is None assert row.created_by is not None card = (await client.get(f"{ENTITIES}/{entity.id}")).json() assert sorted(membership["kind"] for membership in card["memberships"]) == ["gsk", "gsz"] events = await _events(seeded, "related_group_member") assert len(events) == 2 # noqa: PLR2004 assert {event.object_id for event in events} == {row.id for row in rows} assert events[0].before_value is None assert events[0].after_value is not None assert events[0].after_value["legal_entity_id"] == str(entity.id) assert events[0].after_value["is_current"] is True @pytest.mark.integration async def test_second_gsk_is_conflict_with_current_group_id( auth_client: AuthClientFactory, seeded: AsyncSession ) -> None: """Сценарий спеки «Вторая ГСК запрещена»: 409 с id ГСК A, данные не меняются.""" group = await make_company_group(seeded) entity = await make_legal_entity(seeded, group=group) client = await auth_client(GO) gsk_a = (await _create(client, group.id, "gsk", "ГСК A")).json() gsk_b = (await _create(client, group.id, "gsk", "ГСК B")).json() assert (await _add(client, gsk_a["id"], entity.id)).status_code == HTTP_CREATED response = await _add(client, gsk_b["id"], entity.id) assert response.status_code == HTTP_CONFLICT detail = response.json()["detail"] assert detail["code"] == "MEMBER_OF_OTHER_GROUP" assert detail["related_group_id"] == gsk_a["id"] rows = await _member_rows(seeded, legal_entity_id=entity.id) assert [str(row.related_group_id) for row in rows] == [gsk_a["id"]] assert len(await _events(seeded, "related_group_member")) == 1 @pytest.mark.integration async def test_repeat_in_same_group_is_member_exists(auth_client: AuthClientFactory, seeded: AsyncSession) -> None: group = await make_company_group(seeded) entity = await make_legal_entity(seeded, group=group) client = await auth_client(GO) gsz = (await _create(client, group.id, "gsz", "ГСЗ")).json() assert (await _add(client, gsz["id"], entity.id)).status_code == HTTP_CREATED response = await _add(client, gsz["id"], entity.id) assert response.status_code == HTTP_CONFLICT assert response.json()["detail"]["code"] == "MEMBER_EXISTS" assert len(await _member_rows(seeded, legal_entity_id=entity.id)) == 1 @pytest.mark.integration async def test_remove_and_readd_creates_new_row(auth_client: AuthClientFactory, seeded: AsyncSession) -> None: """Сценарий спеки «Повторное включение после исключения» (SCD2).""" group = await make_company_group(seeded) entity = await make_legal_entity(seeded, group=group) client = await auth_client(GO) gsk = (await _create(client, group.id)).json() assert (await _add(client, gsk["id"], entity.id)).status_code == HTTP_CREATED removed = await client.delete(f"{RELATED}/{gsk['id']}/members/{entity.id}") assert removed.status_code == HTTP_NO_CONTENT, removed.text (closed,) = await _member_rows(seeded, legal_entity_id=entity.id) assert closed.is_current is False assert closed.valid_to is not None detail = (await client.get(f"{RELATED}/{gsk['id']}")).json() assert (detail["members_count"], detail["members"]) == (0, []) again = await _add(client, gsk["id"], entity.id) assert again.status_code == HTTP_CREATED, again.text rows = await _member_rows(seeded, legal_entity_id=entity.id) assert len(rows) == 2 # noqa: PLR2004 old, new = rows assert (old.id, old.is_current) == (closed.id, False) assert old.valid_to is not None assert (new.is_current, new.valid_to, new.kind) == (True, None, RelatedGroupKind.gsk) events = await _events(seeded, "related_group_member") assert len(events) == 3 # noqa: PLR2004 - добавление, исключение, повторное добавление removal = events[1] assert removal.object_id == closed.id assert removal.before_value is not None assert removal.after_value is not None assert (removal.before_value["is_current"], removal.after_value["is_current"]) == (True, False) assert removal.after_value["valid_to"] is not None @pytest.mark.integration async def test_remove_not_a_member_is_404(auth_client: AuthClientFactory, seeded: AsyncSession) -> None: group = await make_company_group(seeded) entity = await make_legal_entity(seeded, group=group) related = await make_related_group(seeded, group=group) await make_membership(seeded, related_group=related, entity=entity, is_current=False) client = await auth_client(GO) for entity_id in (entity.id, uuid.uuid4()): response = await client.delete(f"{RELATED}/{related.id}/members/{entity_id}") assert response.status_code == HTTP_NOT_FOUND assert not await _events(seeded) @pytest.mark.integration async def test_delete_non_empty_group_is_conflict(auth_client: AuthClientFactory, seeded: AsyncSession) -> None: """Сценарий спеки «Удаление непустой группы».""" group = await make_company_group(seeded) entity = await make_legal_entity(seeded, group=group) related = await make_related_group(seeded, group=group) await make_membership(seeded, related_group=related, entity=entity) client = await auth_client(ADMIN) response = await client.delete(f"{RELATED}/{related.id}") assert response.status_code == HTTP_CONFLICT assert response.json()["detail"]["code"] == "GROUP_NOT_EMPTY" assert await seeded.get(RelatedGroup, related.id) is not None assert not await _events(seeded) @pytest.mark.integration async def test_delete_empty_group_removes_closed_rows(auth_client: AuthClientFactory, seeded: AsyncSession) -> None: group = await make_company_group(seeded) entity = await make_legal_entity(seeded, group=group) related = await make_related_group(seeded, group=group, kind=RelatedGroupKind.gsz, name="ГСЗ пустая") closed = await make_membership(seeded, related_group=related, entity=entity, is_current=False) related_id, code = related.id, related.code client = await auth_client(ADMIN) response = await client.delete(f"{RELATED}/{related_id}") assert response.status_code == HTTP_NO_CONTENT, response.text seeded.expire_all() assert (await seeded.execute(select(RelatedGroup).where(RelatedGroup.id == related_id))).first() is None assert not await _member_rows(seeded, related_group_id=related_id) (event,) = await _events(seeded, "related_group") assert event.object_id == related_id assert event.after_value is None assert event.before_value is not None assert event.before_value["code"] == code assert event.before_value["kind"] == "gsz" assert event.before_value["name"] == "ГСЗ пустая" assert [row["id"] for row in event.before_value["members"]] == [str(closed.id)] assert (await client.get(f"{RELATED}/{related_id}")).status_code == HTTP_NOT_FOUND @pytest.mark.integration async def test_delete_requires_directory_delete(auth_client: AuthClientFactory, seeded: AsyncSession) -> None: group = await make_company_group(seeded) related = await make_related_group(seeded, group=group) client = await auth_client(GO) # KA_GO: VIEW + EDIT, без DELETE assert (await client.delete(f"{RELATED}/{related.id}")).status_code == HTTP_FORBIDDEN assert await seeded.get(RelatedGroup, related.id) is not None @pytest.mark.integration async def test_entity_of_other_company_group_is_422(auth_client: AuthClientFactory, seeded: AsyncSession) -> None: group = await make_company_group(seeded, name="ГК 1") other = await make_company_group(seeded, name="ГК 2") foreign = await make_legal_entity(seeded, group=other) mono = await make_legal_entity(seeded, name="ООО Монозаёмщик") client = await auth_client(GO) gsk = (await _create(client, group.id)).json() for entity in (foreign, mono): response = await _add(client, gsk["id"], entity.id) assert response.status_code == HTTP_UNPROCESSABLE assert response.json()["detail"]["code"] == "DIFFERENT_COMPANY_GROUP" assert not await _member_rows(seeded) @pytest.mark.integration async def test_closed_entity_without_group_is_422(auth_client: AuthClientFactory, seeded: AsyncSession) -> None: """Review Focus 4: закрытый ЮЛ (`close_date`, `company_group_id = NULL`) не в той же ГК.""" group = await make_company_group(seeded) closed = await make_legal_entity( seeded, name="ООО Закрытое", close_date=datetime.date(2025, 1, 1), office=await _office(seeded, "Головной%") ) assert closed.company_group_id is None client = await auth_client(GO) gsk = (await _create(client, group.id)).json() response = await _add(client, gsk["id"], closed.id) assert response.status_code == HTTP_UNPROCESSABLE assert response.json()["detail"]["code"] == "DIFFERENT_COMPANY_GROUP" assert not await _member_rows(seeded) assert not await _events(seeded, "related_group_member") @pytest.mark.integration async def test_patch_name_and_created_date(auth_client: AuthClientFactory, seeded: AsyncSession) -> None: group = await make_company_group(seeded) client = await auth_client(GO) gsk = (await _create(client, group.id, "gsk", "Старое")).json() response = await client.patch(f"{RELATED}/{gsk['id']}", json={"name": " Новое ", "created_date": "2024-12-31"}) assert response.status_code == HTTP_OK, response.text body = response.json() assert (body["name"], body["created_date"], body["kind"], body["code"]) == ( "Новое", "2024-12-31", "gsk", gsk["code"], ) events = await _events(seeded, "related_group") assert len(events) == 2 # noqa: PLR2004 - создание и изменение assert events[1].before_value == {"name": "Старое", "created_date": None} assert events[1].after_value == {"name": "Новое", "created_date": "2024-12-31"} same = await client.patch(f"{RELATED}/{gsk['id']}", json={"name": "Новое"}) assert same.status_code == HTTP_OK assert len(await _events(seeded, "related_group")) == 2 # noqa: PLR2004 - no-op без аудита one = await client.patch(f"{RELATED}/{gsk['id']}", json={"name": "Третье"}) assert one.status_code == HTTP_OK assert (await _events(seeded, "related_group"))[-1].changed_field == "name" for bad in ({"name": " "}, {"name": None}, {"kind": "gsz"}): assert (await client.patch(f"{RELATED}/{gsk['id']}", json=bad)).status_code == HTTP_UNPROCESSABLE @pytest.mark.integration async def test_list_and_detail_with_current_members(auth_client: AuthClientFactory, seeded: AsyncSession) -> None: group = await make_company_group(seeded) other_group = await make_company_group(seeded, name="Другая ГК") first = await make_legal_entity(seeded, name="Б ЮЛ", group=group) second = await make_legal_entity(seeded, name="А ЮЛ", group=group) gone = await make_legal_entity(seeded, name="В ЮЛ", group=group) gsz = await make_related_group(seeded, group=group, kind=RelatedGroupKind.gsz, name="Б ГСЗ") gsk = await make_related_group(seeded, group=group, kind=RelatedGroupKind.gsk, name="А ГСК") await make_related_group(seeded, group=other_group, name="Чужая") await make_membership(seeded, related_group=gsk, entity=first) await make_membership(seeded, related_group=gsk, entity=second) await make_membership(seeded, related_group=gsk, entity=gone, is_current=False) client = await auth_client(EXPERT) # только DIRECTORY_VIEW, ГО listed = await client.get(f"{GROUPS}/{group.id}/related-groups") assert listed.status_code == HTTP_OK, listed.text assert [(row["name"], row["members_count"]) for row in listed.json()] == [("А ГСК", 2), ("Б ГСЗ", 0)] detail = await client.get(f"{RELATED}/{gsk.id}") assert detail.status_code == HTTP_OK body = detail.json() assert body["members_count"] == 2 # noqa: PLR2004 assert [member["name"] for member in body["members"]] == ["А ЮЛ", "Б ЮЛ"] assert set(body["members"][0]) == {"legal_entity_id", "inn", "name", "valid_from"} assert body["members"][0]["inn"] == second.inn assert (await client.get(f"{RELATED}/{gsz.id}")).json()["members"] == [] @pytest.mark.integration async def test_expert_cannot_edit(auth_client: AuthClientFactory, seeded: AsyncSession) -> None: group = await make_company_group(seeded) entity = await make_legal_entity(seeded, group=group) related = await make_related_group(seeded, group=group) client = await auth_client(EXPERT) assert (await _create(client, group.id)).status_code == HTTP_FORBIDDEN assert (await client.patch(f"{RELATED}/{related.id}", json={"name": "X"})).status_code == HTTP_FORBIDDEN assert (await _add(client, related.id, entity.id)).status_code == HTTP_FORBIDDEN assert (await client.delete(f"{RELATED}/{related.id}/members/{entity.id}")).status_code == HTTP_FORBIDDEN assert not await _member_rows(seeded) assert not await _events(seeded) @pytest.mark.integration async def test_branch_user_gets_404_for_foreign_company_group( auth_client: AuthClientFactory, seeded: AsyncSession ) -> None: """`ka_rf` (Ростов) и группа ГК Краснодара — 404 с тем же телом, что у несуществующего id.""" krasnodar = await _office(seeded, KRASNODAR) foreign_group = await make_company_group(seeded, name="ГК Краснодар", office=krasnodar) foreign_entity = await make_legal_entity(seeded, group=foreign_group) foreign = await make_related_group(seeded, group=foreign_group) await make_membership(seeded, related_group=foreign, entity=foreign_entity) client = await auth_client(KA_RF) missing = await client.get(f"{RELATED}/{uuid.uuid4()}") assert missing.status_code == HTTP_NOT_FOUND responses = [ await client.get(f"{RELATED}/{foreign.id}"), await client.patch(f"{RELATED}/{foreign.id}", json={"name": "X"}), await _add(client, foreign.id, foreign_entity.id), await client.delete(f"{RELATED}/{foreign.id}/members/{foreign_entity.id}"), ] for response in responses: assert response.status_code == HTTP_NOT_FOUND, response.text assert response.json() == missing.json() missing_group = await client.get(f"{GROUPS}/{uuid.uuid4()}/related-groups") for response in ( await client.get(f"{GROUPS}/{foreign_group.id}/related-groups"), await _create(client, foreign_group.id), ): assert response.status_code == HTTP_NOT_FOUND assert response.json() == missing_group.json() assert (await seeded.get(RelatedGroup, foreign.id)) is not None assert len(await _member_rows(seeded, is_current=True)) == 1 assert not await _events(seeded) @pytest.mark.integration async def test_branch_user_works_in_own_company_group(auth_client: AuthClientFactory, seeded: AsyncSession) -> None: rostov = await _office(seeded, "Ростов%") krasnodar = await _office(seeded, KRASNODAR) own_group = await make_company_group(seeded, name="ГК Ростов", office=rostov) own = await make_legal_entity(seeded, group=own_group) foreign_group = await make_company_group(seeded, name="ГК Краснодар", office=krasnodar) foreign_entity = await make_legal_entity(seeded, group=foreign_group) client = await auth_client(KA_RF) gsk = await _create(client, own_group.id) assert gsk.status_code == HTTP_CREATED, gsk.text assert (await _add(client, gsk.json()["id"], own.id)).status_code == HTTP_CREATED invisible = await _add(client, gsk.json()["id"], foreign_entity.id) assert invisible.status_code == HTTP_NOT_FOUND # ЮЛ вне видимости — 404, не 422 и не утечка assert invisible.json()["detail"]["message"] == "ЮЛ не найдено" rows = await _member_rows(seeded, is_current=True) assert [row.legal_entity_id for row in rows] == [own.id]