"""Чистый домен маппинга `osvDocument` (spec osv-result-mapping; design D-6…D-9; §1.3, §1.5, §9.3). Без БД и S3: строки → черновики `osv_entry`, шапка → черновик `document`, проверка контрагента шапки, статус документа, сверка сальдо на начало для нарастающей (D-9). """ import json import uuid from decimal import Decimal from pathlib import Path from typing import Any import pytest from scripts.seed_lib.loaders import load_chart_of_accounts from src.contract.result import OsvChecks, OsvDocument, OsvHeader from src.domain.osv_mapping import ( DocumentDraft, EntryDraft, account_root, build_entries, counterparty_check, document_fields, document_status, normalize_account_code, opening_matches, ) from src.models.enums import DocumentCheckResult, DocumentStatus, OsvDataType, OsvGroupType, PeriodQuarter API_DIR = Path(__file__).resolve().parent.parent SEED_DIR = API_DIR / "seed" MINIMAL_EXAMPLE = API_DIR / "contracts" / "examples" / "result-osv-minimal.json" GOLDEN_OSV_66 = Path("golden") / "expected_osv_parse" / "rb" / "2024-FY" / "ОСВ_66 за 2024.json" OWNER_INN = "3528052718" OWNER_NAMES = frozenset({"русский бисквит", "рб"}) SUM_COUNT = 6 def _load_doc(path: Path) -> OsvDocument: data = json.loads(path.read_text(encoding="utf-8")) return OsvDocument.model_validate(data["documents"][0]) def _seed_chart() -> dict[str, uuid.UUID]: """`code → id` из seed-плана счетов (как `chart_of_account` после `make seed`).""" codes = load_chart_of_accounts(SEED_DIR / "core", SEED_DIR) return {code: uuid.uuid5(uuid.NAMESPACE_URL, f"chart:{code}") for code in codes} def _header(**fields: Any) -> OsvHeader: base: dict[str, Any] = {"organization_name": "", "account_code_raw": "60", "period_text": "2024 г."} return OsvHeader.model_validate(base | fields) def _entry(opening_debit: Decimal | None, opening_credit: Decimal | None) -> EntryDraft: return EntryDraft( id=uuid.uuid4(), row_number=1, level=0, parent_id=None, row_value="60", group_type=OsvGroupType.ACCOUNT, group_name=None, data_type=OsvDataType.BU, account_code="60", account_root="60", account_id=None, counterparty_name=None, counterparty_inn=None, sums=(opening_debit, opening_credit, None, None, None, None), ) # --- нормализация кода счёта (§1.3) --------------------------------------------------------------------------------- @pytest.mark.parametrize( ("raw", "expected"), [ ("60.1", "60.01"), ("76.7.1", "76.07.1"), ("60.01.1", "60.01.1"), ("58.03", "58.03"), ("76.АВ", "76.АВ"), ("МЦ.04", "МЦ.04"), ("001", "001"), ("60", "60"), ], ) def test_normalize_account_code(raw: str, expected: str) -> None: assert normalize_account_code(raw) == expected @pytest.mark.parametrize(("code", "expected"), [("60", "60"), ("60.01", "60"), ("001", "001"), ("МЦ.04", None)]) def test_account_root_is_numeric_first_segment(code: str, expected: str | None) -> None: """ASSUMPTION(OQ-103): нечисловой первый сегмент (забалансовые «МЦ.04») → `None`.""" assert account_root(code) == expected # --- строки ОСВ (D-6) ----------------------------------------------------------------------------------------------- def test_build_entries_on_minimal_example() -> None: doc = _load_doc(MINIMAL_EXAMPLE) chart = {"60": uuid.uuid4()} sheet = build_entries(doc, chart) assert [e.row_number for e in sheet.entries] == [1, 2] root, child = sheet.entries assert sheet.root == root assert sheet.total_present is False assert sheet.orphans == () assert root.parent_id is None assert child.parent_id == root.id assert root.id != child.id assert root.group_type is OsvGroupType.ACCOUNT assert child.group_type is OsvGroupType.COUNTERPARTY assert child.group_name == "Контрагенты" assert child.data_type is OsvDataType.BU assert child.account_code == "60" assert child.account_root == "60" assert child.account_id == chart["60"] assert child.counterparty_name == "ООО Лютик" assert child.counterparty_inn == "7711111111" assert child.counterparty_id is None # R-3: резолв — в цикле маппинга, не здесь assert child.sums == ( Decimal("0.00"), Decimal("1000.00"), Decimal("500.00"), Decimal("200.50"), Decimal("0.00"), Decimal("700.50"), ) assert all(isinstance(value, Decimal) for value in child.sums) def test_build_entries_skips_total_and_reports_orphans() -> None: data = json.loads(MINIMAL_EXAMPLE.read_text(encoding="utf-8"))["documents"][0] rows = data["rows"] total = rows[0] | {"row_number": 3, "row_value": "Итого", "group_type": "TOTAL", "account_code_raw": None} orphan = rows[1] | {"row_number": 4, "parent_row_number": 99, "data_type": "NU", "account_code_raw": "МЦ.04"} doc = OsvDocument.model_validate(data | {"rows": [*rows, total, orphan]}) sheet = build_entries(doc, {}) assert [e.row_number for e in sheet.entries] == [1, 2, 4] assert sheet.total_present is True assert sheet.orphans == (4,) last = sheet.entries[-1] assert last.parent_id is None assert last.data_type is OsvDataType.NU assert last.account_code == "МЦ.04" assert last.account_root is None assert last.account_id is None def test_build_entries_keeps_absent_account_and_data_type() -> None: data = json.loads(MINIMAL_EXAMPLE.read_text(encoding="utf-8"))["documents"][0] row = data["rows"][1] | {"account_code_raw": None, "data_type": None, "opening_debit": None} doc = OsvDocument.model_validate(data | {"rows": [data["rows"][0], row]}) entry = build_entries(doc, {"60": uuid.uuid4()}).entries[1] assert entry.account_code is None assert entry.account_root is None assert entry.account_id is None assert entry.data_type is None assert entry.sums[0] is None @pytest.mark.fixtures def test_build_entries_on_golden_osv_66(fixtures_dir: Path) -> None: doc = _load_doc(fixtures_dir / GOLDEN_OSV_66) chart = _seed_chart() sheet = build_entries(doc, chart) assert len(sheet.entries) == len(doc.rows) - 1 # всё, кроме TOTAL (OQ-01) assert sheet.total_present is True assert sheet.orphans == () assert sheet.root is not None assert sheet.root.row_value == "66" by_number = {e.row_number: e for e in sheet.entries} by_id = {e.id: e for e in sheet.entries} bakery = [e for e in sheet.entries if e.counterparty_name == "Юг-Финанс / ПЕКАРНЯ СЧАСТЛИВОЧНАЯ"] assert [e.account_code for e in bakery] == ["66.03", "66.04"] first = bakery[0] assert first.parent_id is not None assert by_id[first.parent_id].row_value == "66.03" assert first.account_root == "66" assert first.account_id == chart["66.03"] assert by_number[13].account_id is None # 66.01 нет в seed-плане счетов for entry in sheet.entries: assert (entry.level == 0) == (entry.parent_id is None) assert len(entry.sums) == SUM_COUNT assert all(value is None or isinstance(value, Decimal) for value in entry.sums) # --- документ (D-7) ------------------------------------------------------------------------------------------------- def test_document_fields_on_minimal_example() -> None: draft = document_fields(_load_doc(MINIMAL_EXAMPLE)) assert draft == DocumentDraft( osv_account_code="60", period_year=2025, period_quarter=PeriodQuarter.Q1, is_cumulative=False, locator="sheet:Лист_1", ) def test_document_fields_normalize_account_and_keep_absent_period() -> None: data = json.loads(MINIMAL_EXAMPLE.read_text(encoding="utf-8"))["documents"][0] header = data["header"] | {"account_code_raw": "60.1", "period_year": None, "period_quarter": None} header.pop("is_cumulative") draft = document_fields(OsvDocument.model_validate(data | {"header": header})) assert draft.osv_account_code == "60.01" assert draft.period_year is None assert draft.period_quarter is None assert draft.is_cumulative is None @pytest.mark.fixtures def test_document_fields_on_golden_osv_66(fixtures_dir: Path) -> None: draft = document_fields(_load_doc(fixtures_dir / GOLDEN_OSV_66)) assert draft == DocumentDraft( osv_account_code="66", period_year=2024, period_quarter=PeriodQuarter.Q4, is_cumulative=True, locator="sheet:Лист_1", ) # --- проверка контрагента шапки (D-8) ------------------------------------------------------------------------------- @pytest.mark.parametrize( ("header", "expected"), [ ({"organization_name": "АО «Русский бисквит»", "organization_inn": OWNER_INN}, DocumentCheckResult.MATCH), ({"organization_name": "", "organization_inn": " 3528 052718 "}, DocumentCheckResult.MATCH), ({"organization_name": "АО «Русский бисквит»", "organization_inn": "3528000000"}, DocumentCheckResult.MISMATCH), ({"organization_name": 'АО "Русский бисквит"'}, DocumentCheckResult.MATCH), # по нормализованному имени ({"organization_name": "ООО РБ"}, DocumentCheckResult.MATCH), # по алиасу ({"organization_name": "Пекарня Счастливочная"}, DocumentCheckResult.MISMATCH), ({"organization_name": "", "organization_inn": None}, DocumentCheckResult.UNKNOWN), # пустая шапка ({"organization_name": " ", "organization_inn": ""}, DocumentCheckResult.UNKNOWN), ({"organization_name": "ООО"}, DocumentCheckResult.UNKNOWN), # только ОПФ: нормализуется в пустое имя ], ) def test_counterparty_check(header: dict[str, Any], expected: DocumentCheckResult) -> None: assert counterparty_check(_header(**header), OWNER_INN, OWNER_NAMES) is expected # --- статус документа (§9.3, OQ-05) --------------------------------------------------------------------------------- CLEAN = OsvChecks(total_row_matches_root=True, balance_identity_violations=[], parent_sum_violations=[]) @pytest.mark.parametrize( ("job_done", "checks", "counterparty", "warnings", "expected"), [ (True, CLEAN, DocumentCheckResult.MATCH, [], DocumentStatus.RECOGNIZED), (True, None, DocumentCheckResult.UNKNOWN, [], DocumentStatus.RECOGNIZED), (False, CLEAN, DocumentCheckResult.MATCH, [], DocumentStatus.NEEDS_CHECK), (True, OsvChecks(total_row_matches_root=False), DocumentCheckResult.MATCH, [], DocumentStatus.NEEDS_CHECK), (True, OsvChecks(balance_identity_violations=[5]), DocumentCheckResult.MATCH, [], DocumentStatus.NEEDS_CHECK), (True, OsvChecks(parent_sum_violations=[7]), DocumentCheckResult.MATCH, [], DocumentStatus.NEEDS_CHECK), (True, CLEAN, DocumentCheckResult.MISMATCH, [], DocumentStatus.NEEDS_CHECK), (True, CLEAN, DocumentCheckResult.MATCH, ["sheet=Лист_2: счёт 60 уже есть"], DocumentStatus.NEEDS_CHECK), ], ) def test_document_status( job_done: bool, checks: OsvChecks | None, counterparty: DocumentCheckResult, warnings: list[str], expected: DocumentStatus, ) -> None: status = document_status(job_done=job_done, checks=checks, counterparty=counterparty, warnings=warnings) assert status is expected # --- сверка сальдо на начало с Q1 (D-9) ----------------------------------------------------------------------------- @pytest.mark.parametrize( ("entry", "q1_opening", "expected"), [ (_entry(Decimal("100.00"), None), (Decimal("100.00"), None), True), (_entry(Decimal("100.00"), None), (Decimal("100.01"), None), True), # граница ±0.01 включительно (_entry(Decimal("100.00"), None), (Decimal("99.99"), Decimal("0.00")), True), # None = 0 (_entry(Decimal("100.00"), None), (Decimal("100.02"), None), False), (_entry(None, Decimal("5.00")), (None, Decimal("4.98")), False), (_entry(None, None), (Decimal("0.01"), Decimal("-0.01")), True), ], ) def test_opening_matches(entry: EntryDraft, q1_opening: tuple[Decimal | None, Decimal | None], expected: bool) -> None: assert opening_matches(entry, q1_opening) is expected