326 lines
14 KiB
Python
326 lines
14 KiB
Python
"""Выбор счетов и расчёт шахматки без обращения к БД."""
|
||
|
||
import datetime
|
||
import uuid
|
||
from decimal import Decimal
|
||
from types import SimpleNamespace
|
||
from unittest.mock import AsyncMock
|
||
|
||
import pytest
|
||
from fastapi import status as http_status
|
||
|
||
from src.api.endpoints.consolidations.intercompany_matrix import router
|
||
from src.models.enums import AccountKind, StatementType
|
||
from src.schemas.consolidations import (
|
||
IntercompanyMatrixCellOut,
|
||
IntercompanyMatrixCompanyOut,
|
||
IntercompanyMatrixOut,
|
||
IntercompanyResultTableOut,
|
||
)
|
||
from src.services import intercompany_matrix as matrix
|
||
|
||
|
||
def _account(kind: AccountKind, parent_id: uuid.UUID | None = None) -> SimpleNamespace:
|
||
return SimpleNamespace(id=uuid.uuid4(), kind=kind, parent_id=parent_id)
|
||
|
||
|
||
def _row(
|
||
owner: uuid.UUID,
|
||
counterparty: uuid.UUID,
|
||
account_id: uuid.UUID | None,
|
||
*,
|
||
debit: str = "0",
|
||
credit: str = "0",
|
||
) -> SimpleNamespace:
|
||
return SimpleNamespace(
|
||
legal_entity_id=owner,
|
||
counterparty_id=counterparty,
|
||
account_id=account_id,
|
||
closing_debit=Decimal(debit),
|
||
closing_credit=Decimal(credit),
|
||
turnover_debit=Decimal("0"),
|
||
turnover_credit=Decimal("0"),
|
||
entry_ids=(),
|
||
)
|
||
|
||
|
||
def test_routes_are_separate_and_matrix_expands_manual_accounts() -> None:
|
||
paths = {(route.path, tuple(sorted(route.methods))) for route in router.routes}
|
||
assert ("/consolidations/versions/{version_id}/intercompany-matrix/all", ("GET",)) in paths
|
||
assert ("/consolidations/versions/{version_id}/intercompany-matrix/manual", ("POST",)) in paths
|
||
assert ("/consolidations/versions/{version_id}/intercompany-matrix/preset/{rule_id}", ("GET",)) in paths
|
||
assert (
|
||
"/consolidations/versions/{version_id}/intercompany-matrix/preset/{rule_id}/unpaired-osv",
|
||
("GET",),
|
||
) in paths
|
||
assert ("/consolidations/versions/{version_id}/intercompany-matrix/result-table", ("GET",)) in paths
|
||
|
||
debit_parent = _account(AccountKind.A)
|
||
debit_child = _account(AccountKind.A, debit_parent.id)
|
||
credit_parent = _account(AccountKind.P)
|
||
credit_child = _account(AccountKind.P, credit_parent.id)
|
||
by_id, children = matrix._make_account_tree([debit_parent, debit_child, credit_parent, credit_child])
|
||
debit = matrix._select_accounts([debit_parent.id], [], "D", by_id, children)
|
||
credit = matrix._select_accounts([credit_parent.id], [], "K", by_id, children)
|
||
assert debit == {debit_parent.id, debit_child.id}
|
||
assert credit == {credit_parent.id, credit_child.id}
|
||
|
||
company_a, company_b = uuid.uuid4(), uuid.uuid4()
|
||
result = matrix.calculate_matrix(
|
||
[
|
||
_row(company_a, company_b, debit_child.id, debit="100"),
|
||
_row(company_b, company_a, credit_child.id, credit="90"),
|
||
],
|
||
[
|
||
IntercompanyMatrixCompanyOut(id=company_a, name="Компания А"),
|
||
IntercompanyMatrixCompanyOut(id=company_b, name="Компания Б"),
|
||
],
|
||
matrix.MatrixSelection(debit, credit, "closing"),
|
||
)
|
||
assert [(company.id, company.name) for company in result.companies] == [
|
||
(company_a, "Компания А"),
|
||
(company_b, "Компания Б"),
|
||
]
|
||
assert result.cells[0].amount_a == Decimal("100")
|
||
assert result.cells[0].amount_b == Decimal("90")
|
||
assert result.cells[0].variance_percent == Decimal("11.11")
|
||
assert isinstance(result, IntercompanyMatrixOut)
|
||
response = result
|
||
assert response.model_dump(mode="json")["companies"] == [
|
||
{"id": str(company_a), "name": "Компания А"},
|
||
{"id": str(company_b), "name": "Компания Б"},
|
||
]
|
||
|
||
|
||
def test_unpaired_rows_uses_source_presence_and_preserves_osv_order() -> None:
|
||
company_a, company_b = uuid.uuid4(), uuid.uuid4()
|
||
debit_account, credit_account = uuid.uuid4(), uuid.uuid4()
|
||
paired_debit = _row(company_a, company_b, debit_account, debit="100")
|
||
paired_credit = _row(company_b, company_a, credit_account, credit="100")
|
||
unpaired_credit = _row(company_a, company_b, credit_account, credit="0")
|
||
for row in (paired_debit, paired_credit, unpaired_credit):
|
||
row.entry_ids = (uuid.uuid4(),)
|
||
rows = [unpaired_credit, paired_debit, paired_credit]
|
||
companies = [
|
||
IntercompanyMatrixCompanyOut(id=company_a, name="А"),
|
||
IntercompanyMatrixCompanyOut(id=company_b, name="Б"),
|
||
]
|
||
result = matrix.calculate_matrix(
|
||
rows, companies, matrix.MatrixSelection(frozenset({debit_account}), frozenset({credit_account}), "closing")
|
||
)
|
||
|
||
assert matrix.get_unpaired_rows(rows, result) == [unpaired_credit]
|
||
|
||
|
||
async def test_preset_applies_exclusions_to_its_own_side_and_rule_measure(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
parent = _account(AccountKind.AP)
|
||
child = _account(AccountKind.AP, parent.id)
|
||
parent.code = "76"
|
||
child.code = "76.01"
|
||
version = SimpleNamespace(elimination_rules_version=1, version_no=2)
|
||
consolidation = SimpleNamespace(company_group_id=uuid.uuid4(), year=2024, quarter=matrix.PeriodQuarter.Q3)
|
||
osv_service = SimpleNamespace(
|
||
build_with_members=AsyncMock(
|
||
return_value=SimpleNamespace(
|
||
consolidated=SimpleNamespace(rows=[]), version=version, consolidation=consolidation, members={}
|
||
)
|
||
)
|
||
)
|
||
monkeypatch.setattr(matrix.directory_repo, "list_accounts", AsyncMock(return_value=[parent, child]))
|
||
rule = SimpleNamespace(
|
||
id=uuid.uuid4(),
|
||
rule_no=6,
|
||
statement_type=StatementType.balance,
|
||
version_no=1,
|
||
side_1="D",
|
||
measure="turnover",
|
||
valid_from=datetime.date(2024, 1, 1),
|
||
valid_to=None,
|
||
)
|
||
rule_accounts = [
|
||
SimpleNamespace(company_no=1, account_id=parent.id, is_excluded=False),
|
||
SimpleNamespace(company_no=1, account_id=child.id, is_excluded=True),
|
||
SimpleNamespace(company_no=2, account_id=parent.id, is_excluded=False),
|
||
]
|
||
monkeypatch.setattr(
|
||
matrix.elimination_rules_repo, "rule_with_account", AsyncMock(return_value=(rule, rule_accounts))
|
||
)
|
||
session = SimpleNamespace(execute=AsyncMock(return_value=SimpleNamespace(scalar_one=lambda: "ГК")))
|
||
service = matrix.IntercompanyMatrixService(session, None, osv_service)
|
||
captured = []
|
||
monkeypatch.setattr(service, "_calculate", lambda rows, members, selection: captured.append(selection))
|
||
|
||
await service.build_accounts(uuid.uuid4(), mode="PRESET", rule_id=uuid.uuid4())
|
||
|
||
assert len(captured) == 1
|
||
assert captured[0].company_1_account_ids == {parent.id}
|
||
assert captured[0].company_2_account_ids == {parent.id, child.id}
|
||
assert captured[0].measure == "turnover"
|
||
|
||
version.elimination_rules_version = None
|
||
monkeypatch.setattr(
|
||
matrix.elimination_rules_repo, "current_open_version_no", AsyncMock(return_value=rule.version_no)
|
||
)
|
||
_, info = await service.build_export(uuid.uuid4(), mode="PRESET", rule_id=uuid.uuid4())
|
||
assert info.mode == "PRESET"
|
||
assert info.debit_account_codes == ("76",)
|
||
assert info.credit_account_codes == ("76", "76.01")
|
||
assert info.measure == "turnover"
|
||
assert info.rule_no == rule.rule_no
|
||
assert info.rule_version_no == rule.version_no
|
||
|
||
|
||
async def test_preset_rejects_income_rule_before_selecting_accounts(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
rule = SimpleNamespace(
|
||
statement_type=StatementType.income,
|
||
version_no=1,
|
||
valid_to=None,
|
||
)
|
||
monkeypatch.setattr(matrix.elimination_rules_repo, "rule_with_account", AsyncMock(return_value=(rule, [])))
|
||
service = matrix.IntercompanyMatrixService(object(), None, object())
|
||
|
||
with pytest.raises(matrix.DomainValidationError) as error:
|
||
await service._resolve_preset_selection(uuid.uuid4(), SimpleNamespace(elimination_rules_version=1), {}, {})
|
||
|
||
assert error.value.status == http_status.HTTP_422_UNPROCESSABLE_CONTENT
|
||
assert error.value.code == "INVALID_RULE"
|
||
|
||
|
||
def test_one_sided_reflects_osv_sources_even_when_amount_is_zero() -> None:
|
||
company_a, company_b = uuid.uuid4(), uuid.uuid4()
|
||
debit_account, credit_account = uuid.uuid4(), uuid.uuid4()
|
||
companies = [
|
||
IntercompanyMatrixCompanyOut(id=company_a, name="Компания А"),
|
||
IntercompanyMatrixCompanyOut(id=company_b, name="Компания Б"),
|
||
]
|
||
selection = matrix.MatrixSelection(frozenset({debit_account}), frozenset({credit_account}), "closing")
|
||
debit = _row(company_a, company_b, debit_account)
|
||
credit = _row(company_b, company_a, credit_account)
|
||
debit.entry_ids = (uuid.uuid4(),)
|
||
credit.entry_ids = (uuid.uuid4(),)
|
||
|
||
for rows, expected in (([], False), ([debit], True), ([credit], True), ([debit, credit], False)):
|
||
cell = matrix.calculate_matrix(rows, companies, selection).cells[0]
|
||
assert cell.one_sided is expected
|
||
assert cell.model_dump(mode="json")["one_sided"] is expected
|
||
|
||
|
||
def test_unmapped_account_entries_are_counted_for_both_directions_of_pair() -> None:
|
||
company_a, company_b, outsider = uuid.uuid4(), uuid.uuid4(), uuid.uuid4()
|
||
account_id = uuid.uuid4()
|
||
missing_a = _row(company_a, company_b, None, debit="100")
|
||
missing_a.entry_ids = (uuid.uuid4(), uuid.uuid4())
|
||
missing_b = _row(company_b, company_a, None, credit="90")
|
||
missing_b.entry_ids = (uuid.uuid4(),)
|
||
outside = _row(company_a, outsider, None, debit="500")
|
||
outside.entry_ids = (uuid.uuid4(),)
|
||
matched = _row(company_a, company_b, account_id, debit="20")
|
||
matched.entry_ids = (uuid.uuid4(),)
|
||
companies = [
|
||
IntercompanyMatrixCompanyOut(id=company_a, name="Компания А"),
|
||
IntercompanyMatrixCompanyOut(id=company_b, name="Компания Б"),
|
||
]
|
||
result = matrix.calculate_matrix(
|
||
[missing_a, missing_b, outside, matched],
|
||
companies,
|
||
matrix.MatrixSelection(frozenset({account_id}), frozenset({account_id}), "closing"),
|
||
)
|
||
expected_skipped = len(missing_a.entry_ids) + len(missing_b.entry_ids)
|
||
assert [cell.skipped_osv_entry_count for cell in result.cells] == [expected_skipped, expected_skipped]
|
||
assert result.cells[0].model_dump(mode="json")["skipped_osv_entry_count"] == expected_skipped
|
||
assert result.cells[0].amount_a == Decimal("20")
|
||
assert result.cells[0].amount_b == Decimal("0")
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("debit", "expected", "exceeds"),
|
||
[
|
||
("111.125", "11.13", True),
|
||
("111.124", "11.12", True),
|
||
("105", "5.00", False),
|
||
("105.004", "5.00", True),
|
||
("104.999", "5.00", False),
|
||
("100", "0.00", False),
|
||
],
|
||
)
|
||
def test_percent_rounding_preserves_threshold(debit: str, expected: str, exceeds: bool) -> None:
|
||
company_a, company_b, account = uuid.uuid4(), uuid.uuid4(), uuid.uuid4()
|
||
result = matrix.calculate_matrix(
|
||
[_row(company_a, company_b, account, debit=debit), _row(company_b, company_a, account, credit="100")],
|
||
[
|
||
IntercompanyMatrixCompanyOut(id=company_a, name="Компания А"),
|
||
IntercompanyMatrixCompanyOut(id=company_b, name="Компания Б"),
|
||
],
|
||
matrix.MatrixSelection(frozenset({account}), frozenset({account}), "closing"),
|
||
)
|
||
assert result.cells[0].variance_percent == Decimal(expected)
|
||
assert result.cells[0].model_dump(mode="json")["variance_percent"] == expected
|
||
assert result.cells[0].exceeds_threshold is exceeds
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("value", "expected"),
|
||
[("0", "0.00"), ("1.005", "1.01"), ("-1.005", "-1.01"), ("3.873533480903112697872235289", "3.87")],
|
||
)
|
||
def test_cell_numbers_have_two_decimals(value: str, expected: str) -> None:
|
||
fields = ("amount_a", "amount_b", "variance_amount", "variance_percent")
|
||
cell = IntercompanyMatrixCellOut(
|
||
company_a_id=uuid.uuid4(),
|
||
company_b_id=uuid.uuid4(),
|
||
**dict.fromkeys(fields, value),
|
||
exceeds_threshold=False,
|
||
one_sided=False,
|
||
skipped_osv_entry_count=0,
|
||
osv_entry_ids_debit=[],
|
||
osv_entry_ids_credit=[],
|
||
)
|
||
assert {field: cell.model_dump(mode="json")[field] for field in fields} == dict.fromkeys(fields, expected)
|
||
|
||
|
||
def test_result_table_uses_fixed_articles_and_group_totals() -> None:
|
||
company_a, company_b, external = uuid.uuid4(), uuid.uuid4(), uuid.uuid4()
|
||
required_codes = {
|
||
code for article in matrix.RESULT_ARTICLES for code in (*article.account_codes, *article.excluded_codes)
|
||
}
|
||
kinds = {
|
||
code: AccountKind.A if article.side == "D" else AccountKind.P
|
||
for article in matrix.RESULT_ARTICLES
|
||
for code in article.account_codes
|
||
}
|
||
kinds.update({"76": AccountKind.AP, "76.07.1": AccountKind.P, "76.07.5": AccountKind.A, "76.07.9": AccountKind.A})
|
||
ids = {code: uuid.uuid4() for code in required_codes}
|
||
parents = {"67.03": "67", "66.03": "66", "66.04": "66", "76.07.1": "76", "76.07.5": "76", "76.07.9": "76"}
|
||
accounts = [
|
||
SimpleNamespace(id=ids[code], code=code, kind=kinds[code], parent_id=ids.get(parents.get(code)))
|
||
for code in required_codes
|
||
]
|
||
rows = [
|
||
_row(company_a, company_b, ids["60.02"], debit="100"),
|
||
_row(company_a, company_b, ids["76.07.5"], debit="50"),
|
||
_row(company_a, company_b, ids["76.07.1"], credit="20"),
|
||
_row(company_a, company_b, ids["58.01"], debit="30"),
|
||
_row(company_b, company_a, ids["60.01"], credit="70"),
|
||
_row(company_b, company_a, ids["67.03"], credit="40"),
|
||
_row(company_a, company_b, ids["66.04"], credit="10"),
|
||
_row(company_a, external, ids["60.02"], debit="999"),
|
||
]
|
||
result = matrix.calculate_result_table(
|
||
rows,
|
||
[
|
||
IntercompanyMatrixCompanyOut(id=company_a, name="Компания А"),
|
||
IntercompanyMatrixCompanyOut(id=company_b, name="Компания Б"),
|
||
],
|
||
accounts,
|
||
)
|
||
articles = {article.code: article for article in result.articles}
|
||
assert list(articles) == [article.code for article in matrix.RESULT_ARTICLES]
|
||
assert [item.amount for item in articles["receivables"].amounts] == [Decimal("100"), Decimal("0")]
|
||
assert articles["receivables"].group_total == Decimal("100")
|
||
assert articles["payables"].group_total == Decimal("70")
|
||
assert articles["financial_investments"].group_total == Decimal("30")
|
||
assert articles["long_term_loans"].group_total == Decimal("40")
|
||
assert articles["short_term_loans"].group_total == Decimal("10")
|
||
assert isinstance(result, IntercompanyResultTableOut)
|
||
response = result
|
||
assert response.articles[0].amounts[0].company_id == company_a
|