311 lines
15 KiB
Python
311 lines
15 KiB
Python
"""Критерии шахматки через 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
|
||
|
||
|
||
async def test_result_accounts_missing(auth_client: AuthClientFactory, seeded: AsyncSession) -> 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, "result-table", 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"]
|
||
|
||
|
||
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": []}),
|
||
}
|