arm-core/api/tests/test_api_intercompany_matrix.py

"""Критерии шахматки через HTTP, реальные сервисы и PostgreSQL."""

import uuid
from dataclasses import dataclass
from decimal import Decimal
from io import BytesIO
from urllib.parse import unquote

import pytest
from httpx import AsyncClient, Response
from openpyxl import load_workbook
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from src.models.access import AppUser
from src.models.accounting import ChartOfAccount
from src.models.audit_ex import AuditEvent
from src.models.consolidation import ConsolidationDocument, ConsolidationVersion
from src.models.directories import CompanyGroup, LegalEntity, RegionalOffice
from src.models.enums import AuditAction, AuditSection, CounterpartyMatchKind, PeriodQuarter
from src.models.osv import OsvEntry
from src.repositories import consolidations as consolidation_repo
from src.repositories import directories as directory_repo
from tests.conftest import AuthClientFactory
from tests.factories import make_company_group, make_confirmed_osv, make_consolidation, make_legal_entity

pytestmark = pytest.mark.integration
OK, FORBIDDEN, NOT_FOUND, UNPROCESSABLE = 200, 403, 404, 422
KA_RF = "ka_rf@armka.local"
ROUTES = ("all", "manual", "result-table", "all/export", "manual/export", "result-table/export")


@dataclass
class World:
    version: ConsolidationVersion
    members: list[LegalEntity]
    accounts: dict[str, ChartOfAccount]

    @property
    def selection(self) -> dict[str, list[str]]:
        return {
            "debit_account_ids": [str(self.accounts["62.01"].id)],
            "credit_account_ids": [str(self.accounts["60.01"].id)],
        }


async def _world(session: AsyncSession, *, members: int = 2) -> World:
    office = (
        await session.scalars(select(RegionalOffice).where(RegionalOffice.name == "Ростовский региональный филиал"))
    ).one()
    user = (await session.scalars(select(AppUser).where(AppUser.email == KA_RF))).one()
    group = await make_company_group(session, office=office)
    entities = [await make_legal_entity(session, name=f"Компания {n}", group=group) for n in range(members)]
    _, version = await make_consolidation(
        session, group=group, year=2024, quarter=PeriodQuarter.Q1, members=entities, initiator=user
    )
    accounts = {account.code: account for account in (await session.scalars(select(ChartOfAccount))).all()}
    return World(version, entities, accounts)


async def _request(client: AsyncClient, version_id: uuid.UUID, suffix: str, selection: dict) -> Response:
    url = f"/consolidations/versions/{version_id}/intercompany-matrix/{suffix}"
    return await client.post(url, json=selection) if suffix.startswith("manual") else await client.get(url)


async def _source(
    session: AsyncSession, world: World, *, credit: bool, amount: str, account: str | None = None
) -> uuid.UUID:
    owner, counterparty = reversed(world.members) if credit else world.members
    code = account or ("60.01" if credit else "62.01")
    document = await make_confirmed_osv(
        session,
        entity=owner,
        account=code,
        year=2024,
        quarter=PeriodQuarter.Q1,
        is_cumulative=False,
        rows=[
            {
                "name": counterparty.name,
                "counterparty_id": counterparty.id,
                "counterparty_match_kind": CounterpartyMatchKind.INN,
                "account_id": world.accounts[code].id,
                "closing_credit" if credit else "closing_debit": Decimal(amount),
            }
        ],
    )
    session.add(ConsolidationDocument(consolidation_version_id=world.version.id, document_id=document.id))
    await session.flush()
    return document.id


@pytest.mark.parametrize("suffix", ["all", "manual"])
async def test_empty_perimeter(auth_client: AuthClientFactory, seeded: AsyncSession, suffix: str) -> None:
    world = await _world(seeded, members=0)
    response = await _request(await auth_client(KA_RF), world.version.id, suffix, world.selection)
    assert response.status_code == OK, response.text
    assert response.json() == {"companies": [], "cells": []}


@pytest.mark.parametrize("suffix", ROUTES)
async def test_matrix_reuses_members_and_reads_accounts_through_repository(
    auth_client: AuthClientFactory, seeded: AsyncSession, monkeypatch: pytest.MonkeyPatch, suffix: str
) -> None:
    world = await _world(seeded)
    calls = {"members": 0, "accounts": 0}
    original_members = consolidation_repo.members
    original_accounts = directory_repo.list_accounts

    async def counted_members(*args: object, **kwargs: object):
        calls["members"] += 1
        return await original_members(*args, **kwargs)

    async def counted_accounts(*args: object, **kwargs: object):
        calls["accounts"] += 1
        return await original_accounts(*args, **kwargs)

    monkeypatch.setattr(consolidation_repo, "members", counted_members)
    monkeypatch.setattr(directory_repo, "list_accounts", counted_accounts)
    response = await _request(await auth_client(KA_RF), world.version.id, suffix, world.selection)
    assert response.status_code == OK, response.text
    assert calls == {"members": 1, "accounts": 1}


async def test_no_osv_for_selected_accounts(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    world = await _world(seeded)
    await _source(seeded, world, credit=False, amount="999", account="58.03")
    response = await _request(await auth_client(KA_RF), world.version.id, "manual", world.selection)
    assert response.status_code == OK, response.text
    assert len(response.json()["cells"]) == len(world.members)
    for cell in response.json()["cells"]:
        assert {"osv_entry_ids_debit", "osv_entry_ids_credit", "exceeds_threshold"} <= cell.keys()
        assert {"entry_osv_ids_debit", "entry_osv_ids_credit", "is_more_5_percent"}.isdisjoint(cell)
        assert cell["amount_a"] == cell["amount_b"] == cell["variance_amount"] == "0.00"
        assert cell["variance_percent"] is None
        assert cell["one_sided"] is False
        assert cell["osv_entry_ids_debit"] == cell["osv_entry_ids_credit"] == []
        assert cell["skipped_osv_entry_count"] == 0


async def test_unmapped_account_source_is_visible_in_cell_count(
    auth_client: AuthClientFactory, seeded: AsyncSession
) -> None:
    world = await _world(seeded)
    document_id = await _source(seeded, world, credit=False, amount="100")
    entry = (await seeded.scalars(select(OsvEntry).where(OsvEntry.document_id == document_id))).one()
    entry.account_id = None
    await seeded.flush()
    await _source(seeded, world, credit=True, amount="80")
    response = await _request(await auth_client(KA_RF), world.version.id, "manual", world.selection)
    assert response.status_code == OK, response.text
    cells = response.json()["cells"]
    assert [cell["skipped_osv_entry_count"] for cell in cells] == [1, 1]
    cell = next(c for c in cells if c["company_a_id"] == str(world.members[0].id))
    assert cell["amount_a"] == "0.00"
    assert cell["amount_b"] == "80.00"
    assert cell["osv_entry_ids_debit"] == []
    assert len(cell["osv_entry_ids_credit"]) == 1


@pytest.mark.parametrize("credit", [False, True])
@pytest.mark.parametrize("amount", ["0", "100.25"])
async def test_pair_without_counterpart(
    auth_client: AuthClientFactory, seeded: AsyncSession, credit: bool, amount: str
) -> None:
    world = await _world(seeded)
    await _source(seeded, world, credit=credit, amount=amount)
    response = await _request(await auth_client(KA_RF), world.version.id, "manual", world.selection)
    assert response.status_code == OK, response.text
    cell = next(c for c in response.json()["cells"] if c["company_a_id"] == str(world.members[0].id))
    assert Decimal(cell["amount_a"]) == (Decimal(0) if credit else Decimal(amount))
    assert Decimal(cell["amount_b"]) == (Decimal(amount) if credit else Decimal(0))
    assert cell["one_sided"] is True
    assert cell["variance_percent"] is None
    assert cell["exceeds_threshold"] is False
    assert bool(cell["osv_entry_ids_credit"]) is credit
    assert bool(cell["osv_entry_ids_debit"]) is not credit


@pytest.mark.parametrize("suffix", ROUTES)
async def test_visibility_and_permission(auth_client: AuthClientFactory, seeded: AsyncSession, suffix: str) -> None:
    world = await _world(seeded)
    foreign = await auth_client("client_manager@armka.local")
    hidden = await _request(foreign, world.version.id, suffix, world.selection)
    unknown = await _request(foreign, uuid.uuid4(), suffix, world.selection)
    assert hidden.status_code == unknown.status_code == NOT_FOUND
    assert hidden.json() == unknown.json()
    forbidden = await _request(await auth_client("admin@armka.local"), world.version.id, suffix, world.selection)
    assert forbidden.status_code == FORBIDDEN
    assert "CONSOLIDATION_VIEW" in forbidden.json()["detail"]


@pytest.mark.parametrize(("debit", "credit", "percent"), [("8890", "8000", "11.13"), ("105001", "100000", "5.00")])
async def test_rounded_percent_in_api(
    auth_client: AuthClientFactory, seeded: AsyncSession, debit: str, credit: str, percent: str
) -> None:
    world = await _world(seeded)
    await _source(seeded, world, credit=False, amount=debit)
    await _source(seeded, world, credit=True, amount=credit)
    response = await _request(await auth_client(KA_RF), world.version.id, "manual", world.selection)
    assert response.status_code == OK, response.text
    cell = next(c for c in response.json()["cells"] if c["company_a_id"] == str(world.members[0].id))
    assert cell["variance_percent"] == percent
    assert cell["exceeds_threshold"] is True
    assert cell["one_sided"] is False


@pytest.mark.parametrize("suffix", ["manual", "manual/export"])
@pytest.mark.parametrize("side", ["debit_account_ids", "credit_account_ids"])
@pytest.mark.parametrize("code", ["ACCOUNT_NOT_FOUND", "ACCOUNT_KIND_MISMATCH"])
async def test_invalid_account_selection(
    auth_client: AuthClientFactory, seeded: AsyncSession, suffix: str, side: str, code: str
) -> None:
    world = await _world(seeded)
    selection = world.selection
    wrong_kind = world.accounts["80" if side == "debit_account_ids" else "01"]
    selection[side] = [str(uuid.uuid4() if code == "ACCOUNT_NOT_FOUND" else wrong_kind.id)]
    response = await _request(await auth_client(KA_RF), world.version.id, suffix, selection)
    assert response.status_code == UNPROCESSABLE, response.text
    assert response.json()["detail"]["code"] == code


@pytest.mark.parametrize("suffix", ["result-table", "result-table/export"])
async def test_result_accounts_missing(auth_client: AuthClientFactory, seeded: AsyncSession, suffix: str) -> None:
    world = await _world(seeded)
    # Меняем код в тестовой транзакции, сохраняя FK существующих правил и потомков.
    world.accounts["58.03"].code = "58.03-missing"
    await seeded.flush()
    response = await _request(await auth_client(KA_RF), world.version.id, suffix, world.selection)
    assert response.status_code == UNPROCESSABLE, response.text
    assert response.json()["detail"]["code"] == "RESULT_ACCOUNTS_MISSING"
    assert "58.03" in response.json()["detail"]["message"]
    assert not (await seeded.scalars(select(AuditEvent).where(AuditEvent.action == AuditAction.EXPORT))).all()


async def test_result_table_export_matches_json(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
    world = await _world(seeded)
    await _source(seeded, world, credit=False, amount="100.25")
    client = await auth_client(KA_RF)
    result = await _request(client, world.version.id, "result-table", world.selection)
    export = await _request(client, world.version.id, "result-table/export", world.selection)
    assert result.status_code == export.status_code == OK
    data = result.json()
    sheet = load_workbook(BytesIO(export.content))["Результирующая таблица"]
    assert [sheet.cell(7, col).value for col in range(2, len(data["companies"]) + 2)] == [
        company["name"] for company in data["companies"]
    ]
    for row, article in enumerate(data["articles"], start=8):
        assert sheet.cell(row, 1).value == article["name"]
        amounts = {item["company_id"]: Decimal(item["amount"]) for item in article["amounts"]}
        for col, company in enumerate(data["companies"], start=2):
            assert Decimal(str(sheet.cell(row, col).value)) == amounts[company["id"]]
        assert Decimal(str(sheet.cell(row, len(data["companies"]) + 2).value)) == Decimal(article["group_total"])
    event = (await seeded.scalars(select(AuditEvent).where(AuditEvent.action == AuditAction.EXPORT))).one()
    assert event.after_value == {"report": "intercompany_result_table", "articles": len(data["articles"])}


@pytest.mark.parametrize("mode", ["all", "manual"])
async def test_export_filename_header_and_percentage(
    auth_client: AuthClientFactory, seeded: AsyncSession, mode: str
) -> None:
    world = await _world(seeded)
    await _source(seeded, world, credit=False, amount="100")
    await _source(seeded, world, credit=True, amount="90")
    group_name = (
        await seeded.scalars(select(CompanyGroup.name).where(CompanyGroup.id == world.members[0].company_group_id))
    ).one()
    response = await _request(await auth_client(KA_RF), world.version.id, f"{mode}/export", world.selection)
    assert response.status_code == OK, response.text
    disposition = response.headers["content-disposition"]
    assert unquote(disposition.split("filename*=UTF-8''", 1)[1]) == (
        f"intercompany-matrix_{group_name}_2024-Q1_v1.xlsx"
    )
    workbook = load_workbook(BytesIO(response.content))
    sheet = workbook["Шахматка"]
    assert [sheet[f"B{row}"].value for row in range(2, 9)] == [
        group_name,
        "2024-Q1",
        "v1",
        "руб.; расхождение — %",
        "Выбранные счета" if mode == "manual" else "Все счета",
        "62.01" if mode == "manual" else "Все счета",
        "60.01" if mode == "manual" else "Все счета",
    ]
    assert sheet["B14"].value == "Расхождение, %"
    expected_percent = Decimal("11.11")
    assert Decimal(str(sheet["D14"].value)) == expected_percent
    assert sheet["D14"].number_format == '0.00"%"'


@pytest.mark.parametrize("mode", ["all", "manual"])
async def test_export_appends_audit(auth_client: AuthClientFactory, seeded: AsyncSession, mode: str) -> None:
    world = await _world(seeded)
    client = await auth_client(KA_RF)
    query = select(AuditEvent).where(AuditEvent.action == AuditAction.EXPORT)
    for _ in range(2):
        response = await _request(client, world.version.id, f"{mode}/export", world.selection)
        assert response.status_code == OK, response.text
    events = (await seeded.scalars(query)).all()
    assert len(events) == 2  # noqa: PLR2004 - отдельное событие на каждую выгрузку
    assert events[0].id != events[1].id
    for event in events:
        assert event.section == AuditSection.EXPORT
        assert event.object_type == "consolidation_version"
        assert event.object_id == event.consolidation_version_id == world.version.id
        assert event.user_id is not None and event.before_value is None
        assert event.after_value == {
            "report": "intercompany_matrix",
            "mode": mode.upper(),
            "cells": len(world.members),
            **(world.selection if mode == "manual" else {"debit_account_ids": [], "credit_account_ids": []}),
        }