280 lines
13 KiB
Python
280 lines
13 KiB
Python
"""Группировка найденных пар ВГО и единственность исходных строк во всей версии."""
|
||
|
||
import uuid
|
||
from decimal import Decimal
|
||
from types import SimpleNamespace
|
||
from unittest.mock import AsyncMock
|
||
|
||
import pytest
|
||
|
||
from src.common.exceptions import DomainValidationError
|
||
from src.domain.consolidated_osv import CounterpartyType
|
||
from src.models.enums import StatementType
|
||
from src.schemas.consolidations import EliminationMatchedPairOut
|
||
from src.services import vgo
|
||
|
||
|
||
def _pair( # noqa: PLR0913 - фабрика тестовой пары с настраиваемой суммой
|
||
rule_id: uuid.UUID,
|
||
company_a: uuid.UUID,
|
||
company_b: uuid.UUID,
|
||
entry_a: uuid.UUID,
|
||
entry_b: uuid.UUID,
|
||
*,
|
||
amount_b: str = "10",
|
||
) -> EliminationMatchedPairOut:
|
||
deviation = Decimal("10") - Decimal(amount_b)
|
||
return EliminationMatchedPairOut(
|
||
rule_id=rule_id,
|
||
company_a_id=company_a,
|
||
company_b_id=company_b,
|
||
amount_a=Decimal("10"),
|
||
amount_b=Decimal(amount_b),
|
||
deviation_amount=deviation,
|
||
deviation_percent=deviation / Decimal(amount_b) * 100,
|
||
osv_entry_ids_a=[entry_a],
|
||
osv_entry_ids_b=[entry_b],
|
||
)
|
||
|
||
|
||
def _row(entry_id: uuid.UUID, company_a: uuid.UUID, company_b: uuid.UUID, index: int) -> SimpleNamespace:
|
||
return SimpleNamespace(
|
||
entry_ids=(entry_id,),
|
||
legal_entity_id=company_a,
|
||
legal_entity_name="A",
|
||
legal_entity_inn="1234567890",
|
||
account_code="76",
|
||
account_root="76",
|
||
period="2026 Q1",
|
||
is_cumulative=True,
|
||
row_number=index,
|
||
counterparty_name="B",
|
||
counterparty_inn="0987654321",
|
||
counterparty_id=company_b,
|
||
counterparty_entity_name="B",
|
||
counterparty_entity_inn="0987654321",
|
||
counterparty_type=CounterpartyType.GSK,
|
||
match_kind=None,
|
||
match_stored=True,
|
||
opening_debit=Decimal("0"),
|
||
opening_credit=Decimal("0"),
|
||
turnover_debit=Decimal("0"),
|
||
turnover_credit=Decimal("0"),
|
||
closing_debit=Decimal("1"),
|
||
closing_credit=Decimal("0"),
|
||
warnings=(),
|
||
)
|
||
|
||
|
||
def _members(company_a: uuid.UUID, company_b: uuid.UUID) -> dict[uuid.UUID, SimpleNamespace]:
|
||
return {company_a: SimpleNamespace(name="A"), company_b: SimpleNamespace(name="B")}
|
||
|
||
|
||
async def test_pairs_are_grouped_and_entry_ids_are_used_once(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
version_id, company_a, company_b, rule_1, rule_2 = (uuid.uuid4() for _ in range(5))
|
||
entry_ids = [uuid.uuid4() for _ in range(11)]
|
||
rows = [_row(entry_id, company_a, company_b, index) for index, entry_id in enumerate(entry_ids)]
|
||
osv_result = SimpleNamespace(
|
||
version=SimpleNamespace(elimination_rules_version=2),
|
||
consolidated=SimpleNamespace(rows=rows),
|
||
members=_members(company_a, company_b),
|
||
)
|
||
rules = [
|
||
SimpleNamespace(
|
||
id=rule_1, rule_no=1, statement_type=StatementType.balance, measure="closing",
|
||
article_1_line="Статья A1", article_2_line="Статья B1",
|
||
),
|
||
SimpleNamespace(
|
||
id=rule_2, rule_no=2, statement_type=StatementType.balance, measure="closing",
|
||
article_1_line="Статья A2", article_2_line="Статья B2",
|
||
),
|
||
]
|
||
monkeypatch.setattr(vgo.elimination_rules_repo, "list_rules", AsyncMock(return_value=rules))
|
||
monkeypatch.setattr(vgo.vgo_assignments_repo, "list_assignments", AsyncMock(return_value=[]))
|
||
exact_1 = _pair(rule_1, company_a, company_b, entry_ids[0], entry_ids[1])
|
||
exact_1_more = _pair(rule_1, company_a, company_b, entry_ids[9], entry_ids[10])
|
||
exact_conflict = _pair(rule_2, company_a, company_b, entry_ids[0], entry_ids[2])
|
||
exact_2 = _pair(rule_2, company_b, company_a, entry_ids[2], entry_ids[3])
|
||
discrepancy_1 = _pair(rule_1, company_a, company_b, entry_ids[4], entry_ids[5])
|
||
discrepancy_conflict = _pair(rule_2, company_b, company_a, entry_ids[5], entry_ids[4])
|
||
discrepancy_unique = _pair(rule_1, company_b, company_a, entry_ids[7], entry_ids[8], amount_b="9")
|
||
|
||
async def build_pairs(
|
||
_version: uuid.UUID, rule: uuid.UUID, a: uuid.UUID, b: uuid.UUID, available: list, **_kwargs: object
|
||
) -> SimpleNamespace:
|
||
if (rule, a, b) == (rule_1, company_a, company_b):
|
||
return SimpleNamespace(rule_no=1, matched=[exact_1, exact_1_more], with_discrepancy=[discrepancy_1])
|
||
if (rule, a, b) == (rule_2, company_a, company_b):
|
||
assert entry_ids[0] not in {row.entry_ids[0] for row in available}
|
||
return SimpleNamespace(rule_no=2, matched=[exact_conflict], with_discrepancy=[])
|
||
if (rule, a, b) == (rule_1, company_b, company_a):
|
||
return SimpleNamespace(rule_no=1, matched=[], with_discrepancy=[discrepancy_unique])
|
||
if (rule, a, b) == (rule_2, company_b, company_a):
|
||
return SimpleNamespace(rule_no=2, matched=[exact_2], with_discrepancy=[discrepancy_conflict])
|
||
return SimpleNamespace(rule_no=1, matched=[], with_discrepancy=[])
|
||
|
||
service = vgo.VgoService(
|
||
AsyncMock(),
|
||
SimpleNamespace(build_with_members=AsyncMock(return_value=osv_result)),
|
||
SimpleNamespace(build_pairs=AsyncMock(side_effect=build_pairs)),
|
||
)
|
||
response = await service.assignment_candidates(version_id)
|
||
groups = response.groups
|
||
|
||
assert [(group.rule_id, group.company_a_id, group.company_b_id) for group in groups] == [
|
||
(rule_1, company_a, company_b),
|
||
(rule_2, company_b, company_a),
|
||
(rule_1, company_b, company_a),
|
||
]
|
||
assert groups[0].matched == [exact_1, exact_1_more]
|
||
assert (groups[0].article_1_line, groups[0].article_2_line) == ("Статья A1", "Статья B1")
|
||
assert (groups[0].company_a_name, groups[0].company_b_name) == ("A", "B")
|
||
assert (groups[0].amount_a, groups[0].amount_b, groups[0].deviation_amount) == (
|
||
Decimal("20.00"),
|
||
Decimal("20.00"),
|
||
Decimal("0.00"),
|
||
)
|
||
assert groups[0].with_discrepancy == []
|
||
assert "manual" not in response.model_dump()["groups"][0]
|
||
assert groups[1].matched == [exact_2]
|
||
assert groups[1].with_discrepancy == []
|
||
assert groups[2].matched == []
|
||
assert groups[2].with_discrepancy == [discrepancy_unique]
|
||
assert (groups[2].amount_a, groups[2].amount_b, groups[2].deviation_amount, groups[2].deviation_percent) == (
|
||
Decimal("10.00"),
|
||
Decimal("9.00"),
|
||
Decimal("1.00"),
|
||
Decimal("11.11"),
|
||
)
|
||
assert [row.entry_ids for row in response.unassigned] == [
|
||
[entry_ids[4]],
|
||
[entry_ids[5]],
|
||
[entry_ids[6]],
|
||
]
|
||
|
||
|
||
async def test_manual_assignments_override_auto_and_keep_exclusion_flag(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
version_id, company_a, company_b, rule_id = (uuid.uuid4() for _ in range(4))
|
||
entry_ids = [uuid.uuid4() for _ in range(4)]
|
||
rows = [_row(entry_id, company_a, company_b, index) for index, entry_id in enumerate(entry_ids)]
|
||
for row, amount in zip(rows, ("10", "20", "5", "1"), strict=True):
|
||
row.closing_debit = Decimal(amount)
|
||
osv_result = SimpleNamespace(
|
||
version=SimpleNamespace(elimination_rules_version=2),
|
||
consolidated=SimpleNamespace(rows=rows),
|
||
members=_members(company_a, company_b),
|
||
)
|
||
rule = SimpleNamespace(
|
||
id=rule_id, rule_no=1, statement_type=StatementType.balance, measure="closing", side_1="D",
|
||
article_1_line="Статья A", article_2_line="Статья B",
|
||
)
|
||
assignments = [
|
||
SimpleNamespace(osv_entry_id=entry_ids[0], rule_id=rule_id, action="ASSIGN", is_exclude=False),
|
||
SimpleNamespace(osv_entry_id=entry_ids[1], rule_id=rule_id, action="ASSIGN", is_exclude=True),
|
||
SimpleNamespace(osv_entry_id=entry_ids[2], rule_id=None, action="UNASSIGNED", is_exclude=True),
|
||
]
|
||
monkeypatch.setattr(vgo.elimination_rules_repo, "list_rules", AsyncMock(return_value=[rule]))
|
||
monkeypatch.setattr(vgo.vgo_assignments_repo, "list_assignments", AsyncMock(return_value=assignments))
|
||
|
||
async def build_pairs(
|
||
_version: uuid.UUID, _rule: uuid.UUID, _a: uuid.UUID, _b: uuid.UUID, available: list, **_kwargs: object
|
||
) -> SimpleNamespace:
|
||
assert [row.entry_ids[0] for row in available] == [entry_ids[3]]
|
||
return SimpleNamespace(matched=[], with_discrepancy=[])
|
||
|
||
service = vgo.VgoService(
|
||
AsyncMock(),
|
||
SimpleNamespace(build_with_members=AsyncMock(return_value=osv_result)),
|
||
SimpleNamespace(build_pairs=AsyncMock(side_effect=build_pairs)),
|
||
)
|
||
response = await service.assignment_candidates(version_id)
|
||
|
||
assert len(response.groups) == 1
|
||
assert (response.groups[0].rule_id, response.groups[0].company_a_id, response.groups[0].company_b_id) == (
|
||
rule_id,
|
||
company_a,
|
||
company_b,
|
||
)
|
||
assert (response.groups[0].amount_a, response.groups[0].amount_b) == (Decimal("10.00"), Decimal("0.00"))
|
||
assert [(row.osv_entry_id, row.is_exclude) for row in response.groups[0].manual] == [
|
||
(entry_ids[0], False),
|
||
(entry_ids[1], True),
|
||
]
|
||
assert "manual" in response.model_dump()["groups"][0]
|
||
assert [(row.entry_ids, row.action, row.is_exclude) for row in response.unassigned] == [
|
||
([entry_ids[2]], "UNASSIGNED", True),
|
||
([entry_ids[3]], None, None),
|
||
]
|
||
|
||
|
||
async def test_glued_row_requires_assignments_for_all_source_entries(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
version_id, company_a, company_b, rule_id, entry_a, entry_b = (uuid.uuid4() for _ in range(6))
|
||
row = _row(entry_a, company_a, company_b, 1)
|
||
row.entry_ids = (entry_a, entry_b)
|
||
osv_result = SimpleNamespace(
|
||
version=SimpleNamespace(elimination_rules_version=2),
|
||
consolidated=SimpleNamespace(rows=[row]),
|
||
members=_members(company_a, company_b),
|
||
)
|
||
assignment = SimpleNamespace(osv_entry_id=entry_a, rule_id=rule_id, action="ASSIGN", is_exclude=False)
|
||
monkeypatch.setattr(vgo.elimination_rules_repo, "list_rules", AsyncMock(return_value=[]))
|
||
monkeypatch.setattr(vgo.vgo_assignments_repo, "list_assignments", AsyncMock(return_value=[assignment]))
|
||
service = vgo.VgoService(
|
||
AsyncMock(),
|
||
SimpleNamespace(build_with_members=AsyncMock(return_value=osv_result)),
|
||
SimpleNamespace(build_pairs=AsyncMock()),
|
||
)
|
||
|
||
with pytest.raises(DomainValidationError, match="для остальных нет"):
|
||
await service.assignment_candidates(version_id)
|
||
|
||
|
||
async def test_glued_row_with_same_assignments_is_counted_once(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
version_id, company_a, company_b, rule_id, entry_a, entry_b = (uuid.uuid4() for _ in range(6))
|
||
row = _row(entry_a, company_a, company_b, 1)
|
||
row.entry_ids = (entry_a, entry_b)
|
||
row.closing_debit = Decimal("10")
|
||
osv_result = SimpleNamespace(
|
||
version=SimpleNamespace(elimination_rules_version=2),
|
||
consolidated=SimpleNamespace(rows=[row]),
|
||
members=_members(company_a, company_b),
|
||
)
|
||
rule = SimpleNamespace(
|
||
id=rule_id, rule_no=1, statement_type=StatementType.balance, measure="closing", side_1="D",
|
||
article_1_line="Статья A", article_2_line="Статья B",
|
||
)
|
||
assignments = [
|
||
SimpleNamespace(osv_entry_id=entry_id, rule_id=rule_id, action="ASSIGN", is_exclude=False)
|
||
for entry_id in (entry_a, entry_b)
|
||
]
|
||
monkeypatch.setattr(vgo.elimination_rules_repo, "list_rules", AsyncMock(return_value=[rule]))
|
||
monkeypatch.setattr(vgo.vgo_assignments_repo, "list_assignments", AsyncMock(return_value=assignments))
|
||
empty_result = SimpleNamespace(matched=[], with_discrepancy=[])
|
||
matrix_service = SimpleNamespace(build_pairs=AsyncMock(return_value=empty_result))
|
||
service = vgo.VgoService(
|
||
AsyncMock(), SimpleNamespace(build_with_members=AsyncMock(return_value=osv_result)), matrix_service
|
||
)
|
||
|
||
response = await service.assignment_candidates(version_id)
|
||
|
||
assert len(response.groups) == 1
|
||
assert response.groups[0].amount_a == Decimal("10.00")
|
||
assert len(response.groups[0].manual) == 1
|
||
assert response.groups[0].manual[0].entry_ids == [entry_a, entry_b]
|
||
assert response.groups[0].manual[0].osv_entry_id is None
|
||
assert response.unassigned == []
|
||
assert all(call.args[4] == [] for call in matrix_service.build_pairs.await_args_list)
|
||
|
||
|
||
def test_glued_row_rejects_conflicting_manual_assignments() -> None:
|
||
company_a, company_b, rule_id, entry_a, entry_b = (uuid.uuid4() for _ in range(5))
|
||
row = _row(entry_a, company_a, company_b, 1)
|
||
row.entry_ids = (entry_a, entry_b)
|
||
assignments = [
|
||
SimpleNamespace(osv_entry_id=entry_a, rule_id=rule_id, action="ASSIGN", is_exclude=False),
|
||
SimpleNamespace(osv_entry_id=entry_b, rule_id=None, action="UNASSIGNED", is_exclude=False),
|
||
]
|
||
|
||
with pytest.raises(DomainValidationError, match="разные ручные назначения"):
|
||
vgo._manual_assignments([row], assignments, [])
|