arm-core/api/tests/test_api_counterparty_mapping.py

"""Кандидаты и ручное сопоставление контрагента (spec consolidated-osv «Кандидаты для сопоставления», «Ручное
сопоставление контрагента»; audit-trail `OSV_COUNTERPARTY_MAP`, `LEGAL_ENTITY_CHANGE`; design D-10, D-11).

Мир теста — версия РБ 2024 Q3 (периметр {АО Русский бисквит, ООО «Пекарня Счастливочная»}, ЧКФ — в ГК вне
периметра); у РБ нарастающие ОСВ 60/62/66/76 за Q3 со строкой «Пекарня Счастливочная ООО» без ИНН. Тёзка
«АО Пекарня Счастливочная» из ГК Краснодарского филиала делает резолв по нормализованному имени неоднозначным, поэтому
строки РБ до ручного сопоставления не сопоставлены (как на фикстурах, сценарий «Признаки на фикстурах РБ»).
"""

import asyncio
import uuid
from dataclasses import dataclass, field
from typing import Any

import pytest
from sqlalchemy import func, select, text
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from src.api.dependencies.audit import RequestAudit
from src.common.clock import utc_now
from src.domain.aliases import normalize_alias
from src.domain.policy import VisibilityScope
from src.models.access import AppUser
from src.models.audit_ex import AuditEvent
from src.models.consolidation import ConsolidationDocument, ConsolidationVersion
from src.models.directories import CompanyGroup, LegalEntity, LegalEntityAlias, RegionalOffice
from src.models.enums import (
    AuditAction,
    AuditSection,
    ConsolidationStatus,
    CounterpartyMatchKind,
    OsvDataType,
    PeriodQuarter,
    RelatedGroupKind,
)
from src.models.files import Document
from src.models.osv import OsvEntry
from src.repositories import directories as directories_repo
from src.schemas.consolidations import MapCounterpartyIn
from src.services.audit import AuditService
from src.services.counterparty_mapping import CounterpartyMappingService
from tests.conftest import AuthClientFactory
from tests.factories import (
    make_company_group,
    make_confirmed_osv,
    make_consolidation,
    make_legal_entity,
    make_lock,
    make_membership,
    make_related_group,
)

HTTP_OK = 200
HTTP_FORBIDDEN = 403
HTTP_NOT_FOUND = 404
HTTP_CONFLICT = 409
HTTP_UNPROCESSABLE = 422
KA_RF = "ka_rf@armka.local"
RKA = "rka@armka.local"
CURATOR = "curator_go@armka.local"
EXPERT = "expert@armka.local"
FOREIGN = "client_manager@armka.local"  # Краснодарский филиал
ROSTOV = "Ростовский региональный филиал"
KRASNODAR = "Краснодарский региональный филиал"
YEAR = 2024
Q1, Q2, Q3 = PeriodQuarter.Q1, PeriodQuarter.Q2, PeriodQuarter.Q3
NAME = "Пекарня Счастливочная ООО"
ALIEN_INN = "7700000000"
MAX_CANDIDATES = 100
LOCK_WAIT_SECONDS = 10.0
ADMIN = "admin@armka.local"
ENTITIES = "/directories/legal-entities"


@dataclass
class World:
    """Версия РБ 2024 Q3 и её участники; `docs` — документы по счёту, `entries` — строки по ключу (счёт, номер)."""

    version: ConsolidationVersion
    group: CompanyGroup
    rb: LegalEntity
    schastl: LegalEntity
    chkf: LegalEntity
    namesake: LegalEntity
    docs: dict[str, Document] = field(default_factory=dict)
    entries: dict[tuple[str, int], uuid.UUID] = field(default_factory=dict)
    gsk_id: uuid.UUID | None = None
    gsz_id: uuid.UUID | None = None


async def _user(session: AsyncSession, email: str) -> AppUser:
    return (await session.execute(select(AppUser).where(AppUser.email == email))).scalar_one()


async def _office(session: AsyncSession, name: str) -> RegionalOffice:
    return (await session.execute(select(RegionalOffice).where(RegionalOffice.name == name))).scalar_one()


async def _bind(session: AsyncSession, version: ConsolidationVersion, *documents: Document) -> None:
    for document in documents:
        session.add(ConsolidationDocument(consolidation_version_id=version.id, document_id=document.id))
    await session.flush()


async def _entries_of(session: AsyncSession, document: Document) -> list[uuid.UUID]:
    stmt = select(OsvEntry.id).where(OsvEntry.document_id == document.id).order_by(OsvEntry.row_number)
    return list((await session.execute(stmt)).scalars())


async def _world(  # noqa: PLR0913 - тестовая фабрика: статус, владелец блокировки и филиал версии
    session: AsyncSession,
    *,
    status: ConsolidationStatus = ConsolidationStatus.DATA_CORRECTION,
    locked_by: str | None = KA_RF,
    office_name: str = ROSTOV,
) -> World:
    group = await make_company_group(session, name="ГК Русский бисквит", office=await _office(session, office_name))
    rb = await make_legal_entity(session, name="АО Русский бисквит", group=group)
    schastl = await make_legal_entity(session, name="ООО «Пекарня Счастливочная»", group=group)
    chkf = await make_legal_entity(session, name="ООО ЧКФ", group=group)
    other_group = await make_company_group(session, name="ГК Кубань", office=await _office(session, KRASNODAR))
    namesake = await make_legal_entity(session, name="АО Пекарня Счастливочная", group=other_group)
    gsk = await make_related_group(session, group=group, name="ГСК Бисквит")
    await make_membership(session, related_group=gsk, entity=rb)
    await make_membership(session, related_group=gsk, entity=schastl)
    gsz = await make_related_group(session, group=group, kind=RelatedGroupKind.gsz, name="ГСЗ Бисквит")
    await make_membership(session, related_group=gsz, entity=chkf)
    _, version = await make_consolidation(
        session, group=group, year=YEAR, quarter=Q3, members=[rb, schastl],
        initiator=await _user(session, KA_RF), status=status,
    )  # fmt: skip
    if locked_by is not None:
        await make_lock(session, version=version, user=await _user(session, locked_by))
    world = World(
        version=version, group=group, rb=rb, schastl=schastl, chkf=chkf, namesake=namesake, gsk_id=gsk.id,
        gsz_id=gsz.id,
    )  # fmt: skip
    rows: dict[str, list[dict[str, Any]]] = {
        "60": [
            {"name": NAME, "closing_debit": 100},
            {"name": "ООО ЧКФ", "counterparty_id": chkf.id, "counterparty_match_kind": CounterpartyMatchKind.MANUAL},
            {"name": NAME, "data_type": OsvDataType.NU, "closing_debit": 1},  # налоговые данные — не в своде
        ],
        "62": [{"name": NAME, "closing_credit": 20}],
        "66": [{"name": f"  {NAME} ", "closing_credit": 30}],  # те же имена после strip(), в т.ч. NBSP
        "76": [{"name": NAME, "inn": ALIEN_INN, "closing_debit": 40}],  # ИНН противоречит выбранному ЮЛ
    }
    for account, account_rows in rows.items():
        world.docs[account] = await make_confirmed_osv(
            session, entity=rb, account=account, year=YEAR, quarter=Q3, is_cumulative=True, rows=account_rows
        )
    world.docs["schastl-60"] = await make_confirmed_osv(
        session, entity=schastl, account="60", year=YEAR, quarter=Q3, is_cumulative=True,
        rows=[{"name": "Русский бисквит", "inn": rb.inn, "counterparty_id": rb.id,
               "counterparty_match_kind": CounterpartyMatchKind.INN}],
    )  # fmt: skip
    await _bind(session, version, *world.docs.values())
    # подтверждённый, но НЕ привязанный к версии документ РБ с тем же именем контрагента
    world.docs["unbound-58"] = await make_confirmed_osv(
        session, entity=rb, account="58", year=YEAR, quarter=Q3, is_cumulative=True, rows=[{"name": NAME}]
    )
    for account, document in world.docs.items():
        for number, entry_id in enumerate(await _entries_of(session, document), start=1):
            world.entries[(account, number)] = entry_id
    return world


def _map_url(world: World, entry_id: uuid.UUID, version_id: uuid.UUID | None = None) -> str:
    return f"/consolidations/versions/{version_id or world.version.id}/osv/entries/{entry_id}/counterparty"


def _candidates_url(version_id: uuid.UUID) -> str:
    return f"/consolidations/versions/{version_id}/counterparty-candidates"


async def _entry(session: AsyncSession, entry_id: uuid.UUID) -> OsvEntry:
    stmt = select(OsvEntry).where(OsvEntry.id == entry_id).execution_options(populate_existing=True)
    return (await session.execute(stmt)).scalar_one()


async def _aliases(session: AsyncSession) -> list[tuple[uuid.UUID, str, str]]:
    stmt = select(LegalEntityAlias).order_by(LegalEntityAlias.alias).execution_options(populate_existing=True)
    return [(a.legal_entity_id, a.alias, a.normalized_alias) for a in (await session.execute(stmt)).scalars()]


async def _events(session: AsyncSession, action: AuditAction) -> list[AuditEvent]:
    stmt = select(AuditEvent).where(AuditEvent.action == action).order_by(AuditEvent.occurred_at, AuditEvent.id)
    return list((await session.execute(stmt)).scalars())


async def _summary(client: Any, world: World) -> dict[str, Any]:
    response = await client.get(f"/consolidations/versions/{world.version.id}/osv/summary")
    assert response.status_code == HTTP_OK, response.text
    body: dict[str, Any] = response.json()
    return body


async def _assert_kind_iff_id(session: AsyncSession) -> None:
    """Инвариант v5.16: `counterparty_match_kind IS NULL` ⇔ `counterparty_id IS NULL` во всех строках."""
    broken = (
        await session.execute(
            select(func.count())
            .select_from(OsvEntry)
            .where((OsvEntry.counterparty_id.is_(None)) != (OsvEntry.counterparty_match_kind.is_(None)))
        )
    ).scalar_one()
    assert broken == 0


async def _primary(session: AsyncSession) -> dict[uuid.UUID, tuple[str | None, str | None]]:
    rows = await session.execute(select(OsvEntry.id, OsvEntry.counterparty_name, OsvEntry.counterparty_inn))
    return {row[0]: (row[1], row[2]) for row in rows}


# --- кандидаты -----------------------------------------------------------------------------------------------------


@pytest.mark.integration
async def test_candidates_recommend_schastlivochnaya_first(
    auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
    """Сценарий «Рекомендуемая Счастливочная»: `scope=group` без `q`, `entry_id` — строка РБ «Пекарня Счастливочная
    ООО»."""
    world = await _world(seeded)
    client = await auth_client(KA_RF)

    response = await client.get(_candidates_url(world.version.id), params={"entry_id": str(world.entries[("60", 1)])})

    assert response.status_code == HTTP_OK, response.text
    items = response.json()
    assert items[0]["id"] == str(world.schastl.id)
    assert (items[0]["recommended"], items[0]["counterparty_type"]) == (True, "GSK")
    assert items[0]["inn"] == world.schastl.inn and items[0]["name"] == "ООО «Пекарня Счастливочная»"
    assert items[0]["company_group"] == {"id": str(world.group.id), "name": "ГК Русский бисквит"}
    assert items[0]["gsk"] == {"id": str(world.gsk_id), "name": "ГСК Бисквит"} and items[0]["gsz"] is None
    by_id = {item["id"]: item for item in items}
    chkf = by_id[str(world.chkf.id)]
    assert (chkf["recommended"], chkf["counterparty_type"]) == (False, "GK")
    assert chkf["gsz"] == {"id": str(world.gsz_id), "name": "ГСЗ Бисквит"} and chkf["gsk"] is None
    assert (by_id[str(world.rb.id)]["recommended"], by_id[str(world.rb.id)]["counterparty_type"]) == (False, "GSK")
    assert str(world.namesake.id) not in by_id  # чужая ГК — только в scope=all
    assert [item["name"] for item in items[1:]] == sorted(item["name"] for item in items[1:])


@pytest.mark.integration
async def test_candidates_without_entry_have_no_recommendation(
    auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
    """Без `entry_id` и с `entry_id` строки непривязанного документа рекомендаций нет; порядок — по имени."""
    world = await _world(seeded)
    client = await auth_client(KA_RF)

    for params in ({}, {"entry_id": str(world.entries[("unbound-58", 1)])}, {"entry_id": str(uuid.uuid4())}):
        response = await client.get(_candidates_url(world.version.id), params=params)
        assert response.status_code == HTTP_OK, response.text
        items = response.json()
        assert {item["id"] for item in items} == {str(world.rb.id), str(world.schastl.id), str(world.chkf.id)}
        assert not any(item["recommended"] for item in items)
        assert [item["name"] for item in items] == sorted(item["name"] for item in items)


@pytest.mark.integration
@pytest.mark.parametrize("params", [{"scope": "all"}, {"scope": "all", "q": "Сч"}, {"scope": "all", "q": "  Сч  "}])
async def test_candidates_all_requires_query_of_three_chars(
    auth_client: AuthClientFactory, seeded: AsyncSession, params: dict[str, str]
) -> None:
    world = await _world(seeded)
    client = await auth_client(KA_RF)

    response = await client.get(_candidates_url(world.version.id), params=params)

    assert response.status_code == HTTP_UNPROCESSABLE, response.text
    assert response.json()["detail"]["code"] == "QUERY_TOO_SHORT"


@pytest.mark.integration
async def test_candidates_all_finds_entity_of_other_branch(
    auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
    """`scope=all` — весь справочник без policy: ЮЛ ГК Краснодарского филиала находится; признак `EXTERNAL`."""
    world = await _world(seeded)
    client = await auth_client(KA_RF)

    response = await client.get(
        _candidates_url(world.version.id),
        params={"scope": "all", "q": "счастлив", "entry_id": str(world.entries[("60", 1)])},
    )

    assert response.status_code == HTTP_OK, response.text
    by_id = {item["id"]: item for item in response.json()}
    assert set(by_id) == {str(world.schastl.id), str(world.namesake.id)}
    namesake = by_id[str(world.namesake.id)]
    assert (namesake["counterparty_type"], namesake["recommended"]) == ("EXTERNAL", True)
    assert namesake["company_group"]["name"] == "ГК Кубань"
    assert (await client.get(_candidates_url(world.version.id), params={"scope": "bogus"})).status_code == 422  # noqa: PLR2004


@pytest.mark.integration
async def test_candidates_search_by_inn_prefix_alias_and_literal_wildcards(
    auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
    world = await _world(seeded)
    seeded.add(LegalEntityAlias(legal_entity_id=world.chkf.id, alias="Вологодская выпечка",
                                normalized_alias="вологодская выпечка"))  # fmt: skip
    await seeded.flush()
    client = await auth_client(KA_RF)

    async def ids(**params: str) -> set[str]:
        response = await client.get(_candidates_url(world.version.id), params=params)
        assert response.status_code == HTTP_OK, response.text
        return {item["id"] for item in response.json()}

    assert await ids(q=world.schastl.inn[:5]) == {str(world.schastl.id)}  # префикс ИНН
    assert await ids(q=world.schastl.inn[2:]) == set()  # ИНН — только префикс
    assert await ids(q="ВОЛОГОДСК") == {str(world.chkf.id)}  # подстрока алиаса без регистра
    assert await ids(q="бисквит") == {str(world.rb.id)}  # подстрока имени без регистра
    assert await ids(q="%") == set() and await ids(q="_") == set()  # шаблоны LIKE экранированы
    assert await ids(scope="all", q=f"{world.namesake.inn[:4]}") >= {str(world.namesake.id)}


@pytest.mark.integration
async def test_candidates_limit_keeps_recommended_first(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    world = await _world(seeded)
    for n in range(MAX_CANDIDATES + 5):
        await make_legal_entity(seeded, name=f"АА Ранний {n:03d}", group=world.group)
    client = await auth_client(KA_RF)

    response = await client.get(_candidates_url(world.version.id), params={"entry_id": str(world.entries[("60", 1)])})

    items = response.json()
    assert len(items) == MAX_CANDIDATES
    assert items[0]["id"] == str(world.schastl.id) and items[0]["recommended"] is True


@pytest.mark.integration
async def test_candidates_rights_and_visibility(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    world = await _world(seeded)
    curator = await auth_client(CURATOR)
    foreign = await auth_client(FOREIGN)
    admin = await auth_client("admin@armka.local")

    assert (await curator.get(_candidates_url(world.version.id))).status_code == HTTP_OK
    hidden = await foreign.get(_candidates_url(world.version.id))
    missing = await foreign.get(_candidates_url(uuid.uuid4()))
    assert hidden.status_code == missing.status_code == HTTP_NOT_FOUND
    assert hidden.json() == missing.json()
    assert (await admin.get(_candidates_url(world.version.id))).status_code == HTTP_FORBIDDEN


# --- ручное сопоставление ------------------------------------------------------------------------------------------


@pytest.mark.integration
async def test_mapping_from_slice_plan_propagates_and_creates_alias(
    auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
    """Сценарий «Сопоставление из плана среза» и «ИНН противоречит»."""
    world = await _world(seeded)
    client = await auth_client(KA_RF)
    before_summary = await _summary(client, world)
    primary_before = await _primary(seeded)
    target = world.entries[("60", 1)]

    response = await client.post(_map_url(world, target), json={"legal_entity_id": str(world.schastl.id)})

    assert response.status_code == HTTP_OK, response.text
    body = response.json()
    assert (body["propagated"], body["alias_created"], body["alias_moved_from"]) == (2, True, None)
    row = body["row"]
    assert row["entry_ids"] == [str(target)]
    assert (row["counterparty_id"], row["counterparty_type"], row["match_kind"], row["match_stored"]) == (
        str(world.schastl.id), "GSK", "MANUAL", True)  # fmt: skip
    assert row["counterparty_entity_name"] == "ООО «Пекарня Счастливочная»"
    stored = await _entry(seeded, target)
    assert (stored.counterparty_id, stored.counterparty_match_kind) == (world.schastl.id, CounterpartyMatchKind.MANUAL)
    assert stored.modified_by is None and stored.modified_at is None and stored.is_modified is False
    for key in (("62", 1), ("66", 1)):
        entry = await _entry(seeded, world.entries[key])
        assert (entry.counterparty_id, entry.counterparty_match_kind) == (
            world.schastl.id, CounterpartyMatchKind.PROPAGATED), key  # fmt: skip
    for key in (("76", 1), ("60", 3), ("unbound-58", 1)):  # ИНН противоречит; NU; документ не привязан
        entry = await _entry(seeded, world.entries[key])
        assert (entry.counterparty_id, entry.counterparty_match_kind) == (None, None), key
    chkf_row = await _entry(seeded, world.entries[("60", 2)])
    assert (chkf_row.counterparty_id, chkf_row.counterparty_match_kind) == (world.chkf.id, CounterpartyMatchKind.MANUAL)
    assert await _aliases(seeded) == [(world.schastl.id, NAME, "пекарня счастливочная")]
    alias = (await seeded.execute(select(LegalEntityAlias))).scalar_one()
    assert alias.created_by == (await _user(seeded, KA_RF)).id and alias.created_at is not None
    after_summary = await _summary(client, world)
    # строка 76 с чужим ИНН в «Осталось N» не входит и до сопоставления: ИНН из файла есть (отмена П-1, arm-core #74);
    # на чтении она резолвится по алиасу (ИНН нет в справочнике и не блокирует поиск по имени, алиас приоритетнее
    # ЮЛ-тёзки, OQ-128)
    assert after_summary["unmapped_total"] == before_summary["unmapped_total"] - (1 + body["propagated"])
    assert after_summary["unmapped_total"] == 0
    page = (await client.get(f"/consolidations/versions/{world.version.id}/osv?limit=100")).json()
    alien = next(r for r in page["items"] if r["account_code"] == "76" and r["counterparty_name"] == NAME)
    assert (alien["counterparty_id"], alien["match_kind"], alien["match_stored"]) == (
        str(world.schastl.id), "ALIAS", False)  # fmt: skip
    assert await _primary(seeded) == primary_before  # counterparty_name/inn не тронуты
    await _assert_kind_iff_id(seeded)


@pytest.mark.integration
async def test_mapping_writes_one_map_event_and_alias_event(
    auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
    world = await _world(seeded)
    client = await auth_client(KA_RF)
    target = world.entries[("60", 1)]

    response = await client.post(_map_url(world, target), json={"legal_entity_id": str(world.schastl.id)})

    assert response.status_code == HTTP_OK, response.text
    user_id = (await _user(seeded, KA_RF)).id
    (event,) = await _events(seeded, AuditAction.OSV_COUNTERPARTY_MAP)
    assert (event.section, event.object_type, event.object_id) == (AuditSection.CONSOLIDATION, "osv_entry", target)
    assert (event.consolidation_version_id, event.user_id) == (world.version.id, user_id)
    assert event.before_value == {"counterparty_id": None, "match_kind": None}
    assert event.after_value == {
        "counterparty_id": str(world.schastl.id),
        "match_kind": "MANUAL",
        "propagated_entry_ids": sorted([str(world.entries[("62", 1)]), str(world.entries[("66", 1)])]),
        "alias_created": True,
        "alias_moved_from": None,
    }
    (alias_event,) = await _events(seeded, AuditAction.LEGAL_ENTITY_CHANGE)
    alias = (await seeded.execute(select(LegalEntityAlias))).scalar_one()
    assert (alias_event.section, alias_event.object_type, alias_event.object_id) == (
        AuditSection.DIRECTORIES, "legal_entity_alias", alias.id)  # fmt: skip
    assert alias_event.before_value is None
    assert alias_event.after_value == {
        "legal_entity_id": str(world.schastl.id), "alias": NAME, "normalized_alias": "пекарня счастливочная",
    }  # fmt: skip
    assert alias_event.consolidation_version_id == world.version.id and alias_event.user_id == user_id


@pytest.mark.integration
async def test_alias_taken_by_other_entity_is_conflict_and_changes_nothing(
    auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
    """Сценарий «Алиас занят»: 409 `ALIAS_TAKEN` с `owners`; строки, алиасы и журнал не изменились."""
    world = await _world(seeded)
    firm = await make_legal_entity(seeded, name="Фирма Эф")
    seeded.add(LegalEntityAlias(legal_entity_id=firm.id, alias="«Пекарня Счастливочная»",
                                normalized_alias="пекарня счастливочная"))  # fmt: skip
    await seeded.flush()
    aliases_before = await _aliases(seeded)
    client = await auth_client(KA_RF)

    response = await client.post(
        _map_url(world, world.entries[("60", 1)]), json={"legal_entity_id": str(world.schastl.id)}
    )

    assert response.status_code == HTTP_CONFLICT, response.text
    detail = response.json()["detail"]
    assert detail["code"] == "ALIAS_TAKEN"
    assert detail["owners"] == [{"legal_entity_id": str(firm.id), "name": "Фирма Эф"}]
    assert await _aliases(seeded) == aliases_before
    for key in (("60", 1), ("62", 1), ("66", 1)):
        entry = await _entry(seeded, world.entries[key])
        assert (entry.counterparty_id, entry.counterparty_match_kind) == (None, None)
    assert await _events(seeded, AuditAction.OSV_COUNTERPARTY_MAP) == []
    assert await _events(seeded, AuditAction.LEGAL_ENTITY_CHANGE) == []


@pytest.mark.integration
async def test_move_alias_deletes_foreign_alias_and_creates_own(
    auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
    """Сценарий «Перенос алиаса»: два `LEGAL_ENTITY_CHANGE` и один `OSV_COUNTERPARTY_MAP`; `DIRECTORY_DELETE` не нужен
    (у `ka_rf` его нет — проверено seed-ом ролей)."""
    world = await _world(seeded)
    firm = await make_legal_entity(seeded, name="Фирма Эф")
    old = LegalEntityAlias(legal_entity_id=firm.id, alias="«Пекарня Счастливочная»",
                           normalized_alias="пекарня счастливочная")  # fmt: skip
    seeded.add(old)
    await seeded.flush()
    old_id = old.id
    client = await auth_client(KA_RF)

    response = await client.post(
        _map_url(world, world.entries[("60", 1)]),
        json={"legal_entity_id": str(world.schastl.id), "move_alias": True},
    )

    assert response.status_code == HTTP_OK, response.text
    body = response.json()
    assert (body["alias_created"], body["alias_moved_from"], body["propagated"]) == (True, str(firm.id), 2)
    assert await _aliases(seeded) == [(world.schastl.id, NAME, "пекарня счастливочная")]
    removed, created = await _events(seeded, AuditAction.LEGAL_ENTITY_CHANGE)
    assert (removed.object_type, removed.object_id, removed.after_value) == ("legal_entity_alias", old_id, None)
    assert removed.before_value == {
        "legal_entity_id": str(firm.id), "alias": "«Пекарня Счастливочная»",
        "normalized_alias": "пекарня счастливочная",
    }  # fmt: skip
    assert created.before_value is None and created.after_value["legal_entity_id"] == str(world.schastl.id)
    (event,) = await _events(seeded, AuditAction.OSV_COUNTERPARTY_MAP)
    assert event.after_value["alias_moved_from"] == str(firm.id)


@pytest.mark.integration
async def test_inn_equal_after_whitespace_is_propagated(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    """Автоприменение: ИНН строки пустой или равный ИНН выбранного (без пробелов) — получает, иной — нет."""
    world = await _world(seeded)
    spaced = f" {world.schastl.inn[:4]} {world.schastl.inn[4:]} "
    extra = await make_confirmed_osv(
        seeded, entity=world.rb, account="68", year=YEAR, quarter=Q3, is_cumulative=True,
        rows=[{"name": NAME, "inn": spaced}, {"name": NAME, "inn": "   "}, {"name": NAME, "inn": ALIEN_INN}],
    )  # fmt: skip
    await _bind(seeded, world.version, extra)
    same_inn, blank_inn, alien_inn = await _entries_of(seeded, extra)
    client = await auth_client(KA_RF)

    response = await client.post(
        _map_url(world, world.entries[("60", 1)]), json={"legal_entity_id": str(world.schastl.id)}
    )

    assert response.status_code == HTTP_OK, response.text
    assert response.json()["propagated"] == 4  # noqa: PLR2004 - 62, 66 и две строки 68
    assert (await _entry(seeded, same_inn)).counterparty_match_kind == CounterpartyMatchKind.PROPAGATED
    assert (await _entry(seeded, blank_inn)).counterparty_match_kind == CounterpartyMatchKind.PROPAGATED
    assert (await _entry(seeded, alien_inn)).counterparty_id is None
    assert (await _entry(seeded, same_inn)).counterparty_inn == spaced


@pytest.mark.integration
async def test_remap_linked_row_keeps_previous_in_before(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    """Перевязка уже связанной строки: ошибочно к ЧКФ, затем к Счастливочной (алиас переносится с ЧКФ)."""
    world = await _world(seeded)
    client = await auth_client(KA_RF)
    target = world.entries[("60", 1)]
    first = await client.post(_map_url(world, target), json={"legal_entity_id": str(world.chkf.id)})
    assert first.status_code == HTTP_OK, first.text
    assert first.json()["row"]["counterparty_type"] == "GK"

    taken = await client.post(_map_url(world, target), json={"legal_entity_id": str(world.schastl.id)})
    assert taken.status_code == HTTP_CONFLICT and taken.json()["detail"]["code"] == "ALIAS_TAKEN"
    second = await client.post(
        _map_url(world, target), json={"legal_entity_id": str(world.schastl.id), "move_alias": True}
    )

    assert second.status_code == HTTP_OK, second.text
    body = second.json()
    assert (body["propagated"], body["alias_created"], body["alias_moved_from"]) == (0, True, str(world.chkf.id))
    assert body["row"]["counterparty_id"] == str(world.schastl.id) and body["row"]["match_kind"] == "MANUAL"
    _, remap = await _events(seeded, AuditAction.OSV_COUNTERPARTY_MAP)
    assert remap.before_value == {"counterparty_id": str(world.chkf.id), "match_kind": "MANUAL"}
    assert remap.after_value["propagated_entry_ids"] == []
    # ранее автоприменённые строки остаются за ЧКФ: автоприменение трогает только пустые `counterparty_id`
    assert (await _entry(seeded, world.entries[("62", 1)])).counterparty_id == world.chkf.id
    await _assert_kind_iff_id(seeded)


@pytest.mark.integration
async def test_second_mapping_to_same_entity_reuses_alias(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    """Алиас у выбранного ЮЛ уже есть → `alias_created = false`, новых `LEGAL_ENTITY_CHANGE` нет."""
    world = await _world(seeded)
    client = await auth_client(KA_RF)
    first = await client.post(
        _map_url(world, world.entries[("60", 1)]), json={"legal_entity_id": str(world.schastl.id)}
    )
    assert first.status_code == HTTP_OK, first.text

    second = await client.post(
        _map_url(world, world.entries[("76", 1)]), json={"legal_entity_id": str(world.schastl.id)}
    )

    assert second.status_code == HTTP_OK, second.text
    assert (second.json()["alias_created"], second.json()["propagated"]) == (False, 0)
    assert len(await _aliases(seeded)) == 1
    assert len(await _events(seeded, AuditAction.LEGAL_ENTITY_CHANGE)) == 1
    assert (await _summary(client, world))["unmapped_total"] == 0


@pytest.mark.integration
async def test_name_of_legal_forms_only_links_without_alias(
    auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
    """Review Focus 3: имя из одних ОПФ — `normalize_alias` пуст: алиас не создаётся, строка связывается,
    автоприменение по `strip()` имени работает (и не цепляет другие имена, нормализующиеся в пустое)."""
    world = await _world(seeded)
    document = await make_confirmed_osv(
        seeded, entity=world.rb, account="68", year=YEAR, quarter=Q3, is_cumulative=True,
        rows=[{"name": "ООО"}, {"name": " ООО "}, {"name": "ООО."}, {"name": "АО"}],
    )  # fmt: skip
    await _bind(seeded, world.version, document)
    target, same, punct, other = await _entries_of(seeded, document)
    client = await auth_client(KA_RF)

    response = await client.post(_map_url(world, target), json={"legal_entity_id": str(world.chkf.id)})

    assert response.status_code == HTTP_OK, response.text
    body = response.json()
    assert (body["alias_created"], body["alias_moved_from"], body["propagated"]) == (False, None, 1)
    assert body["row"]["match_kind"] == "MANUAL" and body["row"]["counterparty_id"] == str(world.chkf.id)
    assert (await _entry(seeded, same)).counterparty_match_kind == CounterpartyMatchKind.PROPAGATED
    assert (await _entry(seeded, punct)).counterparty_id is None
    assert (await _entry(seeded, other)).counterparty_id is None
    assert await _aliases(seeded) == []
    assert await _events(seeded, AuditAction.LEGAL_ENTITY_CHANGE) == []
    (event,) = await _events(seeded, AuditAction.OSV_COUNTERPARTY_MAP)
    assert event.after_value["alias_created"] is False and event.after_value["propagated_entry_ids"] == [str(same)]


@pytest.mark.integration
async def test_mapping_in_one_version_is_visible_in_other(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    """Review Focus 1: консолидации ГК за Q2 и Q3; квартальная ОСВ Q1 привязана к обеим. Сопоставление в версии Q2
    меняет `osv_entry` — в версии Q3 строка видна с `MANUAL`; строка того же имени из ОСВ Q3 (привязана только к
    версии Q3) автоприменением версии Q2 не затронута."""
    world = await _world(seeded)
    owner = await _user(seeded, KA_RF)
    _, q2 = await make_consolidation(
        seeded, group=world.group, year=YEAR, quarter=Q2, members=[world.rb, world.schastl], initiator=owner,
        status=ConsolidationStatus.DATA_CORRECTION,
    )  # fmt: skip
    await make_lock(seeded, version=q2, user=owner)
    quarterly = {
        quarter: await make_confirmed_osv(
            seeded,
            entity=world.rb,
            account="51",
            year=YEAR,
            quarter=quarter,
            is_cumulative=False,
            rows=[{"name": "Банк Альфа"}] if quarter is not Q2 else [{"name": "Прочий"}],
        )  # fmt: skip
        for quarter in (Q1, Q2, Q3)
    }
    await _bind(seeded, q2, quarterly[Q1], quarterly[Q2])
    await _bind(seeded, world.version, quarterly[Q1], quarterly[Q2], quarterly[Q3])
    (shared_entry,) = await _entries_of(seeded, quarterly[Q1])
    (q3_entry,) = await _entries_of(seeded, quarterly[Q3])
    client = await auth_client(KA_RF)

    response = await client.post(_map_url(world, shared_entry, q2.id), json={"legal_entity_id": str(world.chkf.id)})

    assert response.status_code == HTTP_OK, response.text
    assert response.json()["propagated"] == 0
    assert (await _entry(seeded, q3_entry)).counterparty_id is None
    page = await client.get(f"/consolidations/versions/{world.version.id}/osv", params={"q": "Банк Альфа"})
    (row,) = page.json()["items"]
    assert set(row["entry_ids"]) == {str(shared_entry), str(q3_entry)}
    assert (row["counterparty_id"], row["match_kind"], row["match_stored"]) == (str(world.chkf.id), "MANUAL", True)
    (event,) = await _events(seeded, AuditAction.OSV_COUNTERPARTY_MAP)
    assert event.consolidation_version_id == q2.id


@pytest.mark.integration
async def test_mapping_of_entry_outside_summary_returns_null_row(
    auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
    """Строка привязанного документа пары с пробелом в своде не участвует: сопоставление пишется, `row = null`."""
    world = await _world(seeded)
    q1 = await make_confirmed_osv(
        seeded, entity=world.rb, account="51", year=YEAR, quarter=Q1, is_cumulative=False, rows=[{"name": "Банк Бета"}]
    )
    await _bind(seeded, world.version, q1)  # за Q2 и Q3 по счёту 51 ничего нет — пробел
    (entry_id,) = await _entries_of(seeded, q1)
    client = await auth_client(KA_RF)

    response = await client.post(_map_url(world, entry_id), json={"legal_entity_id": str(world.chkf.id)})

    assert response.status_code == HTTP_OK, response.text
    assert response.json()["row"] is None
    assert (await _entry(seeded, entry_id)).counterparty_match_kind == CounterpartyMatchKind.MANUAL


@pytest.mark.integration
async def test_entry_not_in_version_is_not_found(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    """Строка непривязанного документа, строка NU и несуществующая — один 404 (`ENTRY_NOT_FOUND`)."""
    world = await _world(seeded)
    client = await auth_client(KA_RF)
    body = {"legal_entity_id": str(world.schastl.id)}

    responses = [
        await client.post(_map_url(world, entry_id), json=body)
        for entry_id in (world.entries[("unbound-58", 1)], world.entries[("60", 3)], uuid.uuid4())
    ]

    assert {response.status_code for response in responses} == {HTTP_NOT_FOUND}
    assert responses[0].json() == responses[1].json() == responses[2].json()
    assert responses[0].json()["detail"]["code"] == "NOT_FOUND"
    assert "строка" in responses[0].json()["detail"]["message"].lower()
    assert (await _entry(seeded, world.entries[("unbound-58", 1)])).counterparty_id is None


@pytest.mark.integration
async def test_legal_entity_must_exist_but_is_not_filtered_by_policy(
    auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
    world = await _world(seeded)
    client = await auth_client(KA_RF)
    target = world.entries[("60", 1)]

    missing = await client.post(_map_url(world, target), json={"legal_entity_id": str(uuid.uuid4())})
    foreign = await client.post(_map_url(world, target), json={"legal_entity_id": str(world.namesake.id)})

    assert missing.status_code == HTTP_NOT_FOUND
    assert missing.json()["detail"] == {"code": "NOT_FOUND", "message": "ЮЛ не найдено"}
    assert foreign.status_code == HTTP_OK, foreign.text  # ЮЛ другого филиала — без policy
    assert foreign.json()["row"]["counterparty_type"] == "EXTERNAL"


@pytest.mark.integration
async def test_mapping_requires_own_lock(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    """Без блокировки — `LOCK_REQUIRED` даже для несуществующей строки (редактируемость проверяется раньше строки)."""
    world = await _world(seeded, locked_by=None)
    client = await auth_client(KA_RF)

    for entry_id in (world.entries[("60", 1)], uuid.uuid4()):
        response = await client.post(_map_url(world, entry_id), json={"legal_entity_id": str(world.schastl.id)})
        assert response.status_code == HTTP_CONFLICT
        assert response.json()["detail"]["code"] == "LOCK_REQUIRED"
    assert (await _entry(seeded, world.entries[("60", 1)])).counterparty_id is None


@pytest.mark.integration
async def test_mapping_with_foreign_lock_is_conflict(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    world = await _world(seeded, locked_by=RKA)
    client = await auth_client(KA_RF)

    response = await client.post(
        _map_url(world, world.entries[("60", 1)]), json={"legal_entity_id": str(world.schastl.id)}
    )

    assert response.status_code == HTTP_CONFLICT
    assert response.json()["detail"]["code"] == "LOCKED_BY_OTHER"


@pytest.mark.integration
async def test_mapping_in_cancelled_version_is_not_editable(
    auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
    world = await _world(seeded, status=ConsolidationStatus.CANCELLED, locked_by=None)
    client = await auth_client(KA_RF)

    response = await client.post(
        _map_url(world, world.entries[("60", 1)]), json={"legal_entity_id": str(world.schastl.id)}
    )

    assert response.status_code == HTTP_CONFLICT
    assert response.json()["detail"]["code"] == "VERSION_NOT_EDITABLE"


@pytest.mark.integration
async def test_mapping_rights_and_visibility(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    world = await _world(seeded, office_name=KRASNODAR, locked_by=None)
    body = {"legal_entity_id": str(world.schastl.id)}
    rostov = await auth_client(KA_RF)

    hidden = await rostov.post(_map_url(world, world.entries[("60", 1)]), json=body)
    missing = await rostov.post(_map_url(world, world.entries[("60", 1)], uuid.uuid4()), json=body)
    assert hidden.status_code == missing.status_code == HTTP_NOT_FOUND
    assert hidden.json() == missing.json()
    for email in (CURATOR, EXPERT):
        client = await auth_client(email)
        assert (await client.post(_map_url(world, world.entries[("60", 1)]), json=body)).status_code == HTTP_FORBIDDEN
    malformed = await rostov.post(_map_url(world, world.entries[("60", 1)]), json={"legal_entity_id": "x"})
    assert malformed.status_code == HTTP_UNPROCESSABLE


# --- конкурентность ------------------------------------------------------------------------------------------------


async def _wait_for_lock_waiter(maker: async_sessionmaker[AsyncSession]) -> None:
    """Ждёт, пока какой-то сеанс БД встанет в ожидание блокировки."""
    deadline = asyncio.get_running_loop().time() + LOCK_WAIT_SECONDS
    while asyncio.get_running_loop().time() < deadline:
        async with maker() as session:
            waiting = (
                await session.execute(
                    text(
                        "SELECT count(*) FROM pg_stat_activity "
                        "WHERE datname = current_database() AND wait_event_type = 'Lock'"
                    )
                )
            ).scalar_one()
        if waiting:
            return
        await asyncio.sleep(0.05)
    raise AssertionError("запрос не встал в ожидание блокировки")


@pytest.mark.integration
async def test_concurrent_mappings_in_two_versions_serialize_on_shared_rows(
    committed_sessionmaker: async_sessionmaker[AsyncSession], committed_client: AuthClientFactory
) -> None:
    """Общий документ двух версий, имя из одних ОПФ (advisory-блокировки алиаса нет): сопоставление в версии Q2
    держит строки; сопоставление соседней строки в версии Q3 ждёт его commit и читает свежее состояние
    (`before` = автоприменённое первым), а автоприменённое первым не перетирает."""
    async with committed_sessionmaker() as setup, setup.begin():
        world = await _world(setup)
        owner = await _user(setup, KA_RF)
        _, q2 = await make_consolidation(
            setup, group=world.group, year=YEAR, quarter=Q2, members=[world.rb, world.schastl], initiator=owner,
            status=ConsolidationStatus.DATA_CORRECTION,
        )  # fmt: skip
        await make_lock(setup, version=q2, user=owner)
        shared = await make_confirmed_osv(
            setup, entity=world.rb, account="51", year=YEAR, quarter=Q1, is_cumulative=False,
            rows=[{"name": "ООО"}, {"name": "ООО"}],
        )  # fmt: skip
        await _bind(setup, q2, shared)
        await _bind(setup, world.version, shared)
        first_entry, second_entry = await _entries_of(setup, shared)
        ids = (q2.id, world.version.id, world.chkf.id, world.schastl.id, owner.id)
    q2_id, q3_id, chkf_id, schastl_id, owner_id = ids
    client = await committed_client(KA_RF)

    async with committed_sessionmaker() as holder, holder.begin():
        audit = RequestAudit(service=AuditService(holder), user_id=owner_id, ip_address=None, user_agent=None)
        service = CounterpartyMappingService(holder, audit, VisibilityScope(office_id=None))
        result = await service.map(q2_id, first_entry, MapCounterpartyIn(legal_entity_id=chkf_id))
        assert result.propagated == 1
        racing = asyncio.create_task(
            client.post(
                f"/consolidations/versions/{q3_id}/osv/entries/{second_entry}/counterparty",
                json={"legal_entity_id": str(schastl_id)},
            )
        )
        await _wait_for_lock_waiter(committed_sessionmaker)
        assert not racing.done()
    response = await asyncio.wait_for(racing, LOCK_WAIT_SECONDS)

    assert response.status_code == HTTP_OK, response.text
    assert response.json()["propagated"] == 0
    async with committed_sessionmaker() as session:
        first = await _entry(session, first_entry)
        second = await _entry(session, second_entry)
        assert (first.counterparty_id, first.counterparty_match_kind) == (chkf_id, CounterpartyMatchKind.MANUAL)
        assert (second.counterparty_id, second.counterparty_match_kind) == (schastl_id, CounterpartyMatchKind.MANUAL)
        events = await _events(session, AuditAction.OSV_COUNTERPARTY_MAP)
        assert [event.consolidation_version_id for event in events] == [q2_id, q3_id]
        assert events[1].before_value == {"counterparty_id": str(chkf_id), "match_kind": "PROPAGATED"}


@pytest.mark.integration
async def test_manual_mapping_waits_for_alias_advisory_lock(
    committed_sessionmaker: async_sessionmaker[AsyncSession], committed_client: AuthClientFactory
) -> None:
    """M-4 (design №66): ручное сопоставление берёт `lock_alias_key` по нормализованному имени строки — пока другая
    транзакция держит блокировку этого алиаса (справочник, перенос), запрос ждёт, а после её commit завершается."""
    async with committed_sessionmaker() as setup, setup.begin():
        world = await _world(setup)
    entry_id = world.entries[("60", 1)]
    client = await committed_client(KA_RF)

    async with committed_sessionmaker() as holder, holder.begin():
        await directories_repo.lock_alias_key(holder, normalize_alias(NAME))
        racing = asyncio.create_task(
            client.post(_map_url(world, entry_id), json={"legal_entity_id": str(world.schastl.id)})
        )
        await _wait_for_lock_waiter(committed_sessionmaker)
        assert not racing.done()
    response = await asyncio.wait_for(racing, LOCK_WAIT_SECONDS)

    assert response.status_code == HTTP_OK, response.text
    assert response.json()["alias_created"] is True


@pytest.mark.integration
async def test_alias_delete_racing_with_move_alias_is_404_without_phantom_event(
    committed_sessionmaker: async_sessionmaker[AsyncSession], committed_client: AuthClientFactory
) -> None:
    """T11.3: `DELETE …/aliases/{id}` берёт ту же advisory-блокировку, что ручное сопоставление: пока перенос
    (`move_alias`) не зафиксирован, удаление ждёт, после commit находит алиас уже удалённым — 404, а фантомного
    события `LEGAL_ENTITY_CHANGE` об удалении (вторая запись про тот же алиас) нет."""
    async with committed_sessionmaker() as setup, setup.begin():
        world = await _world(setup)
        owner = await _user(setup, KA_RF)
        alias = LegalEntityAlias(
            legal_entity_id=world.schastl.id,
            alias=NAME,
            normalized_alias=normalize_alias(NAME),
            created_by=owner.id,
            created_at=utc_now(),
        )
        setup.add(alias)
        await setup.flush()
        alias_id, owner_id = alias.id, owner.id
    entry_id = world.entries[("60", 1)]
    admin = await committed_client(ADMIN)

    async with committed_sessionmaker() as holder, holder.begin():
        audit = RequestAudit(service=AuditService(holder), user_id=owner_id, ip_address=None, user_agent=None)
        service = CounterpartyMappingService(holder, audit, VisibilityScope(office_id=None))
        moved = await service.map(
            world.version.id, entry_id, MapCounterpartyIn(legal_entity_id=world.rb.id, move_alias=True)
        )
        assert moved.alias_moved_from == world.schastl.id
        racing = asyncio.create_task(admin.delete(f"{ENTITIES}/{world.schastl.id}/aliases/{alias_id}"))
        await _wait_for_lock_waiter(committed_sessionmaker)
        assert not racing.done()
    response = await asyncio.wait_for(racing, LOCK_WAIT_SECONDS)

    assert response.status_code == HTTP_NOT_FOUND, response.text
    async with committed_sessionmaker() as session:
        removals = [
            event
            for event in await _events(session, AuditAction.LEGAL_ENTITY_CHANGE)
            if event.object_id == alias_id and event.before_value is not None
        ]
        assert len(removals) == 1  # только удаление при переносе
        assert await session.get(LegalEntityAlias, alias_id) is None