462a30fcf6
`office_for_create` применяется к `regional_office_id` только при фактической смене филиала: тело с тем же значением — no-op (Review Focus 1), правка `name` в том же запросе проходит. Тест на ЮЛ в ГК своего филиала с чужим `regional_office_id`. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
566 lines
29 KiB
Python
566 lines
29 KiB
Python
"""API ЮЛ/ИП (spec directories-api, directory-policy, audit-trail): CRUD, перевод, вывод из состава, история, откат."""
|
||
|
||
import datetime
|
||
import uuid
|
||
from collections.abc import Sequence
|
||
from typing import Any
|
||
|
||
import httpx
|
||
import pytest
|
||
from fastapi import FastAPI
|
||
from sqlalchemy import func, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from src.models.audit_ex import AuditEvent
|
||
from src.models.directories import (
|
||
LegalEntity,
|
||
LegalEntityAlias,
|
||
LegalEntityHistory,
|
||
RegionalOffice,
|
||
RelatedGroupMember,
|
||
)
|
||
from src.models.enums import AuditAction, AuditSection, EntityType, 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_FORBIDDEN = 403
|
||
HTTP_NOT_FOUND = 404
|
||
HTTP_CONFLICT = 409
|
||
HTTP_UNPROCESSABLE = 422
|
||
HTTP_SERVER_ERROR = 500
|
||
ROSTOV = "Ростовский региональный филиал"
|
||
KRASNODAR = "Краснодарский региональный филиал"
|
||
ENTITIES = "/directories/legal-entities"
|
||
ENTITY_NOT_FOUND = "ЮЛ не найдено"
|
||
GROUP_NOT_FOUND = "ГК не найдена"
|
||
|
||
|
||
def _today() -> datetime.date:
|
||
"""«Сегодня» сервиса — дата UTC (тест не зависит от часового пояса машины)."""
|
||
return datetime.datetime.now(datetime.UTC).date()
|
||
|
||
|
||
async def _office(session: AsyncSession, name: str) -> RegionalOffice:
|
||
return (await session.execute(select(RegionalOffice).where(RegionalOffice.name == name))).scalar_one()
|
||
|
||
|
||
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 _history_count(session: AsyncSession) -> int:
|
||
return (await session.execute(select(func.count()).select_from(LegalEntityHistory))).scalar_one()
|
||
|
||
|
||
async def _create(client: httpx.AsyncClient, **body: Any) -> dict[str, Any]:
|
||
response = await client.post(ENTITIES, json=body)
|
||
assert response.status_code == HTTP_CREATED, response.text
|
||
created: dict[str, Any] = response.json()
|
||
return created
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_sole_proprietor_with_ten_digits_is_invalid(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
client = await auth_client("ka_go@armka.local")
|
||
response = await client.post(ENTITIES, json={"inn": "3528052718", "entity_type": "ИП", "name": "ИП Иванов"})
|
||
assert response.status_code == HTTP_UNPROCESSABLE
|
||
assert response.json()["detail"]["code"] == "INVALID_INN"
|
||
assert (await seeded.execute(select(func.count()).select_from(LegalEntity))).scalar_one() == 0
|
||
assert not await _events(seeded, AuditAction.LEGAL_ENTITY_CHANGE)
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_duplicate_inn_is_conflict(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
client = await auth_client("ka_go@armka.local")
|
||
await _create(client, inn="3528052718", entity_type="ЮЛ", name="ООО Первое")
|
||
response = await client.post(ENTITIES, json={"inn": "35 2805 2718", "entity_type": "ЮЛ", "name": "ООО Второе"})
|
||
assert response.status_code == HTTP_CONFLICT
|
||
assert response.json()["detail"]["code"] == "INN_EXISTS"
|
||
assert len(await _events(seeded, AuditAction.LEGAL_ENTITY_CHANGE)) == 1
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_create_audit_and_default_office_from_group(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
rostov = await _office(seeded, ROSTOV)
|
||
group = await make_company_group(seeded, name="ГК Ростов", office=rostov)
|
||
client = await auth_client("ka_go@armka.local")
|
||
created = await _create(
|
||
client, inn="0123456789", entity_type="ЮЛ", name=" ООО Ноль ", company_group_id=str(group.id), crm_id="1-AB"
|
||
)
|
||
assert created["inn"] == "0123456789" # Review Focus 2: строка, ведущий ноль сохранён
|
||
assert created["name"] == "ООО Ноль"
|
||
assert created["regional_office_id"] == str(rostov.id)
|
||
assert (created["outside_group"], created["closed"]) == (False, False)
|
||
assert created["valid_from"]
|
||
|
||
(event,) = await _events(seeded, AuditAction.LEGAL_ENTITY_CHANGE)
|
||
assert (event.section, event.object_type, str(event.object_id)) == (
|
||
AuditSection.DIRECTORIES,
|
||
"legal_entity",
|
||
created["id"],
|
||
)
|
||
assert event.after_value == {
|
||
"inn": "0123456789",
|
||
"entity_type": "ЮЛ",
|
||
"name": "ООО Ноль",
|
||
"crm_id": "1-AB",
|
||
"company_group_id": str(group.id),
|
||
"regional_office_id": str(rostov.id),
|
||
}
|
||
assert event.before_value is None
|
||
assert await _history_count(seeded) == 0 # при создании история не пишется (D-6)
|
||
|
||
duplicate_crm = await client.post(
|
||
ENTITIES, json={"inn": "3528052718", "entity_type": "ЮЛ", "name": "Х", "crm_id": "1-AB"}
|
||
)
|
||
assert duplicate_crm.status_code == HTTP_CONFLICT
|
||
assert duplicate_crm.json()["detail"]["code"] == "CRM_ID_EXISTS"
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_mono_borrower_visible_to_own_branch_and_head_office(
|
||
auth_client: AuthClientFactory, seeded: AsyncSession
|
||
) -> None:
|
||
rostov = await _office(seeded, ROSTOV)
|
||
branch = await auth_client("ka_rf@armka.local") # Ростовский филиал
|
||
created = await _create(branch, inn="6164000001", entity_type="ЮЛ", name="ООО Монозаёмщик")
|
||
assert created["company_group_id"] is None
|
||
assert created["outside_group"] is True
|
||
assert created["regional_office_id"] == str(rostov.id)
|
||
|
||
for email in ("ka_rf@armka.local", "ka_go@armka.local"):
|
||
client = await auth_client(email)
|
||
page = (await client.get(ENTITIES, params={"outside_group": "true"})).json()
|
||
assert created["id"] in {item["id"] for item in page["items"]}, email
|
||
assert (await client.get(f"{ENTITIES}/{created['id']}")).status_code == HTTP_OK
|
||
|
||
foreign = await auth_client("client_manager@armka.local") # Краснодарский филиал
|
||
page = (await foreign.get(ENTITIES)).json()
|
||
assert created["id"] not in {item["id"] for item in page["items"]}
|
||
card = await foreign.get(f"{ENTITIES}/{created['id']}")
|
||
missing = await foreign.get(f"{ENTITIES}/{uuid.uuid4()}")
|
||
assert card.status_code == missing.status_code == HTTP_NOT_FOUND
|
||
assert card.json() == missing.json()
|
||
assert card.json()["detail"]["message"] == ENTITY_NOT_FOUND
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_move_closes_memberships_and_writes_history(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
rostov = await _office(seeded, ROSTOV)
|
||
old_group = await make_company_group(seeded, name="ГК Старая", office=rostov)
|
||
new_group = await make_company_group(seeded, name="ГК Новая", office=rostov)
|
||
entity = await make_legal_entity(seeded, name="ООО Переводимое", group=old_group)
|
||
gsk = await make_related_group(seeded, group=old_group, kind=RelatedGroupKind.gsk)
|
||
gsz = await make_related_group(seeded, group=old_group, kind=RelatedGroupKind.gsz, name="ГСЗ Тест")
|
||
gsk_member = await make_membership(seeded, related_group=gsk, entity=entity)
|
||
gsz_member = await make_membership(seeded, related_group=gsz, entity=entity)
|
||
old_valid_from = entity.valid_from
|
||
|
||
client = await auth_client("ka_rf@armka.local")
|
||
effective = _today()
|
||
response = await client.post(
|
||
f"{ENTITIES}/{entity.id}/move",
|
||
json={"company_group_id": str(new_group.id), "effective_date": effective.isoformat()},
|
||
)
|
||
assert response.status_code == HTTP_OK, response.text
|
||
assert response.json()["company_group_id"] == str(new_group.id)
|
||
|
||
for member in (gsk_member, gsz_member):
|
||
await seeded.refresh(member)
|
||
assert member.is_current is False
|
||
assert member.valid_to is not None
|
||
|
||
history = (await client.get(f"{ENTITIES}/{entity.id}/history")).json()
|
||
assert len(history) == 1
|
||
snapshot = history[0]
|
||
assert snapshot["event_type"] == "MOVE_TO_GROUP"
|
||
assert snapshot["effective_date"] == effective.isoformat()
|
||
assert snapshot["company_group_id"] == str(old_group.id) # снимок — прежнее состояние
|
||
assert datetime.datetime.fromisoformat(snapshot["valid_from"]) == old_valid_from
|
||
assert snapshot["valid_to"] is not None and snapshot["changed_by"] is not None
|
||
|
||
(change,) = await _events(seeded, AuditAction.LEGAL_ENTITY_CHANGE)
|
||
assert change.before_value == {"company_group_id": str(old_group.id)}
|
||
assert change.after_value == {"company_group_id": str(new_group.id)}
|
||
assert change.changed_field == "company_group_id"
|
||
memberships = await _events(seeded, AuditAction.MEMBERSHIP_CHANGE)
|
||
assert {(e.object_type, e.object_id) for e in memberships} == {
|
||
("related_group_member", gsk_member.id),
|
||
("related_group_member", gsz_member.id),
|
||
}
|
||
assert all(e.before_value["is_current"] is True and e.after_value["is_current"] is False for e in memberships)
|
||
|
||
card = (await client.get(f"{ENTITIES}/{entity.id}")).json()
|
||
assert card["memberships"] == []
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_branch_user_cannot_move_to_foreign_group(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
"""Review Focus 3: целевая ГК чужого филиала — 404, как несуществующая (не 422, без утечки)."""
|
||
rostov, krasnodar = await _office(seeded, ROSTOV), await _office(seeded, KRASNODAR)
|
||
own = await make_company_group(seeded, name="ГК Своя", office=rostov)
|
||
foreign = await make_company_group(seeded, name="ГК Чужая", office=krasnodar)
|
||
entity = await make_legal_entity(seeded, group=own)
|
||
client = await auth_client("ka_rf@armka.local")
|
||
|
||
body = {"effective_date": _today().isoformat()}
|
||
to_foreign = await client.post(f"{ENTITIES}/{entity.id}/move", json={**body, "company_group_id": str(foreign.id)})
|
||
to_missing = await client.post(f"{ENTITIES}/{entity.id}/move", json={**body, "company_group_id": str(uuid.uuid4())})
|
||
assert to_foreign.status_code == to_missing.status_code == HTTP_NOT_FOUND
|
||
assert to_foreign.json() == to_missing.json()
|
||
assert to_foreign.json()["detail"]["message"] == GROUP_NOT_FOUND
|
||
await seeded.refresh(entity)
|
||
assert entity.company_group_id == own.id
|
||
assert await _history_count(seeded) == 0
|
||
assert not await _events(seeded, AuditAction.LEGAL_ENTITY_CHANGE)
|
||
|
||
create_in_foreign = await client.post(
|
||
ENTITIES, json={"inn": "6164000002", "entity_type": "ЮЛ", "name": "ООО Х", "company_group_id": str(foreign.id)}
|
||
)
|
||
assert create_in_foreign.status_code == HTTP_NOT_FOUND
|
||
assert create_in_foreign.json() == to_missing.json()
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_effective_date_in_past_is_invalid(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
group = await make_company_group(seeded, name="ГК А")
|
||
target = await make_company_group(seeded, name="ГК Б")
|
||
entity = await make_legal_entity(seeded, group=group)
|
||
client = await auth_client("ka_go@armka.local")
|
||
yesterday = (_today() - datetime.timedelta(days=1)).isoformat()
|
||
|
||
move = await client.post(
|
||
f"{ENTITIES}/{entity.id}/move", json={"company_group_id": str(target.id), "effective_date": yesterday}
|
||
)
|
||
assert move.status_code == HTTP_UNPROCESSABLE
|
||
assert move.json()["detail"]["code"] == "EFFECTIVE_DATE_IN_PAST"
|
||
withdraw = await client.post(
|
||
f"{ENTITIES}/{entity.id}/withdraw", json={"event": "CLOSE", "effective_date": yesterday}
|
||
)
|
||
assert withdraw.status_code == HTTP_UNPROCESSABLE
|
||
assert withdraw.json()["detail"]["code"] == "EFFECTIVE_DATE_IN_PAST"
|
||
await seeded.refresh(entity)
|
||
assert (entity.company_group_id, entity.close_date) == (group.id, None)
|
||
assert await _history_count(seeded) == 0
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_withdraw_close(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
group = await make_company_group(seeded, name="ГК Закрытие")
|
||
entity = await make_legal_entity(seeded, name="ООО Закрываемое", group=group)
|
||
other = await make_legal_entity(seeded, name="ООО Живое", group=group)
|
||
gsk = await make_related_group(seeded, group=group)
|
||
member = await make_membership(seeded, related_group=gsk, entity=entity)
|
||
client = await auth_client("ka_go@armka.local")
|
||
close_date = _today() + datetime.timedelta(days=3)
|
||
|
||
response = await client.post(
|
||
f"{ENTITIES}/{entity.id}/withdraw",
|
||
json={"event": "CLOSE", "effective_date": close_date.isoformat(), "comment": "Ликвидация"},
|
||
)
|
||
assert response.status_code == HTTP_OK, response.text
|
||
body = response.json()
|
||
assert body["close_date"] == close_date.isoformat()
|
||
assert body["company_group_id"] is None
|
||
assert body["closed"] is True
|
||
|
||
closed = (await client.get(ENTITIES, params={"closed": "true"})).json()
|
||
assert [item["id"] for item in closed["items"]] == [str(entity.id)]
|
||
assert closed["items"][0]["closed"] is True
|
||
active_ids = {item["id"] for item in (await client.get(ENTITIES, params={"closed": "false"})).json()["items"]}
|
||
assert str(other.id) in active_ids and str(entity.id) not in active_ids
|
||
|
||
await seeded.refresh(member)
|
||
assert member.is_current is False
|
||
history = (await client.get(f"{ENTITIES}/{entity.id}/history")).json()
|
||
assert [(row["event_type"], row["effective_date"], row["company_group_id"]) for row in history] == [
|
||
("CLOSE", close_date.isoformat(), str(group.id))
|
||
]
|
||
(change,) = await _events(seeded, AuditAction.LEGAL_ENTITY_CHANGE)
|
||
assert change.comment == "Ликвидация"
|
||
assert change.before_value == {"company_group_id": str(group.id), "close_date": None}
|
||
assert change.after_value == {"company_group_id": None, "close_date": close_date.isoformat()}
|
||
assert len(await _events(seeded, AuditAction.MEMBERSHIP_CHANGE)) == 1
|
||
|
||
again = await client.post(
|
||
f"{ENTITIES}/{entity.id}/withdraw", json={"event": "CLOSE", "effective_date": close_date.isoformat()}
|
||
)
|
||
assert again.status_code == HTTP_CONFLICT
|
||
assert again.json()["detail"]["code"] == "ALREADY_CLOSED"
|
||
target = await make_company_group(seeded, name="ГК Куда")
|
||
move_closed = await client.post(
|
||
f"{ENTITIES}/{entity.id}/move",
|
||
json={"company_group_id": str(target.id), "effective_date": _today().isoformat()},
|
||
)
|
||
assert move_closed.status_code == HTTP_CONFLICT
|
||
assert move_closed.json()["detail"]["code"] == "LEGAL_ENTITY_CLOSED"
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_withdraw_remove_from_group(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
group = await make_company_group(seeded, name="ГК Вывод")
|
||
entity = await make_legal_entity(seeded, group=group)
|
||
client = await auth_client("ka_go@armka.local")
|
||
today = _today().isoformat()
|
||
|
||
response = await client.post(
|
||
f"{ENTITIES}/{entity.id}/withdraw", json={"event": "REMOVE_FROM_GROUP", "effective_date": today}
|
||
)
|
||
assert response.status_code == HTTP_OK, response.text
|
||
assert (response.json()["company_group_id"], response.json()["close_date"]) == (None, None)
|
||
assert response.json()["outside_group"] is True
|
||
history = (await client.get(f"{ENTITIES}/{entity.id}/history")).json()
|
||
assert [row["event_type"] for row in history] == ["REMOVE_FROM_GROUP"]
|
||
|
||
again = await client.post(
|
||
f"{ENTITIES}/{entity.id}/withdraw", json={"event": "REMOVE_FROM_GROUP", "effective_date": today}
|
||
)
|
||
assert again.status_code == HTTP_CONFLICT
|
||
assert again.json()["detail"]["code"] == "NOT_IN_COMPANY_GROUP"
|
||
same = await client.post(f"{ENTITIES}/{entity.id}/withdraw", json={"event": "UNKNOWN", "effective_date": today})
|
||
assert same.status_code == HTTP_UNPROCESSABLE
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_move_to_same_group_is_conflict(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
group = await make_company_group(seeded, name="ГК Та же")
|
||
entity = await make_legal_entity(seeded, group=group)
|
||
client = await auth_client("ka_go@armka.local")
|
||
response = await client.post(
|
||
f"{ENTITIES}/{entity.id}/move", json={"company_group_id": str(group.id), "effective_date": _today().isoformat()}
|
||
)
|
||
assert response.status_code == HTTP_CONFLICT
|
||
assert response.json()["detail"]["code"] == "SAME_COMPANY_GROUP"
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_noop_patch_writes_no_history_and_no_audit(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
"""Review Focus 1: тело с теми же значениями не плодит историю и аудит."""
|
||
client = await auth_client("ka_go@armka.local")
|
||
created = await _create(client, inn="3528052718", entity_type="ЮЛ", name="ООО Без изменений", crm_id="1-NOOP")
|
||
audit_before = len(await _events(seeded, AuditAction.LEGAL_ENTITY_CHANGE))
|
||
|
||
response = await client.patch(
|
||
f"{ENTITIES}/{created['id']}", json={"name": " ООО Без изменений ", "entity_type": "ЮЛ", "crm_id": "1-NOOP"}
|
||
)
|
||
assert response.status_code == HTTP_OK
|
||
assert response.json()["valid_from"] == created["valid_from"]
|
||
assert await _history_count(seeded) == 0
|
||
assert len(await _events(seeded, AuditAction.LEGAL_ENTITY_CHANGE)) == audit_before
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_patch_writes_update_snapshot_and_audit(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
client = await auth_client("ka_go@armka.local")
|
||
created = await _create(client, inn="3528052718", entity_type="ЮЛ", name="Имя 0")
|
||
for index in (1, 2):
|
||
response = await client.patch(f"{ENTITIES}/{created['id']}", json={"name": f"Имя {index}"})
|
||
assert response.status_code == HTTP_OK, response.text
|
||
assert response.json()["valid_from"] != created["valid_from"]
|
||
|
||
history = (await client.get(f"{ENTITIES}/{created['id']}/history")).json()
|
||
assert [(row["name"], row["event_type"]) for row in history] == [("Имя 1", "UPDATE"), ("Имя 0", "UPDATE")] # R-3
|
||
assert history[0]["effective_date"] is None
|
||
assert set(history[0]) == {
|
||
"id",
|
||
"inn",
|
||
"crm_id",
|
||
"entity_type",
|
||
"name",
|
||
"company_group_id",
|
||
"regional_office_id",
|
||
"close_date",
|
||
"event_type",
|
||
"effective_date",
|
||
"valid_from",
|
||
"valid_to",
|
||
"changed_by",
|
||
}
|
||
last = (await _events(seeded, AuditAction.LEGAL_ENTITY_CHANGE))[-1]
|
||
assert (last.changed_field, last.before_value, last.after_value) == ("name", {"name": "Имя 1"}, {"name": "Имя 2"})
|
||
|
||
wrong_type = await client.patch(f"{ENTITIES}/{created['id']}", json={"entity_type": "ИП"})
|
||
assert wrong_type.status_code == HTTP_UNPROCESSABLE
|
||
assert wrong_type.json()["detail"]["code"] == "INVALID_INN" # 10 цифр не ИНН ИП
|
||
extra = await client.patch(f"{ENTITIES}/{created['id']}", json={"inn": "0000000000"})
|
||
assert extra.status_code == HTTP_UNPROCESSABLE # extra="forbid": ИНН не меняется PATCH
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_inn_filter_strips_spaces_and_q_searches_name_and_inn(
|
||
auth_client: AuthClientFactory, seeded: AsyncSession
|
||
) -> None:
|
||
"""Review Focus 2 и R-4: `inn` — точное сравнение после strip; `q` — подстрока name/inn, LIKE экранирован."""
|
||
target = await make_legal_entity(seeded, name="ООО Вологодский", inn="3528052718")
|
||
await make_legal_entity(seeded, name="ООО 100% Другое", inn="3528052719")
|
||
client = await auth_client("ka_go@armka.local")
|
||
|
||
by_inn = (await client.get(ENTITIES, params={"inn": " 3528052718 "})).json()
|
||
assert [item["id"] for item in by_inn["items"]] == [str(target.id)]
|
||
assert by_inn["total"] == 1
|
||
assert (await client.get(ENTITIES, params={"inn": "352805271"})).json()["total"] == 0 # не подстрока
|
||
|
||
assert {item["name"] for item in (await client.get(ENTITIES, params={"q": "вологод"})).json()["items"]} == {
|
||
"ООО Вологодский"
|
||
}
|
||
by_inn_part = (await client.get(ENTITIES, params={"q": "52805271"})).json()
|
||
assert by_inn_part["total"] == 2 # noqa: PLR2004
|
||
assert [item["name"] for item in by_inn_part["items"]] == ["ООО 100% Другое", "ООО Вологодский"]
|
||
percent = (await client.get(ENTITIES, params={"q": "%"})).json()
|
||
assert [item["name"] for item in percent["items"]] == ["ООО 100% Другое"]
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_list_filters_by_group_and_outside_group(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
group = await make_company_group(seeded, name="ГК Фильтр")
|
||
in_group = await make_legal_entity(seeded, name="ООО В группе", group=group)
|
||
mono = await make_legal_entity(seeded, name="ООО Сам по себе")
|
||
client = await auth_client("ka_go@armka.local")
|
||
|
||
by_group = (await client.get(ENTITIES, params={"company_group_id": str(group.id)})).json()
|
||
assert [item["id"] for item in by_group["items"]] == [str(in_group.id)]
|
||
outside = {item["id"] for item in (await client.get(ENTITIES, params={"outside_group": "true"})).json()["items"]}
|
||
assert str(mono.id) in outside and str(in_group.id) not in outside
|
||
inside = {item["id"] for item in (await client.get(ENTITIES, params={"outside_group": "false"})).json()["items"]}
|
||
assert str(in_group.id) in inside and str(mono.id) not in inside
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_card_has_current_memberships_and_aliases(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
group = await make_company_group(seeded, name="ГК Карточка")
|
||
entity = await make_legal_entity(seeded, name="ООО Карточка", group=group)
|
||
gsk = await make_related_group(seeded, group=group, name="ГСК Карточка")
|
||
old_gsz = await make_related_group(seeded, group=group, kind=RelatedGroupKind.gsz, name="ГСЗ Старая")
|
||
await make_membership(seeded, related_group=gsk, entity=entity)
|
||
await make_membership(seeded, related_group=old_gsz, entity=entity, is_current=False)
|
||
seeded.add(
|
||
LegalEntityAlias(
|
||
legal_entity_id=entity.id,
|
||
alias="ООО «Карточка»",
|
||
normalized_alias="карточка",
|
||
created_at=datetime.datetime.now(datetime.UTC),
|
||
)
|
||
)
|
||
await seeded.flush()
|
||
client = await auth_client("ka_go@armka.local")
|
||
|
||
card = (await client.get(f"{ENTITIES}/{entity.id}")).json()
|
||
assert [(m["related_group_id"], m["code"], m["kind"], m["name"]) for m in card["memberships"]] == [
|
||
(str(gsk.id), gsk.code, "gsk", "ГСК Карточка")
|
||
]
|
||
assert card["memberships"][0]["valid_from"]
|
||
assert [(a["alias"], a["normalized_alias"]) for a in card["aliases"]] == [("ООО «Карточка»", "карточка")]
|
||
assert card["entity_type"] == EntityType.ЮЛ.value
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_foreign_entity_operations_are_404(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
krasnodar = await _office(seeded, KRASNODAR)
|
||
foreign_group = await make_company_group(seeded, name="ГК Краснодар", office=krasnodar)
|
||
entity = await make_legal_entity(seeded, group=foreign_group)
|
||
client = await auth_client("ka_rf@armka.local")
|
||
missing = await client.get(f"{ENTITIES}/{uuid.uuid4()}")
|
||
today = _today().isoformat()
|
||
|
||
for call in (
|
||
client.get(f"{ENTITIES}/{entity.id}"),
|
||
client.patch(f"{ENTITIES}/{entity.id}", json={"name": "Взлом"}),
|
||
client.post(
|
||
f"{ENTITIES}/{entity.id}/move", json={"company_group_id": str(foreign_group.id), "effective_date": today}
|
||
),
|
||
client.post(f"{ENTITIES}/{entity.id}/withdraw", json={"event": "CLOSE", "effective_date": today}),
|
||
client.get(f"{ENTITIES}/{entity.id}/history"),
|
||
):
|
||
response = await call
|
||
assert response.status_code == HTTP_NOT_FOUND
|
||
assert response.json() == missing.json()
|
||
|
||
|
||
@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(ENTITIES, json={"inn": "3528052718", "entity_type": "ЮЛ", "name": "ООО Эксперта"})
|
||
assert response.status_code == HTTP_FORBIDDEN
|
||
assert not await _events(seeded, AuditAction.LEGAL_ENTITY_CHANGE)
|
||
assert (await client.get(ENTITIES)).status_code == HTTP_OK
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_failed_move_rolls_back_everything(
|
||
app: FastAPI, auth_client: AuthClientFactory, seeded: AsyncSession, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""spec «Ошибка внутри мутации»: сбой закрытия членств → 500, ЮЛ в прежней ГК, истории и аудита нет."""
|
||
old_group = await make_company_group(seeded, name="ГК Откат")
|
||
new_group = await make_company_group(seeded, name="ГК Цель")
|
||
entity = await make_legal_entity(seeded, group=old_group)
|
||
gsk = await make_related_group(seeded, group=old_group)
|
||
member = await make_membership(seeded, related_group=gsk, entity=entity)
|
||
entity_id, member_id, old_group_id = entity.id, member.id, old_group.id # откат SAVEPOINT протухит ORM-объекты
|
||
authorized = await auth_client("ka_go@armka.local")
|
||
|
||
async def broken_close_memberships(*args: Any, **kwargs: Any) -> list[RelatedGroupMember]:
|
||
raise RuntimeError("сбой БД при закрытии членств")
|
||
|
||
monkeypatch.setattr("src.repositories.directories.close_memberships", broken_close_memberships)
|
||
transport = httpx.ASGITransport(app=app, raise_app_exceptions=False)
|
||
async with httpx.AsyncClient(
|
||
transport=transport, base_url=authorized.base_url, headers=authorized.headers
|
||
) as client:
|
||
response = await client.post(
|
||
f"{ENTITIES}/{entity_id}/move",
|
||
json={"company_group_id": str(new_group.id), "effective_date": _today().isoformat()},
|
||
)
|
||
assert response.status_code == HTTP_SERVER_ERROR
|
||
|
||
# SAVEPOINT запроса откатился — проверяем в той же сессии (populate_existing: читать из БД, не из identity map)
|
||
reloaded = (
|
||
await seeded.execute(
|
||
select(LegalEntity).where(LegalEntity.id == entity_id).execution_options(populate_existing=True)
|
||
)
|
||
).scalar_one()
|
||
assert reloaded.company_group_id == old_group_id
|
||
reloaded_member = (
|
||
await seeded.execute(
|
||
select(RelatedGroupMember)
|
||
.where(RelatedGroupMember.id == member_id)
|
||
.execution_options(populate_existing=True)
|
||
)
|
||
).scalar_one()
|
||
assert reloaded_member.is_current is True
|
||
assert reloaded_member.valid_to is None
|
||
assert await _history_count(seeded) == 0
|
||
assert not await _events(seeded, AuditAction.LEGAL_ENTITY_CHANGE, AuditAction.MEMBERSHIP_CHANGE)
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_branch_patch_with_unchanged_foreign_office_is_not_rejected(
|
||
auth_client: AuthClientFactory, seeded: AsyncSession
|
||
) -> None:
|
||
"""I-1 (Review Focus 1): ЮЛ в ГК своего филиала с чужим `regional_office_id`; форма шлёт тот же филиал — не 422."""
|
||
rostov, krasnodar = await _office(seeded, ROSTOV), await _office(seeded, KRASNODAR)
|
||
group = await make_company_group(seeded, name="ГК Ростов", office=rostov)
|
||
entity = await make_legal_entity(seeded, name="ООО Старое", group=group, office=krasnodar) # как создал бы ГО
|
||
client = await auth_client("ka_rf@armka.local")
|
||
|
||
same_office_only = await client.patch(f"{ENTITIES}/{entity.id}", json={"regional_office_id": str(krasnodar.id)})
|
||
assert same_office_only.status_code == HTTP_OK, same_office_only.text
|
||
assert same_office_only.json()["regional_office_id"] == str(krasnodar.id)
|
||
assert await _history_count(seeded) == 0
|
||
assert not await _events(seeded, AuditAction.LEGAL_ENTITY_CHANGE)
|
||
|
||
renamed = await client.patch(
|
||
f"{ENTITIES}/{entity.id}", json={"name": "ООО Новое", "regional_office_id": str(krasnodar.id)}
|
||
)
|
||
assert renamed.status_code == HTTP_OK, renamed.text
|
||
assert (renamed.json()["name"], renamed.json()["regional_office_id"]) == ("ООО Новое", str(krasnodar.id))
|
||
assert await _history_count(seeded) == 1
|
||
(event,) = await _events(seeded, AuditAction.LEGAL_ENTITY_CHANGE)
|
||
assert (event.before_value, event.after_value) == ({"name": "ООО Старое"}, {"name": "ООО Новое"})
|
||
|
||
to_foreign = await client.patch(f"{ENTITIES}/{entity.id}", json={"regional_office_id": None})
|
||
assert to_foreign.status_code == HTTP_OK # явный null у филиала — свой филиал (это изменение)
|
||
assert to_foreign.json()["regional_office_id"] == str(rostov.id)
|