"""Выбор счетов и показателя в режимах шахматки без обращения к БД.""" import uuid from decimal import Decimal from types import SimpleNamespace from unittest.mock import AsyncMock import pytest from src.api.endpoints.consolidations.intercompany_matrix import router from src.common.exceptions import DomainValidationError from src.models.enums import AccountKind, StatementType from src.schemas.consolidations import 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, *, 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"), ) 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/result-table", ("GET",)) in paths assert ( "/consolidations/versions/{version_id}/intercompany-matrix/presets/{rule_no}", ("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._account_tree([debit_parent, debit_child, credit_parent, credit_child]) debit = matrix._selected_accounts([debit_parent.id], [], "D", by_id, children) credit = matrix._selected_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", "D"), ) 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("100") / Decimal("9") assert isinstance(result, IntercompanyMatrixOut) response = result assert "exceeds_threshold" not in response.model_dump(mode="json")["cells"][0] assert response.model_dump(mode="json")["companies"] == [ {"id": str(company_a), "name": "Компания А"}, {"id": str(company_b), "name": "Компания Б"}, ] 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 @pytest.mark.asyncio async def test_preset_uses_rule_version_measure_side_and_exclusions(monkeypatch: pytest.MonkeyPatch) -> None: version_id, company_a, company_b = uuid.uuid4(), uuid.uuid4(), uuid.uuid4() debit_root = _account(AccountKind.A) debit_excluded = _account(AccountKind.A, debit_root.id) credit_root = _account(AccountKind.P) rows = [ _row(company_a, company_b, credit_root.id, credit="999"), _row(company_b, company_a, debit_root.id, debit="999"), _row(company_b, company_a, debit_excluded.id, debit="999"), ] rows[0].turnover_credit = Decimal("80") rows[1].turnover_debit = Decimal("70") rows[2].turnover_debit = Decimal("20") version = SimpleNamespace(id=version_id, elimination_rules_version=2) osv = SimpleNamespace(build=AsyncMock(return_value=(SimpleNamespace(rows=rows), version, None))) session = SimpleNamespace( scalars=AsyncMock(return_value=SimpleNamespace(all=lambda: [debit_root, debit_excluded, credit_root])) ) rule = SimpleNamespace( id=uuid.uuid4(), statement_type=StatementType.income, rule_no=3, measure="turnover", side_1="K" ) rule_accounts = [ SimpleNamespace(company_no=1, account_id=credit_root.id, is_excluded=False), SimpleNamespace(company_no=2, account_id=debit_root.id, is_excluded=False), SimpleNamespace(company_no=2, account_id=debit_excluded.id, is_excluded=True), ] list_rules = AsyncMock(return_value=[(rule, account) for account in rule_accounts]) members = AsyncMock( return_value=[ (None, SimpleNamespace(id=company_a, name="Компания А")), (None, SimpleNamespace(id=company_b, name="Компания Б")), ] ) monkeypatch.setattr(matrix.rules_repo, "list_rules_with_accounts", list_rules) monkeypatch.setattr(matrix.repo, "members", members) service = matrix.IntercompanyMatrixService(session, SimpleNamespace(), osv) result = await service.build_preset(version_id, 3) list_rules.assert_awaited_once_with(session, version_id) assert result.cells[0].amount_a == Decimal("80") assert result.cells[0].amount_b == Decimal("70") assert result.cells[0].variance_amount == Decimal("10") assert [company.name for company in result.companies] == ["Компания А", "Компания Б"] duplicate = SimpleNamespace(id=uuid.uuid4(), statement_type=StatementType.balance, rule_no=3) list_rules.return_value = [(rule, rule_accounts[0]), (duplicate, None)] with pytest.raises(DomainValidationError, match="Номер правила не уникален"): await service.build_preset(version_id, 3)