66 lines
2.8 KiB
Python
66 lines
2.8 KiB
Python
"""Инвалидация корректировок затронутой пары без полного пересчёта распределения VGO."""
|
|
|
|
import uuid
|
|
from decimal import Decimal
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from src.models.enums import CorrectionStatus
|
|
from src.models.kfo import Correction
|
|
from src.services import correction_invalidation
|
|
|
|
|
|
def _correction(company_a: uuid.UUID, company_b: uuid.UUID, status: CorrectionStatus) -> Correction:
|
|
return Correction(
|
|
id=uuid.uuid4(),
|
|
legal_entity_id=company_a,
|
|
counterparty_id=company_b,
|
|
debit_line_code="1230",
|
|
debit_adjustment=Decimal("10"),
|
|
comment="Причина расхождения",
|
|
status=status,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_all_applied_corrections_of_affected_pair_become_draft(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
version_id, company_a, company_b, company_c = (uuid.uuid4() for _ in range(4))
|
|
first_rule = _correction(company_a, company_b, CorrectionStatus.APPLIED)
|
|
second_rule = _correction(company_b, company_a, CorrectionStatus.APPLIED)
|
|
already_draft = _correction(company_a, company_b, CorrectionStatus.DRAFT)
|
|
unrelated = _correction(company_a, company_c, CorrectionStatus.APPLIED)
|
|
monkeypatch.setattr(
|
|
correction_invalidation.corrections_repo,
|
|
"active_corrections",
|
|
AsyncMock(return_value=[first_rule, second_rule, already_draft, unrelated]),
|
|
)
|
|
session = SimpleNamespace(flush=AsyncMock())
|
|
audit = SimpleNamespace(user_id=uuid.uuid4(), record=AsyncMock())
|
|
service = correction_invalidation.CorrectionInvalidationService(session, audit)
|
|
|
|
changed = await service.mark_draft(version_id, {frozenset((company_a, company_b))})
|
|
|
|
assert set(changed) == {first_rule.id, second_rule.id}
|
|
assert first_rule.status == second_rule.status == CorrectionStatus.DRAFT
|
|
assert already_draft.status == CorrectionStatus.DRAFT
|
|
assert unrelated.status == CorrectionStatus.APPLIED
|
|
assert first_rule.debit_adjustment == Decimal("10")
|
|
assert first_rule.comment == "Причина расхождения"
|
|
session.flush.assert_awaited_once()
|
|
audit.record.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_affected_pairs_are_read_directly_from_osv_entries() -> None:
|
|
owner_id, counterparty_id, entry_id = uuid.uuid4(), uuid.uuid4(), uuid.uuid4()
|
|
query_result = SimpleNamespace(all=lambda: [(owner_id, counterparty_id), (owner_id, None)])
|
|
session = SimpleNamespace(execute=AsyncMock(return_value=query_result))
|
|
service = correction_invalidation.CorrectionInvalidationService(session, SimpleNamespace(user_id=uuid.uuid4()))
|
|
|
|
pairs = await service.affected_pairs([entry_id])
|
|
|
|
assert pairs == {frozenset((owner_id, counterparty_id))}
|
|
session.execute.assert_awaited_once()
|