198 lines
8.3 KiB
Python
198 lines
8.3 KiB
Python
"""Схемы консолидаций (Срез 4, D-16): границы валидации, суммы-строки в JSON, неизменяемость доменного состояния."""
|
||
|
||
import dataclasses
|
||
import uuid
|
||
from decimal import Decimal
|
||
from typing import Any
|
||
|
||
import pytest
|
||
from pydantic import ValidationError
|
||
|
||
from src.domain.checklist import CellStatus, EntityFilesState
|
||
from src.domain.consolidated_osv import CounterpartyType as DomainCounterpartyType
|
||
from src.models.enums import CounterpartyMatchKind, DocumentType, FileStatus, PeriodQuarter
|
||
from src.schemas import consolidations as schemas
|
||
from src.schemas.consolidations import (
|
||
MAX_MEMBERS,
|
||
MAX_OSV_QUERY,
|
||
MAX_STATUS_COMMENT,
|
||
ChecklistCellOut,
|
||
ConsolidatedRowOut,
|
||
ConsolidationCreate,
|
||
MapCounterpartyIn,
|
||
MembersIn,
|
||
OsvFilters,
|
||
OsvPageParams,
|
||
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_consolidation_create_selected_perimeter_bounds() -> None:
|
||
"""D-1: `legal_entity_ids` необязателен (`None` — периметр из ГСК), не больше `MAX_MEMBERS`."""
|
||
ids = {"company_group_id": uuid.uuid4(), "related_group_id": uuid.uuid4(), "year": 2024, "quarter": "Q3"}
|
||
assert ConsolidationCreate(**ids).legal_entity_ids is None # type: ignore[arg-type]
|
||
selected = [uuid.uuid4() for _ in range(MAX_MEMBERS)]
|
||
assert len(ConsolidationCreate(**ids, legal_entity_ids=selected).legal_entity_ids or []) == MAX_MEMBERS # type: ignore[arg-type]
|
||
with pytest.raises(ValidationError):
|
||
ConsolidationCreate(**ids, legal_entity_ids=[*selected, uuid.uuid4()]) # type: ignore[arg-type]
|
||
|
||
|
||
def test_osv_page_params_limit_bounds() -> None:
|
||
assert (OsvPageParams().limit, OsvPageParams().offset) == (500, 0) # noqa: PLR2004
|
||
assert OsvPageParams(limit=MAX_LIMIT).limit == MAX_LIMIT
|
||
for bad in ({"limit": MAX_LIMIT + 1}, {"limit": 0}, {"offset": -1}):
|
||
with pytest.raises(ValidationError):
|
||
OsvPageParams(**bad)
|
||
|
||
|
||
def test_osv_filters_repeated_match_kind_and_no_pagination() -> None:
|
||
"""Фильтры — отдельная модель от пагинации (T2.7): `limit`/`offset` в ней нет."""
|
||
assert "limit" not in OsvFilters.model_fields
|
||
assert "offset" not in OsvFilters.model_fields
|
||
with pytest.raises(ValidationError):
|
||
OsvFilters(counterparty_type="OTHER") # type: ignore[arg-type]
|
||
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 _row(**amounts: Decimal | None) -> ConsolidatedRowOut:
|
||
fields: dict[str, Any] = {
|
||
"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_entity_inn": None,
|
||
"counterparty_type": "EXTERNAL",
|
||
"match_kind": None,
|
||
"match_stored": False,
|
||
"opening_debit": None,
|
||
"opening_credit": None,
|
||
"turnover_debit": None,
|
||
"turnover_credit": None,
|
||
"closing_debit": None,
|
||
"closing_credit": None,
|
||
"warnings": [],
|
||
}
|
||
return ConsolidatedRowOut(**{**fields, **amounts})
|
||
|
||
|
||
def test_consolidated_row_serializes_money_as_strings() -> None:
|
||
row = _row(opening_debit=Decimal("100.50"), turnover_debit=Decimal("0.00"))
|
||
dumped = row.model_dump(mode="json")
|
||
assert (dumped["opening_debit"], dumped["opening_credit"], dumped["turnover_debit"]) == ("100.50", None, "0.00")
|
||
|
||
|
||
def test_consolidated_row_money_has_two_decimals() -> None:
|
||
"""Суммы свода — всегда с двумя знаками (`"0.00"`, не `"0"`): ноль склейки и сумма Decimal не меняют вид."""
|
||
row = _row(opening_debit=Decimal("0"), turnover_debit=Decimal("30.00"), closing_debit=Decimal("7"))
|
||
dumped = row.model_dump(mode="json")
|
||
assert (dumped["opening_debit"], dumped["turnover_debit"], dumped["closing_debit"]) == ("0.00", "30.00", "7.00")
|
||
|
||
|
||
def test_result_table_money_has_two_decimals_with_half_up_rounding() -> None:
|
||
company_id = uuid.uuid4()
|
||
article = schemas.IntercompanyResultArticleOut(
|
||
code="receivables",
|
||
name="Дебиторская задолженность",
|
||
amounts=[
|
||
schemas.IntercompanyResultAmountOut(company_id=company_id, amount=Decimal("0")),
|
||
schemas.IntercompanyResultAmountOut(company_id=uuid.uuid4(), amount=Decimal("1.005")),
|
||
],
|
||
group_total=Decimal("1.005"),
|
||
)
|
||
dumped = article.model_dump(mode="json")
|
||
assert [item["amount"] for item in dumped["amounts"]] == ["0.00", "1.01"]
|
||
assert dumped["group_total"] == "1.01"
|
||
|
||
|
||
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(
|
||
document_type=DocumentType.OSV,
|
||
account="60",
|
||
status=status.value, # type: ignore[arg-type]
|
||
missing_quarters=[],
|
||
document_ids=[],
|
||
)
|
||
assert cell.status is status
|
||
with pytest.raises(ValidationError):
|
||
ChecklistCellOut(
|
||
document_type=DocumentType.OSV,
|
||
account="60",
|
||
status="UNKNOWN", # type: ignore[arg-type]
|
||
missing_quarters=[],
|
||
document_ids=[],
|
||
)
|
||
|
||
|
||
def test_checklist_cell_requires_its_rule() -> None:
|
||
"""D-2: `document_type` обязателен, `account` — обязателен, но может быть `None` (ББ/ОФР)."""
|
||
with pytest.raises(ValidationError):
|
||
ChecklistCellOut(status=CellStatus.MISSING, missing_quarters=[], document_ids=[]) # type: ignore[call-arg]
|
||
cell = ChecklistCellOut(
|
||
document_type=DocumentType.BALANCE,
|
||
account=None,
|
||
status=CellStatus.MISSING,
|
||
missing_quarters=[],
|
||
document_ids=[],
|
||
)
|
||
assert cell.account is None
|
||
|
||
|
||
def test_counterparty_type_is_the_domain_enum() -> None:
|
||
"""Признак контрагента в схеме — `domain.consolidated_osv.CounterpartyType` (одно определение, T2.8)."""
|
||
assert schemas.CounterpartyType is DomainCounterpartyType
|
||
assert OsvFilters(counterparty_type="GK").counterparty_type is DomainCounterpartyType.GK
|
||
with pytest.raises(ValidationError):
|
||
OsvFilters(counterparty_type="OTHER") # type: ignore[arg-type]
|
||
|
||
|
||
def test_members_in_upper_bound() -> None:
|
||
"""M-6: периметр версии — не больше `MAX_MEMBERS` идентификаторов (до разбора дублей сервисом)."""
|
||
ids = [uuid.uuid4() for _ in range(MAX_MEMBERS)]
|
||
assert len(MembersIn(legal_entity_ids=ids).legal_entity_ids) == MAX_MEMBERS
|
||
with pytest.raises(ValidationError):
|
||
MembersIn(legal_entity_ids=[*ids, uuid.uuid4()])
|
||
|
||
|
||
def test_status_comment_and_osv_query_upper_bounds() -> None:
|
||
assert StatusIn(to_status="CANCELLED", comment="а" * MAX_STATUS_COMMENT).comment
|
||
with pytest.raises(ValidationError):
|
||
StatusIn(to_status="CANCELLED", comment="а" * (MAX_STATUS_COMMENT + 1))
|
||
assert OsvFilters(q="а" * MAX_OSV_QUERY).q
|
||
with pytest.raises(ValidationError):
|
||
OsvFilters(q="а" * (MAX_OSV_QUERY + 1))
|