arm-core/api/tests/test_schemas_consolidations.py
Andrey Epifancev f46cf2fec5 feat(consolidation): чек-лист комплекта документов по правилам подхода
Домен `checklist` (Rule, CellStatus, Cell, order_rules, build) — чистые функции без ORM; ячейка (ЮЛ, правило):
CONFIRMED по привязанным документам (ОСВ: нарастающий за квартал версии либо квартальные Q1..Qn, пробел →
missing_quarters; ББ/ОФР: документ за квартал версии), NEEDS_CHECK/RECOGNIZED по неподтверждённому подходящему,
PROCESSING, ERROR, иначе MISSING; правило ОСВ закрывает только точный код счёта. Репозиторий: rules_for,
unconfirmed_documents. Сервис: checklist(), ячейки в карточке версии. Маршрут GET /consolidations/versions/{id}/checklist.
Схема использует CellStatus домена (T2.8). Решения №44-49 в design.md.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
2026-10-06 05:50:48 +04:00

92 lines
3.4 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Схемы консолидаций (Срез 4, D-16): границы валидации, суммы-строки в JSON, неизменяемость доменного состояния."""
import dataclasses
import uuid
from decimal import Decimal
import pytest
from pydantic import ValidationError
from src.domain.checklist import CellStatus, EntityFilesState
from src.models.enums import CounterpartyMatchKind, FileStatus, PeriodQuarter
from src.schemas.consolidations import (
ChecklistCellOut,
ConsolidatedRowOut,
ConsolidationCreate,
MapCounterpartyIn,
OsvFilters,
StatusIn,
)
MAX_LIMIT = 2000
def test_consolidation_create_year_bounds() -> None:
ids = {"company_group_id": uuid.uuid4(), "related_group_id": uuid.uuid4(), "quarter": "Q3"}
assert ConsolidationCreate(year=2024, **ids).quarter == PeriodQuarter.Q3
for year in (1999, 2101):
with pytest.raises(ValidationError):
ConsolidationCreate(year=year, **ids)
def test_osv_filters_limit_bounds_and_repeated_match_kind() -> None:
assert OsvFilters().limit == 500 # noqa: PLR2004
assert OsvFilters(limit=MAX_LIMIT).limit == MAX_LIMIT
for bad in ({"limit": MAX_LIMIT + 1}, {"limit": 0}, {"offset": -1}, {"counterparty_type": "OTHER"}):
with pytest.raises(ValidationError):
OsvFilters(**bad)
assert OsvFilters(match_kind=["NAME", "ALIAS"]).match_kind == [
CounterpartyMatchKind.NAME,
CounterpartyMatchKind.ALIAS,
]
def test_defaults_of_input_schemas() -> None:
assert MapCounterpartyIn(legal_entity_id=uuid.uuid4()).move_alias is False
assert StatusIn(to_status="CANCELLED").comment is None
def test_consolidated_row_serializes_money_as_strings() -> None:
row = ConsolidatedRowOut(
entry_ids=[uuid.uuid4()],
legal_entity_id=uuid.uuid4(),
legal_entity_name="ООО",
legal_entity_inn="7701000001",
account_code="60",
account_root="60",
period="2024 Q3",
is_cumulative=False,
row_number=1,
counterparty_name="А",
counterparty_inn=None,
counterparty_id=None,
counterparty_entity_name=None,
counterparty_type="EXTERNAL",
match_kind=None,
match_stored=False,
opening_debit=Decimal("100.50"),
opening_credit=None,
turnover_debit=Decimal("0.00"),
turnover_credit=None,
closing_debit=None,
closing_credit=None,
warnings=[],
)
dumped = row.model_dump(mode="json")
assert (dumped["opening_debit"], dumped["opening_credit"], dumped["turnover_debit"]) == ("100.50", None, "0.00")
def test_entity_files_state_is_frozen() -> None:
state = EntityFilesState(has_in_progress=True, last_file_status=FileStatus.ERROR)
with pytest.raises(dataclasses.FrozenInstanceError):
state.has_in_progress = False # type: ignore[misc]
def test_checklist_cell_status_is_the_domain_enum() -> None:
"""Статус ячейки в схеме — `domain.checklist.CellStatus` (одно определение, T2.8): все значения принимаются."""
for status in CellStatus:
cell = ChecklistCellOut(status=status.value, missing_quarters=[], document_ids=[]) # type: ignore[arg-type]
assert cell.status is status
with pytest.raises(ValidationError):
ChecklistCellOut(status="UNKNOWN", missing_quarters=[], document_ids=[]) # type: ignore[arg-type]