arm-core/api/tests/test_api_consolidation_status.py

"""Ручные статусы версии консолидации Среза 4 и общий переход `transition` (spec consolidations-api «Статусы Среза 4»;
directory-policy «Видимость консолидаций и редактируемость версии»; audit-trail `CONSOLIDATION_STATUS_CHANGE`;
design D-6, D-8)."""

import datetime
import uuid
from typing import Any

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

from src.models.access import AppUser
from src.models.audit_ex import AuditEvent
from src.models.consolidation import (
    ConsolidationDocument,
    ConsolidationLock,
    ConsolidationStatusHistory,
    ConsolidationVersion,
)
from src.models.directories import LegalEntity, RegionalOffice
from src.models.enums import AuditAction, AuditSection, ConsolidationStatus, PeriodQuarter
from src.services import consolidation_locks, consolidation_status
from src.services.audit import AuditService
from tests.conftest import AuthClientFactory
from tests.factories import make_company_group, make_confirmed_osv, make_consolidation, make_legal_entity, make_lock

HTTP_OK = 200
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"
CURATOR = "curator_go@armka.local"
EXPERT = "expert@armka.local"
VERSIONS = "/consolidations/versions"
IN_WORK = (
    ConsolidationStatus.INITIATED,
    ConsolidationStatus.DOCUMENTS_PROCESSING,
    ConsolidationStatus.DATA_CORRECTION,
)


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,
    *,
    status: ConsolidationStatus = ConsolidationStatus.DATA_CORRECTION,
    office_name: str = ROSTOV,
    locked_by: str | None = KA_RF,
) -> tuple[ConsolidationVersion, list[LegalEntity]]:
    """Версия ГК филиала за 2024 Q3 из двух ЮЛ в статусе `status`; `locked_by` — владелец активной блокировки."""
    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,
    )
    if locked_by is not None:
        await make_lock(session, version=version, user=await _user(session, locked_by))
    return version, entities


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


async def _history(session: AsyncSession, version_id: uuid.UUID) -> list[ConsolidationStatusHistory]:
    stmt = (
        select(ConsolidationStatusHistory)
        .where(ConsolidationStatusHistory.version_id == version_id)
        .order_by(ConsolidationStatusHistory.changed_at, ConsolidationStatusHistory.id)
    )
    return list((await session.execute(stmt)).scalars())


async def _status_events(session: AsyncSession) -> list[AuditEvent]:
    stmt = select(AuditEvent).where(AuditEvent.action == AuditAction.CONSOLIDATION_STATUS_CHANGE)
    return list((await session.execute(stmt)).scalars())


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


async def _active_locks(session: AsyncSession, version_id: uuid.UUID) -> int:
    stmt = (
        select(func.count())
        .select_from(ConsolidationLock)
        .where(ConsolidationLock.consolidation_version_id == version_id, ConsolidationLock.released_at.is_(None))
    )
    return (await session.execute(stmt)).scalar_one()


# --- аннулирование -------------------------------------------------------------------------------------------------


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

    response = await client.post(
        _status_url(version.id),
        json={"to_status": "CANCELLED", "comment": "Дубль"},
        headers={"User-Agent": "pytest-status"},
    )

    assert response.status_code == HTTP_OK, response.text
    card = response.json()
    assert (card["status"], card["is_in_work"], card["lock"]) == ("CANCELLED", False, None)
    assert card["history"][-1]["from_status"] == "DATA_CORRECTION"
    assert (card["history"][-1]["to_status"], card["history"][-1]["comment"]) == ("CANCELLED", "Дубль")
    stored = await _reload(seeded, version.id)
    assert (stored.status, stored.cancelled_reason, stored.is_in_work) == (
        ConsolidationStatus.CANCELLED,
        "Дубль",
        False,
    )
    assert stored.updated_by == user.id
    assert stored.updated_at is not None
    assert await _active_locks(seeded, version.id) == 0
    last = (await _history(seeded, version.id))[-1]
    assert (last.from_status, last.to_status, last.changed_by, last.comment) == (
        ConsolidationStatus.DATA_CORRECTION,
        ConsolidationStatus.CANCELLED,
        user.id,
        "Дубль",
    )
    assert last.changed_at == stored.updated_at
    [event] = await _status_events(seeded)
    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 == user.id
    assert (event.before_value, event.after_value, event.comment) == (
        {"status": "DATA_CORRECTION"},
        {"status": "CANCELLED"},
        "Дубль",
    )
    # контекст запроса передан в аудит (ASGITransport отдаёт клиента 127.0.0.1)
    assert (event.ip_address, event.user_agent) == ("127.0.0.1", "pytest-status")


@pytest.mark.integration
@pytest.mark.parametrize("status", IN_WORK)
async def test_cancel_allowed_from_every_in_work_status(
    auth_client: AuthClientFactory, seeded: AsyncSession, status: ConsolidationStatus
) -> None:
    version, _ = await _version(seeded, status=status)
    client = await auth_client(KA_RF)

    response = await client.post(_status_url(version.id), json={"to_status": "CANCELLED", "comment": "Ошибка"})

    assert response.status_code == HTTP_OK, response.text
    assert response.json()["status"] == "CANCELLED"


@pytest.mark.integration
@pytest.mark.parametrize("payload", [{"to_status": "CANCELLED"}, {"to_status": "CANCELLED", "comment": "  "}])
async def test_cancel_without_comment(
    auth_client: AuthClientFactory, seeded: AsyncSession, payload: dict[str, Any]
) -> None:
    version, _ = await _version(seeded)
    client = await auth_client(KA_RF)

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

    assert response.status_code == HTTP_UNPROCESSABLE
    assert response.json()["detail"]["code"] == "COMMENT_REQUIRED"
    assert (await _reload(seeded, version.id)).status == ConsolidationStatus.DATA_CORRECTION
    assert await _active_locks(seeded, version.id) == 1


# --- возврат в INITIATED -------------------------------------------------------------------------------------------


@pytest.mark.integration
@pytest.mark.parametrize("status", [ConsolidationStatus.DOCUMENTS_PROCESSING, ConsolidationStatus.DATA_CORRECTION])
async def test_return_to_initiated_keeps_bound_documents(
    auth_client: AuthClientFactory, seeded: AsyncSession, status: ConsolidationStatus
) -> None:
    version, entities = await _version(seeded, status=status)
    document = await make_confirmed_osv(
        seeded, entity=entities[0], account="60", year=2024, quarter=PeriodQuarter.Q3, is_cumulative=True, rows=[]
    )
    seeded.add(ConsolidationDocument(consolidation_version_id=version.id, document_id=document.id))
    await seeded.flush()
    client = await auth_client(KA_RF)

    response = await client.post(_status_url(version.id), json={"to_status": "INITIATED"})

    assert response.status_code == HTTP_OK, response.text
    card = response.json()
    assert (card["status"], card["is_in_work"]) == ("INITIATED", True)
    assert card["lock"]["locked_by"] == str((await _user(seeded, KA_RF)).id)  # блокировка остаётся
    last = (await _history(seeded, version.id))[-1]
    assert (last.from_status, last.to_status, last.comment) == (status, ConsolidationStatus.INITIATED, None)
    bound = (
        await seeded.execute(
            select(ConsolidationDocument.document_id).where(
                ConsolidationDocument.consolidation_version_id == version.id
            )
        )
    ).scalars()
    assert list(bound) == [document.id]
    [event] = await _status_events(seeded)
    assert (event.before_value, event.after_value, event.comment) == (
        {"status": status.value},
        {"status": "INITIATED"},
        None,
    )


# --- запрещённые переходы ------------------------------------------------------------------------------------------


@pytest.mark.integration
@pytest.mark.parametrize(
    ("status", "to_status"),
    [
        (ConsolidationStatus.DATA_CORRECTION, "CONSOLIDATION_RF"),  # переход Среза 7
        (ConsolidationStatus.INITIATED, "INITIATED"),  # повтор статуса
        (ConsolidationStatus.DATA_CORRECTION, "DATA_CORRECTION"),
        (ConsolidationStatus.INITIATED, "DOCUMENTS_PROCESSING"),  # автоматический, не ручной
        (ConsolidationStatus.DOCUMENTS_PROCESSING, "DATA_CORRECTION"),
        (ConsolidationStatus.DATA_CORRECTION, "VERIFIED"),
    ],
)
async def test_transition_not_allowed(
    auth_client: AuthClientFactory, seeded: AsyncSession, status: ConsolidationStatus, to_status: str
) -> None:
    version, _ = await _version(seeded, status=status)
    history_before = len(await _history(seeded, version.id))
    client = await auth_client(KA_RF)

    response = await client.post(_status_url(version.id), json={"to_status": to_status, "comment": "попытка"})

    assert response.status_code == HTTP_CONFLICT, response.text
    detail = response.json()["detail"]
    assert (detail["code"], detail["from_status"], detail["to_status"]) == (
        "TRANSITION_NOT_ALLOWED",
        status.value,
        to_status,
    )
    assert (await _reload(seeded, version.id)).status == status
    assert len(await _history(seeded, version.id)) == history_before
    assert await _status_events(seeded) == []


# --- политика редактируемости на ручной смене статуса --------------------------------------------------------------


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

    response = await client.post(_status_url(version.id), json={"to_status": "CANCELLED", "comment": "x"})

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


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

    response = await client.post(_status_url(version.id), json={"to_status": "CANCELLED", "comment": "x"})

    assert response.status_code == HTTP_CONFLICT
    detail = response.json()["detail"]
    assert (detail["code"], detail["locked_by"]) == ("LOCKED_BY_OTHER", str(owner.id))
    assert "locked_at" in detail


@pytest.mark.integration
async def test_status_of_cancelled_version_is_not_editable(
    auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
    """Аннулированная версия: даже владелец прежней блокировки получает 409 `VERSION_NOT_EDITABLE`."""
    version, _ = await _version(seeded, status=ConsolidationStatus.CANCELLED)
    client = await auth_client(KA_RF)

    response = await client.post(_status_url(version.id), json={"to_status": "INITIATED"})

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


@pytest.mark.integration
async def test_status_of_foreign_version_is_not_found(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    version, _ = await _version(seeded, office_name=KRASNODAR, locked_by=None)
    client = await auth_client(KA_RF)
    body = {"to_status": "CANCELLED", "comment": "x"}

    foreign = await client.post(_status_url(version.id), json=body)
    missing = await client.post(_status_url(uuid.uuid4()), json=body)

    assert foreign.status_code == HTTP_NOT_FOUND
    assert foreign.json() == missing.json()


@pytest.mark.integration
@pytest.mark.parametrize("email", [CURATOR, EXPERT])
async def test_status_requires_consolidation_edit(
    auth_client: AuthClientFactory, seeded: AsyncSession, email: str
) -> None:
    version, _ = await _version(seeded)
    client = await auth_client(email)

    response = await client.post(_status_url(version.id), json={"to_status": "CANCELLED", "comment": "x"})

    assert response.status_code == HTTP_FORBIDDEN


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

    response = await client.post(_status_url(version.id), json={"to_status": "DONE"})

    assert response.status_code == HTTP_UNPROCESSABLE


# --- transition: единственное место смены статуса (автопереходы Task 6–7 зовут её же) ------------------------------


def test_manual_transitions_table() -> None:
    cancelled, initiated = ConsolidationStatus.CANCELLED, ConsolidationStatus.INITIATED
    assert consolidation_status.MANUAL_TRANSITIONS == {
        ConsolidationStatus.INITIATED: frozenset({cancelled}),
        ConsolidationStatus.DOCUMENTS_PROCESSING: frozenset({cancelled, initiated}),
        ConsolidationStatus.DATA_CORRECTION: frozenset({cancelled, initiated}),
    }


@pytest.mark.integration
async def test_transition_to_cancelled_releases_active_lock(
    seeded: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    version, _ = await _version(seeded, status=ConsolidationStatus.INITIATED)
    user = await _user(seeded, RKA)
    now = datetime.datetime(2026, 10, 6, 12, 0, tzinfo=datetime.UTC)
    calls: list[tuple[uuid.UUID, datetime.datetime]] = []
    original = consolidation_locks.release_active_lock

    async def spy(session: AsyncSession, version_id: uuid.UUID, at: datetime.datetime) -> ConsolidationLock | None:
        calls.append((version_id, at))
        return await original(session, version_id, at)

    monkeypatch.setattr(consolidation_locks, "release_active_lock", spy)

    record = await consolidation_status.transition(
        seeded,
        AuditService(seeded),
        version=version,
        to=ConsolidationStatus.CANCELLED,
        changed_by=user.id,
        comment="Аннулирована",
        ip_address="10.0.0.1",
        user_agent="pytest",
        now=now,
    )

    assert calls == [(version.id, now)]
    assert await _active_locks(seeded, version.id) == 0
    released = (
        await seeded.execute(select(ConsolidationLock).where(ConsolidationLock.consolidation_version_id == version.id))
    ).scalar_one()
    assert released.released_at == now
    assert (record.from_status, record.to_status, record.changed_by, record.changed_at, record.comment) == (
        ConsolidationStatus.INITIATED,
        ConsolidationStatus.CANCELLED,
        user.id,
        now,
        "Аннулирована",
    )
    assert (version.status, version.is_in_work, version.cancelled_reason) == (
        ConsolidationStatus.CANCELLED,
        False,
        "Аннулирована",
    )
    assert (version.updated_by, version.updated_at) == (user.id, now)
    [event] = await _status_events(seeded)
    assert (event.user_id, event.ip_address, event.user_agent) == (user.id, "10.0.0.1", "pytest")


@pytest.mark.integration
async def test_transition_other_than_cancelled_keeps_lock(
    seeded: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Автоматический переход (Task 6–7) и возврат в `INITIATED` блокировку не снимают; `user_id` — `changed_by`."""
    version, _ = await _version(seeded, status=ConsolidationStatus.INITIATED)
    confirmer = await _user(seeded, RKA)
    calls: list[uuid.UUID] = []

    async def spy(_session: AsyncSession, version_id: uuid.UUID, _at: datetime.datetime) -> None:
        calls.append(version_id)

    monkeypatch.setattr(consolidation_locks, "release_active_lock", spy)

    await consolidation_status.transition(
        seeded,
        AuditService(seeded),
        version=version,
        to=ConsolidationStatus.DATA_CORRECTION,
        changed_by=confirmer.id,
        comment=None,
        now=datetime.datetime.now(datetime.UTC),
    )

    assert calls == []
    assert await _active_locks(seeded, version.id) == 1
    assert (version.status, version.is_in_work, version.cancelled_reason) == (
        ConsolidationStatus.DATA_CORRECTION,
        True,
        None,
    )
    [event] = await _status_events(seeded)
    assert (event.user_id, event.ip_address, event.before_value, event.after_value) == (
        confirmer.id,
        None,
        {"status": "INITIATED"},
        {"status": "DATA_CORRECTION"},
    )