arm-core/api/tests/test_api_consolidation_locks.py

"""Блокировка версии консолидации: взять в работу, перехват, снятие (spec consolidations-api «Взять в работу,
перехват, снятие»; audit-trail `CONSOLIDATION_LOCK`, `CONSOLIDATION_LOCK_TAKEOVER`; design D-7, D-9; Review Focus 4).

Гонки — на `committed_*` (настоящие параллельные транзакции): два одновременных `POST …/lock` и перехват во время
правки версии владельцем (правка держит `consolidation_version FOR UPDATE`, перехват ждёт её commit).
"""

import asyncio
import datetime
import uuid
from typing import Any

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

from src.models.access import AppUser
from src.models.audit_ex import AuditEvent
from src.models.consolidation import Consolidation, ConsolidationLock, ConsolidationMember, ConsolidationVersion
from src.models.directories import CompanyGroup, RegionalOffice
from src.models.enums import AuditAction, AuditSection, ConsolidationStatus, PeriodQuarter
from src.repositories import consolidations as repo
from src.services import consolidation_status
from src.services.audit import AuditService
from tests.conftest import AuthClientFactory
from tests.factories import make_company_group, make_consolidation, make_legal_entity, make_lock

HTTP_OK = 200
HTTP_CREATED = 201
HTTP_NO_CONTENT = 204
HTTP_FORBIDDEN = 403
HTTP_NOT_FOUND = 404
HTTP_CONFLICT = 409
HTTP_UNPROCESSABLE = 422
ROSTOV = "Ростовский региональный филиал"
KRASNODAR = "Краснодарский региональный филиал"
KA_RF = "ka_rf@armka.local"  # «А»
RKA = "rka@armka.local"  # «Б», тот же филиал
EXPERT = "expert@armka.local"
VERSIONS = "/consolidations/versions"
LOCK_WAIT_SECONDS = 5
MAX_REASON = 1000


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


async def _version(
    session: AsyncSession,
    *,
    office_name: str = ROSTOV,
    status: ConsolidationStatus = ConsolidationStatus.INITIATED,
) -> ConsolidationVersion:
    """Версия консолидации ГК филиала `office_name` за 2024 Q3 из двух ЮЛ (инициатор — ka_rf)."""
    office = (await session.execute(select(RegionalOffice).where(RegionalOffice.name == office_name))).scalar_one()
    group = await make_company_group(session, name=f"ГК {uuid.uuid4().hex[:6]}", office=office)
    entities = [await make_legal_entity(session, name=f"ООО Член {n}", group=group) for n in range(2)]
    _, version = await make_consolidation(
        session,
        group=group,
        year=2024,
        quarter=PeriodQuarter.Q3,
        members=entities,
        initiator=await _user(session, KA_RF),
        status=status,
    )
    return version


def _lock_url(version_id: uuid.UUID) -> str:
    return f"{VERSIONS}/{version_id}/lock"


async def _locks(session: AsyncSession, version_id: uuid.UUID) -> list[ConsolidationLock]:
    stmt = (
        select(ConsolidationLock)
        .where(ConsolidationLock.consolidation_version_id == version_id)
        .order_by(ConsolidationLock.locked_at)
        .execution_options(populate_existing=True)
    )
    return list((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())


# --- взять в работу ------------------------------------------------------------------------------------------------


@pytest.mark.integration
async def test_take_free_version_creates_lock(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    version = await _version(seeded)
    user = await _user(seeded, KA_RF)
    client = await auth_client(KA_RF)

    response = await client.post(_lock_url(version.id))

    assert response.status_code == HTTP_CREATED, response.text
    body = response.json()
    assert (body["locked_by"], body["locked_by_name"]) == (str(user.id), user.full_name)
    assert (body["taken_over_from"], body["takeover_reason"]) == (None, None)
    [lock] = await _locks(seeded, version.id)
    assert (lock.locked_by, lock.released_at) == (user.id, None)
    assert datetime.datetime.fromisoformat(body["locked_at"]) == lock.locked_at
    [event] = await _events(seeded, AuditAction.CONSOLIDATION_LOCK)
    assert (event.section, event.object_type, event.object_id) == (
        AuditSection.CONSOLIDATION,
        "consolidation_version",
        version.id,
    )
    assert (event.consolidation_version_id, event.user_id) == (version.id, user.id)
    assert event.before_value is None
    assert event.after_value == {"locked_by": str(user.id), "released": False}
    card = (await client.get(f"{VERSIONS}/{version.id}")).json()
    assert card["lock"]["locked_by"] == str(user.id)


@pytest.mark.integration
async def test_take_own_lock_returns_it_unchanged(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    version = await _version(seeded)
    client = await auth_client(KA_RF)
    first = await client.post(_lock_url(version.id))

    again = await client.post(_lock_url(version.id), json={"takeover_reason": "повтор"})

    assert again.status_code == HTTP_OK, again.text
    assert again.json() == first.json()
    assert len(await _locks(seeded, version.id)) == 1
    assert len(await _events(seeded, AuditAction.CONSOLIDATION_LOCK)) == 1
    assert await _events(seeded, AuditAction.CONSOLIDATION_LOCK_TAKEOVER) == []


@pytest.mark.integration
@pytest.mark.parametrize("payload", [None, {}, {"takeover_reason": None}, {"takeover_reason": "   "}])
async def test_take_foreign_lock_without_reason_is_conflict(
    auth_client: AuthClientFactory, seeded: AsyncSession, payload: dict[str, Any] | None
) -> None:
    version = await _version(seeded)
    owner = await _user(seeded, KA_RF)
    lock = await make_lock(seeded, version=version, user=owner)
    client = await auth_client(RKA)

    response = await client.post(_lock_url(version.id), json=payload)

    assert response.status_code == HTTP_CONFLICT, response.text
    detail = response.json()["detail"]
    assert detail["code"] == "LOCKED_BY_OTHER"
    assert (detail["locked_by"], detail["locked_by_name"]) == (str(owner.id), owner.full_name)
    assert datetime.datetime.fromisoformat(detail["locked_at"]) == lock.locked_at
    [still] = await _locks(seeded, version.id)
    assert still.released_at is None


@pytest.mark.integration
async def test_lock_does_not_expire(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    """ASSUMPTION(OQ-119): блокировка двухлетней давности по-прежнему чужая."""
    version = await _version(seeded)
    lock = await make_lock(seeded, version=version, user=await _user(seeded, KA_RF))
    lock.locked_at = lock.locked_at - datetime.timedelta(days=730)
    await seeded.flush()
    client = await auth_client(RKA)

    response = await client.post(_lock_url(version.id))

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


@pytest.mark.integration
async def test_takeover_with_reason(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    version = await _version(seeded)
    owner = await _user(seeded, KA_RF)
    taker = await _user(seeded, RKA)
    await make_lock(seeded, version=version, user=owner)
    client = await auth_client(RKA)

    response = await client.post(_lock_url(version.id), json={"takeover_reason": "Отпуск"})

    assert response.status_code == HTTP_CREATED, response.text
    body = response.json()
    assert (body["locked_by"], body["locked_by_name"]) == (str(taker.id), taker.full_name)
    assert (body["taken_over_from"], body["takeover_reason"]) == (str(owner.id), "Отпуск")
    old, new = await _locks(seeded, version.id)
    assert old.locked_by == owner.id
    assert old.released_at is not None
    assert (new.locked_by, new.released_at, new.taken_over_from, new.takeover_reason) == (
        taker.id,
        None,
        owner.id,
        "Отпуск",
    )
    assert old.released_at == new.locked_at  # одно `now` на операцию
    [event] = await _events(seeded, AuditAction.CONSOLIDATION_LOCK_TAKEOVER)
    assert (event.section, event.object_type, event.object_id, event.consolidation_version_id) == (
        AuditSection.CONSOLIDATION,
        "consolidation_version",
        version.id,
        version.id,
    )
    assert event.user_id == taker.id
    assert event.before_value == {"locked_by": str(owner.id)}
    assert event.after_value == {"locked_by": str(taker.id)}
    assert event.comment == "Отпуск"
    assert await _events(seeded, AuditAction.CONSOLIDATION_LOCK) == []


@pytest.mark.integration
async def test_takeover_reason_longer_than_limit_is_rejected(
    auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
    version = await _version(seeded)
    await make_lock(seeded, version=version, user=await _user(seeded, KA_RF))
    client = await auth_client(RKA)

    too_long = await client.post(_lock_url(version.id), json={"takeover_reason": "я" * (MAX_REASON + 1)})
    at_limit = await client.post(_lock_url(version.id), json={"takeover_reason": "я" * MAX_REASON})

    assert too_long.status_code == HTTP_UNPROCESSABLE
    assert at_limit.status_code == HTTP_CREATED, at_limit.text


@pytest.mark.integration
async def test_take_cancelled_version_is_not_editable(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    version = await _version(seeded, status=ConsolidationStatus.CANCELLED)
    client = await auth_client(KA_RF)

    response = await client.post(_lock_url(version.id))

    assert response.status_code == HTTP_CONFLICT
    assert response.json()["detail"]["code"] == "VERSION_NOT_EDITABLE"
    assert await _locks(seeded, version.id) == []


@pytest.mark.integration
async def test_take_version_outside_in_work_but_not_cancelled_is_allowed(
    auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
    """«В любом статусе, кроме CANCELLED»: версию на маршруте согласования (Срез 7) можно взять в работу."""
    version = await _version(seeded, status=ConsolidationStatus.CONSOLIDATION_RF)
    client = await auth_client(KA_RF)

    response = await client.post(_lock_url(version.id))

    assert response.status_code == HTTP_CREATED, response.text


@pytest.mark.integration
async def test_lock_routes_hide_foreign_version(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    version = await _version(seeded, office_name=KRASNODAR)
    client = await auth_client(KA_RF)

    for method in ("post", "delete"):
        foreign = await client.request(method, _lock_url(version.id))
        missing = await client.request(method, _lock_url(uuid.uuid4()))
        assert foreign.status_code == HTTP_NOT_FOUND
        assert foreign.json() == missing.json()
    assert await _locks(seeded, version.id) == []


@pytest.mark.integration
async def test_lock_routes_require_consolidation_edit(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    version = await _version(seeded)
    client = await auth_client(EXPERT)

    assert (await client.post(_lock_url(version.id))).status_code == HTTP_FORBIDDEN
    assert (await client.delete(_lock_url(version.id))).status_code == HTTP_FORBIDDEN


# --- снятие --------------------------------------------------------------------------------------------------------


@pytest.mark.integration
async def test_release_own_lock(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    version = await _version(seeded)
    user = await _user(seeded, KA_RF)
    await make_lock(seeded, version=version, user=user)
    client = await auth_client(KA_RF)

    response = await client.delete(_lock_url(version.id))

    assert response.status_code == HTTP_NO_CONTENT, response.text
    assert response.content == b""
    [lock] = await _locks(seeded, version.id)
    assert lock.released_at is not None
    [event] = await _events(seeded, AuditAction.CONSOLIDATION_LOCK)
    assert (event.object_id, event.consolidation_version_id, event.user_id) == (version.id, version.id, user.id)
    assert event.after_value == {"locked_by": str(user.id), "released": True}
    assert (await client.get(f"{VERSIONS}/{version.id}")).json()["lock"] is None


@pytest.mark.integration
async def test_release_foreign_lock_is_conflict(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    version = await _version(seeded)
    owner = await _user(seeded, KA_RF)
    await make_lock(seeded, version=version, user=owner)
    client = await auth_client(RKA)

    response = await client.delete(_lock_url(version.id))

    assert response.status_code == HTTP_CONFLICT
    detail = response.json()["detail"]
    assert (detail["code"], detail["locked_by"]) == ("NOT_LOCK_OWNER", str(owner.id))
    [lock] = await _locks(seeded, version.id)
    assert lock.released_at is None
    assert await _events(seeded, AuditAction.CONSOLIDATION_LOCK) == []


@pytest.mark.integration
async def test_release_without_active_lock_is_noop(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    version = await _version(seeded)
    await make_lock(seeded, version=version, user=await _user(seeded, KA_RF), released=True)
    client = await auth_client(KA_RF)

    response = await client.delete(_lock_url(version.id))

    assert response.status_code == HTTP_NO_CONTENT
    assert await _events(seeded, AuditAction.CONSOLIDATION_LOCK) == []


# --- гонки (настоящие параллельные транзакции) ---------------------------------------------------------------------


async def _committed_version(maker: async_sessionmaker[AsyncSession]) -> uuid.UUID:
    async with maker() as session, session.begin():
        return (await _version(session)).id


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_two_simultaneous_takes_one_wins(
    committed_sessionmaker: async_sessionmaker[AsyncSession], committed_client: AuthClientFactory
) -> None:
    """Два `POST …/lock` свободной версии одновременно: у одного 201, у другого 409 `LOCKED_BY_OTHER`."""
    version_id = await _committed_version(committed_sessionmaker)
    first, second = await committed_client(KA_RF), await committed_client(RKA)

    responses = await asyncio.wait_for(
        asyncio.gather(first.post(_lock_url(version_id)), second.post(_lock_url(version_id))), LOCK_WAIT_SECONDS
    )

    assert sorted(response.status_code for response in responses) == [HTTP_CREATED, HTTP_CONFLICT]
    winner = next(response for response in responses if response.status_code == HTTP_CREATED)
    loser = next(response for response in responses if response.status_code == HTTP_CONFLICT)
    assert loser.json()["detail"]["code"] == "LOCKED_BY_OTHER"
    assert loser.json()["detail"]["locked_by"] == winner.json()["locked_by"]
    async with committed_sessionmaker() as session:
        active = (
            await session.execute(
                select(func.count())
                .select_from(ConsolidationLock)
                .where(
                    ConsolidationLock.consolidation_version_id == version_id, ConsolidationLock.released_at.is_(None)
                )
            )
        ).scalar_one()
        assert active == 1


@pytest.mark.integration
async def test_takeover_waits_for_owner_edit_then_owner_loses_lock(
    committed_sessionmaker: async_sessionmaker[AsyncSession], committed_client: AuthClientFactory
) -> None:
    """Review Focus 4: владелец А правит периметр (`consolidation_version FOR UPDATE`), Б в это время перехватывает.

    Перехват ждёт транзакцию А; после её commit правка А зафиксирована, Б — владелец, следующая правка А —
    409 `LOCKED_BY_OTHER`. Этот сценарий ждал бы и без `FOR UPDATE` версии в `take` (вставка блокировки берёт FK
    `KEY SHARE` на версию); порядок D-9 различает `test_takeover_during_cancellation_sees_cancelled_version`.
    """
    async with committed_sessionmaker() as setup, setup.begin():
        version = await _version(setup)
        owner = await _user(setup, KA_RF)
        await make_lock(setup, version=version, user=owner)
        consolidation = await setup.get(Consolidation, version.consolidation_id)
        assert consolidation is not None
        group = await setup.get(CompanyGroup, consolidation.company_group_id)
        assert group is not None
        newcomer_id = (await make_legal_entity(setup, name="ООО Новый член", group=group)).id
        version_id = version.id
    taker_client = await committed_client(RKA)
    owner_client = await committed_client(KA_RF)

    async with committed_sessionmaker() as editor, editor.begin():
        # правка периметра владельцем: версия под FOR UPDATE первой (D-9), затем строки версии
        locked = await repo.lock_version(editor, version_id)
        assert locked is not None
        editor.add(ConsolidationMember(consolidation_version_id=version_id, legal_entity_id=newcomer_id))
        await editor.flush()
        takeover = asyncio.create_task(taker_client.post(_lock_url(version_id), json={"takeover_reason": "Отпуск"}))
        await _wait_for_lock_waiter(committed_sessionmaker)
        assert not takeover.done()
    response = await asyncio.wait_for(takeover, LOCK_WAIT_SECONDS)

    assert response.status_code == HTTP_CREATED, response.text
    assert response.json()["taken_over_from"] == str(owner.id)
    async with committed_sessionmaker() as session:
        members = (
            await session.execute(
                select(ConsolidationMember.legal_entity_id).where(
                    ConsolidationMember.consolidation_version_id == version_id
                )
            )
        ).scalars()
        assert newcomer_id in set(members)  # правка владельца зафиксирована до перехвата
    next_edit = await owner_client.post(
        f"{VERSIONS}/{version_id}/status", json={"to_status": "CANCELLED", "comment": "x"}
    )
    assert next_edit.status_code == HTTP_CONFLICT, next_edit.text
    assert next_edit.json()["detail"]["code"] == "LOCKED_BY_OTHER"
    assert next_edit.json()["detail"]["locked_by"] == response.json()["locked_by"]


@pytest.mark.integration
async def test_takeover_during_cancellation_sees_cancelled_version(
    committed_sessionmaker: async_sessionmaker[AsyncSession], committed_client: AuthClientFactory
) -> None:
    """Перехват читает статус и блокировку только ПОСЛЕ `FOR UPDATE` версии (D-9).

    Владелец аннулирует версию (транзакция держит версию и снимает свою блокировку); перехват, пришедший в это время,
    ждёт и после commit видит `CANCELLED` → 409 `VERSION_NOT_EDITABLE`. Прочитай он состояние до ожидания —
    перехватил бы аннулированную версию (вставку блокировки задержал бы лишь FK, а не проверка статуса).
    """
    async with committed_sessionmaker() as setup, setup.begin():
        version = await _version(setup)
        owner = await _user(setup, KA_RF)
        await make_lock(setup, version=version, user=owner)
        version_id, owner_id = version.id, owner.id
    taker_client = await committed_client(RKA)

    async with committed_sessionmaker() as editor, editor.begin():
        locked = await repo.lock_version(editor, version_id)
        assert locked is not None
        await consolidation_status.transition(
            editor,
            AuditService(editor),
            version=locked,
            to=ConsolidationStatus.CANCELLED,
            changed_by=owner_id,
            comment="Дубль",
            now=datetime.datetime.now(datetime.UTC),
        )
        takeover = asyncio.create_task(taker_client.post(_lock_url(version_id), json={"takeover_reason": "Отпуск"}))
        await _wait_for_lock_waiter(committed_sessionmaker)
        assert not takeover.done()
    response = await asyncio.wait_for(takeover, LOCK_WAIT_SECONDS)

    assert response.status_code == HTTP_CONFLICT, response.text
    assert response.json()["detail"]["code"] == "VERSION_NOT_EDITABLE"
    async with committed_sessionmaker() as session:
        assert all(lock.released_at is not None for lock in await _locks(session, version_id))


@pytest.mark.integration
async def test_foreign_version_is_hidden_without_waiting_for_row_lock(
    committed_sessionmaker: async_sessionmaker[AsyncSession], committed_client: AuthClientFactory
) -> None:
    """M-6: видимость проверяется до `FOR UPDATE` — чужая версия отвечает 404 сразу, даже когда её строку держит
    правка владельца (иначе невидимый пользователь ждал бы чужую транзакцию и мог влиять на её длительность)."""
    async with committed_sessionmaker() as setup, setup.begin():
        version_id = (await _version(setup, office_name=KRASNODAR)).id
    client = await committed_client(KA_RF)

    async with committed_sessionmaker() as holder, holder.begin():
        assert await repo.lock_version(holder, version_id) is not None
        response = await asyncio.wait_for(client.post(_lock_url(version_id)), LOCK_WAIT_SECONDS)

    assert response.status_code == HTTP_NOT_FOUND