2b5e70f7c3
Схемы формы, StatementService (видимость, 409 DOCUMENT_NOT_STATEMENT, проверки §7 и межформенная), репозиторий (get_meta, load_form, pair_amounts), маршрут GET с DOCUMENT_VIEW, тесты API и детерминированный тест межформенной проверки. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
252 lines
12 KiB
Python
252 lines
12 KiB
Python
"""API формы ББ/ОФР (spec documents-api «Форма ББ/ОФР для экрана проверки», «Правка формы», «Акцепт проверок формы»,
|
||
«Подтверждение документа»; statement-checks; design D-5, D-7…D-9, D-14)."""
|
||
|
||
import datetime
|
||
import json
|
||
from decimal import Decimal
|
||
from http import HTTPStatus
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from sqlalchemy import select, update
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from src.domain.statement_checks import CROSS_FORM_RELATION
|
||
from src.models.directories import LegalEntity, RegionalOffice
|
||
from src.models.enums import DocumentType, FileStatus, RecognitionStatus
|
||
from src.models.files import Document
|
||
from src.models.statements import (
|
||
FinancialStatementColumn,
|
||
FinancialStatementLine,
|
||
FinancialStatementMeta,
|
||
FinancialStatementValue,
|
||
)
|
||
from tests.conftest import AuthClientFactory
|
||
from tests.factories import make_company_group, make_document, make_legal_entity, make_stored_file
|
||
from tests.fixtures_helpers import GOLDEN_STATEMENTS, map_golden_snapshot
|
||
|
||
pytestmark = [pytest.mark.integration, pytest.mark.fixtures]
|
||
|
||
KA_RF = "ka_rf@armka.local"
|
||
RKA = "rka@armka.local" # только DOCUMENT_VIEW
|
||
ROSTOV = "Ростовский региональный филиал"
|
||
MEZH_BALANCE = GOLDEN_STATEMENTS / "mezheninovskaya" / "2025-Q3" / "1_Форма1_бух_09.json"
|
||
MEZH_INCOME = GOLDEN_STATEMENTS / "mezheninovskaya" / "2025-Q3" / "1_Форма2_бух_09.json"
|
||
MEZH_BALANCE_LINES = 37
|
||
|
||
|
||
async def rostov_entity(session: AsyncSession, path: Path, name: str = "ООО Межениновская") -> LegalEntity:
|
||
office = (await session.execute(select(RegionalOffice).where(RegionalOffice.name == ROSTOV))).scalar_one()
|
||
group = await make_company_group(session, name=f"ГК {name}", office=office)
|
||
data = json.loads(path.read_text("utf-8"))
|
||
inn = next((d["header"].get("inn") for d in data["documents"] if d.get("header", {}).get("inn")), None)
|
||
return await make_legal_entity(session, name=name, inn=inn, group=group)
|
||
|
||
|
||
async def form(session: AsyncSession, fixtures_dir: Path, path: Path = MEZH_BALANCE) -> Document:
|
||
entity = await rostov_entity(session, fixtures_dir / path)
|
||
documents = await map_golden_snapshot(session, entity=entity, path=fixtures_dir / path, uploaded_by=None)
|
||
return next(d for d in documents if d.document_type in (DocumentType.BALANCE, DocumentType.INCOME))
|
||
|
||
|
||
async def two_forms(session: AsyncSession, fixtures_dir: Path) -> tuple[Document, Document]:
|
||
"""ББ и ОФР Межениновской за 2025 Q3 одного ЮЛ (два файла)."""
|
||
entity = await rostov_entity(session, fixtures_dir / MEZH_BALANCE)
|
||
balance = next(
|
||
d
|
||
for d in await map_golden_snapshot(session, entity=entity, path=fixtures_dir / MEZH_BALANCE, uploaded_by=None)
|
||
if d.document_type == DocumentType.BALANCE
|
||
)
|
||
income = next(
|
||
d
|
||
for d in await map_golden_snapshot(session, entity=entity, path=fixtures_dir / MEZH_INCOME, uploaded_by=None)
|
||
if d.document_type == DocumentType.INCOME
|
||
)
|
||
return balance, income
|
||
|
||
|
||
async def set_amount( # noqa: PLR0913 - адресация ячейки: документ, строка, колонка
|
||
session: AsyncSession,
|
||
document: Document,
|
||
line_code: str,
|
||
amount: Decimal,
|
||
*,
|
||
current: bool = False,
|
||
period_date: datetime.date | None = None,
|
||
) -> None:
|
||
"""Прямая правка суммы ячейки в БД теста (колонка `is_current` или с датой `period_date`)."""
|
||
meta = (
|
||
await session.execute(select(FinancialStatementMeta).where(FinancialStatementMeta.document_id == document.id))
|
||
).scalar_one()
|
||
column_filter = (
|
||
FinancialStatementColumn.is_current.is_(True)
|
||
if current
|
||
else FinancialStatementColumn.period_date == period_date
|
||
)
|
||
column_id = (
|
||
await session.execute(
|
||
select(FinancialStatementColumn.id)
|
||
.where(FinancialStatementColumn.meta_id == meta.id, column_filter)
|
||
.order_by(FinancialStatementColumn.column_index)
|
||
.limit(1)
|
||
)
|
||
).scalar_one()
|
||
line_id = (
|
||
await session.execute(
|
||
select(FinancialStatementLine.id).where(
|
||
FinancialStatementLine.meta_id == meta.id, FinancialStatementLine.line_code == line_code
|
||
)
|
||
)
|
||
).scalar_one()
|
||
await session.execute(
|
||
update(FinancialStatementValue)
|
||
.where(FinancialStatementValue.line_id == line_id, FinancialStatementValue.column_id == column_id)
|
||
.values(amount=amount)
|
||
)
|
||
await session.flush()
|
||
|
||
|
||
async def test_get_statement(seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory) -> None:
|
||
document = await form(seeded, fixtures_dir)
|
||
client = await auth_client(KA_RF)
|
||
response = await client.get(f"/documents/{document.id}/statement")
|
||
assert response.status_code == HTTPStatus.OK, response.text
|
||
body = response.json()
|
||
assert body["edition"] == "fsbu4"
|
||
assert body["document"]["document_type"] == "BALANCE" and body["document"]["period_quarter"] == "Q3"
|
||
assert len(body["lines"]) == MEZH_BALANCE_LINES
|
||
codes = [line["line_code"] for line in body["lines"]]
|
||
assert codes == sorted(codes)
|
||
columns = len(body["columns"])
|
||
assert all(len(line["values"]) == columns for line in body["lines"])
|
||
assert body["checks"]["failed"] == [] and body["checks"]["accepted"] is None
|
||
assert set(body["header"]["inn"]) == {"value", "original_value", "confidence", "verified", "modified"}
|
||
|
||
|
||
async def test_get_statement_shape_and_decimal_strings(
|
||
seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory
|
||
) -> None:
|
||
"""Документ, шапка и значения: поля спеки, суммы — строкой, `column_index` значения совпадает с колонкой."""
|
||
document = await form(seeded, fixtures_dir)
|
||
client = await auth_client(RKA) # DOCUMENT_VIEW достаточно
|
||
body = (await client.get(f"/documents/{document.id}/statement")).json()
|
||
assert set(body["document"]) == {
|
||
"id",
|
||
"document_type",
|
||
"status",
|
||
"period_year",
|
||
"period_quarter",
|
||
"reporting_date",
|
||
"counterparty_check",
|
||
"locator",
|
||
"file_id",
|
||
}
|
||
assert {"organization_name", "okud", "okpo", "okopf", "okfs", "address"} <= set(body["header"])
|
||
assert set(body["columns"][0]) == {"id", "column_index", "label", "period_date", "is_current"}
|
||
line = next(line for line in body["lines"] if any(v["amount"] is not None for v in line["values"]))
|
||
assert set(line) >= {"id", "line_code", "line_code_confidence", "line_name", "line_name_confidence", "values"}
|
||
value = next(v for v in line["values"] if v["amount"] is not None)
|
||
assert isinstance(value["amount"], str)
|
||
assert set(value) == {
|
||
"id",
|
||
"column_id",
|
||
"column_index",
|
||
"amount",
|
||
"string_value",
|
||
"confidence",
|
||
"is_recognized",
|
||
"is_modified",
|
||
"original_amount",
|
||
"modified_by",
|
||
"modified_at",
|
||
}
|
||
index_by_id = {column["id"]: column["column_index"] for column in body["columns"]}
|
||
assert all(index_by_id[v["column_id"]] == v["column_index"] for line in body["lines"] for v in line["values"])
|
||
assert [v["column_index"] for v in line["values"]] == sorted(v["column_index"] for v in line["values"])
|
||
|
||
|
||
async def test_get_income_statement(seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory) -> None:
|
||
document = await form(seeded, fixtures_dir, MEZH_INCOME)
|
||
client = await auth_client(KA_RF)
|
||
body = (await client.get(f"/documents/{document.id}/statement")).json()
|
||
assert body["document"]["document_type"] == "INCOME" and body["edition"] == "fsbu4"
|
||
assert body["lines"]
|
||
|
||
|
||
async def test_statement_of_osv_is_conflict(
|
||
seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory
|
||
) -> None:
|
||
entity = await rostov_entity(seeded, fixtures_dir / MEZH_BALANCE, name="ООО ОСВ")
|
||
file, job = await make_stored_file(
|
||
seeded, entity=entity, status=FileStatus.PROCESSED, job_status=RecognitionStatus.done
|
||
)
|
||
osv = await make_document(seeded, file=file, job=job)
|
||
client = await auth_client(KA_RF)
|
||
response = await client.get(f"/documents/{osv.id}/statement")
|
||
assert response.status_code == HTTPStatus.CONFLICT and response.json()["detail"]["code"] == "DOCUMENT_NOT_STATEMENT"
|
||
|
||
|
||
async def test_statement_hidden_is_404(
|
||
seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory
|
||
) -> None:
|
||
document = await form(seeded, fixtures_dir)
|
||
client = await auth_client("client_manager@armka.local") # другой филиал
|
||
assert (await client.get(f"/documents/{document.id}/statement")).status_code == HTTPStatus.NOT_FOUND
|
||
|
||
|
||
async def test_cross_form_check(seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory) -> None:
|
||
balance, _ = await two_forms(seeded, fixtures_dir)
|
||
client = await auth_client(KA_RF)
|
||
body = (await client.get(f"/documents/{balance.id}/statement")).json()
|
||
cross = body["checks"]["cross_form"]
|
||
assert cross is None or cross["relation"] == "1370 − 1370(начало года) = 2400"
|
||
|
||
|
||
async def test_cross_form_violation_is_reported_on_both_forms(
|
||
seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory
|
||
) -> None:
|
||
"""Детерминированно (D-5): 1370 = 500 на 30.09.2025 и 300 на 31.12.2024, 2400 = 150 → expected 200, actual 150."""
|
||
balance, income = await two_forms(seeded, fixtures_dir)
|
||
await set_amount(seeded, balance, "1370", Decimal("500"), current=True)
|
||
await set_amount(seeded, balance, "1370", Decimal("300"), period_date=datetime.date(2024, 12, 31))
|
||
await set_amount(seeded, income, "2400", Decimal("150"), current=True)
|
||
client = await auth_client(KA_RF)
|
||
for document in (balance, income):
|
||
body = (await client.get(f"/documents/{document.id}/statement")).json()
|
||
assert body["checks"]["cross_form"] == {
|
||
"column_index": -1,
|
||
"relation": CROSS_FORM_RELATION,
|
||
"expected": "200.00",
|
||
"actual": "150.00",
|
||
}, document.document_type
|
||
# соотношение выполнено: 500 − 300 = 200 → проверка не нарушена
|
||
await set_amount(seeded, income, "2400", Decimal("200"), current=True)
|
||
body = (await client.get(f"/documents/{balance.id}/statement")).json()
|
||
assert body["checks"]["cross_form"] is None
|
||
|
||
|
||
async def test_cross_form_without_pair_is_null(
|
||
seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory
|
||
) -> None:
|
||
document = await form(seeded, fixtures_dir)
|
||
client = await auth_client(KA_RF)
|
||
assert (await client.get(f"/documents/{document.id}/statement")).json()["checks"]["cross_form"] is None
|
||
|
||
|
||
async def test_cross_form_ignores_superseded_pair(
|
||
seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory
|
||
) -> None:
|
||
"""Пара из файла `SUPERSEDED` не участвует (D-5)."""
|
||
balance, income = await two_forms(seeded, fixtures_dir)
|
||
await set_amount(seeded, balance, "1370", Decimal("500"), current=True)
|
||
await set_amount(seeded, balance, "1370", Decimal("300"), period_date=datetime.date(2024, 12, 31))
|
||
await set_amount(seeded, income, "2400", Decimal("150"), current=True)
|
||
from src.models.files import StoredFile # noqa: PLC0415
|
||
|
||
await seeded.execute(
|
||
update(StoredFile).where(StoredFile.id == income.stored_file_id).values(status=FileStatus.SUPERSEDED)
|
||
)
|
||
await seeded.flush()
|
||
client = await auth_client(KA_RF)
|
||
assert (await client.get(f"/documents/{balance.id}/statement")).json()["checks"]["cross_form"] is None
|