"""Реальные вложения 58.01 в API с настоящей БД и стандартными правами.""" import datetime as dt from decimal import Decimal from pathlib import Path from uuid import uuid4 import pytest from sqlalchemy import delete, func, select from src.models.accounting import ChartOfAccount from src.models.consolidation import Consolidation, ConsolidationDocument from src.models.enums import ( ConsolidationStatus, DocumentStatus, DocumentType, OsvDataType, PeriodQuarter, StatementType, ) from src.models.files import Document, StoredFile from src.models.kfo import Goodwill from src.models.osv import OsvEntry from src.models.statements import ( FinancialStatementColumn, FinancialStatementLine, FinancialStatementMeta, FinancialStatementValue, ) from src.repositories import consolidations as consolidation_repo from tests.factories import make_confirmed_osv from tests.fixtures_helpers import GOLDEN_STATEMENTS, golden_result_v1, map_golden_snapshot from tests.test_api_consolidated_osv import FOREIGN, KA_RF, _world from tests.test_goodwill_golden import investment_rows HTTP_OK, HTTP_NOT_FOUND, HTTP_CONFLICT, HTTP_FORBIDDEN = 200, 404, 409, 403 HTTP_CREATED = 201 pytestmark = [pytest.mark.integration, pytest.mark.fixtures] async def goodwill_world(session, fixtures_dir: Path): w = await _world(session) raw = investment_rows(fixtures_dir)[0] account = (await session.scalars(select(ChartOfAccount).where(ChartOfAccount.code == "58.01"))).one() doc = await make_confirmed_osv( session, entity=w.a, account="58.01", year=2024, quarter=PeriodQuarter.Q3, is_cumulative=True, rows=[ dict( name=raw["counterparty_name"], counterparty_id=w.b.id, account_id=account.id, closing_debit=Decimal(raw["closing_debit"]), ) ], ) session.add(ConsolidationDocument(consolidation_version_id=w.version.id, document_id=doc.id)) balance = await make_confirmed_osv( session, entity=w.b, account="80", year=2024, quarter=PeriodQuarter.Q3, is_cumulative=True, rows=[] ) balance.document_type = DocumentType.BALANCE balance.osv_account_code = None balance.status = DocumentStatus.CONFIRMED meta = FinancialStatementMeta( document_id=balance.id, recognition_job_id=(await session.get(StoredFile, balance.stored_file_id)).current_recognition_job_id, statement_type=StatementType.balance, units="тыс. руб.", ) session.add(meta) await session.flush() column = FinancialStatementColumn( meta_id=meta.id, column_index=0, period_date=dt.date(2024, 9, 30), is_current=True ) line = FinancialStatementLine(meta_id=meta.id, line_code="1310", row_index=0) session.add_all([column, line]) await session.flush() session.add(FinancialStatementValue(line_id=line.id, column_id=column.id, amount=Decimal("100000"))) session.add(ConsolidationDocument(consolidation_version_id=w.version.id, document_id=balance.id)) await session.flush() return w async def test_table_no_write_and_visibility(db_session, auth_client, fixtures_dir): w = await goodwill_world(db_session, fixtures_dir) client = await auth_client(KA_RF) base = f"/consolidations/versions/{w.version.id}" response = await client.get(base + "/goodwill") assert response.status_code == HTTP_OK, response.text row = response.json()["rows"][0] assert row["financial_investment_amount"] == "269997300.00" assert row["charter_capital_amount"] == "100000000.00" assert row["goodwill_amount"] is None assert row["has_investments"] is True assert (await db_session.scalar(select(func.count()).select_from(Goodwill))) == 0 foreign = await auth_client(FOREIGN) assert (await foreign.get(base + "/goodwill")).status_code == HTTP_NOT_FOUND async def test_empty_and_readonly(db_session, auth_client, fixtures_dir): w = await _world(db_session) client = await auth_client(KA_RF) base = f"/consolidations/versions/{w.version.id}" table = (await client.get(base + "/goodwill")).json() assert table["rows"] == [] and table["totals"]["financial_investment_amount"] == "0.00" w.version.status = ConsolidationStatus.VERIFIED await db_session.flush() table = (await client.get(base + "/goodwill")).json() assert table["editable"] is False response = await client.put(base + "/goodwill-settings", json={"goodwill_enabled": True}) assert response.status_code == HTTP_CONFLICT and response.json()["detail"]["code"] == "VERSION_NOT_EDITABLE" async def test_capital_missing_duplicate_and_zero(db_session, auth_client, fixtures_dir): w = await goodwill_world(db_session, fixtures_dir) client = await auth_client(KA_RF) base = f"/consolidations/versions/{w.version.id}" meta = (await db_session.scalars(select(FinancialStatementMeta))).one() column = (await db_session.scalars(select(FinancialStatementColumn))).one() # Дубликаты 1310 суммируются штатным агрегатом формы. line = FinancialStatementLine(meta_id=meta.id, line_code="1310", row_index=1) db_session.add(line) await db_session.flush() value = FinancialStatementValue(line_id=line.id, column_id=column.id, amount=Decimal("1")) db_session.add(value) await db_session.flush() row = (await client.get(base + "/goodwill")).json()["rows"][0] assert row["charter_capital_amount"] == "100001000.00" column.period_date = dt.date(2023, 9, 30) await db_session.flush() row = (await client.get(base + "/goodwill")).json()["rows"][0] assert row["charter_capital_amount"] is None assert row["warnings"][0]["code"] == "CAPITAL_SOURCE_MISSING" column.period_date = dt.date(2024, 9, 30) for val in (await db_session.scalars(select(FinancialStatementValue))).all(): val.amount = 0 await db_session.flush() assert (await client.get(base + "/goodwill")).json()["rows"][0]["charter_capital_amount"] == "0.00" async def test_investments_do_not_apply_custom_document_or_account_priority(db_session, auth_client, fixtures_dir): w = await goodwill_world(db_session, fixtures_dir) client = await auth_client(KA_RF) amount = Decimal(investment_rows(fixtures_dir)[0]["closing_debit"]) account = (await db_session.scalars(select(ChartOfAccount).where(ChartOfAccount.code == "58.01"))).one() broad = await make_confirmed_osv( db_session, entity=w.a, account="58", year=2024, quarter=PeriodQuarter.Q3, is_cumulative=True, rows=[ dict( name="СЕМИКАРАКОРСКИЙ ЭЛЕВАТОР ООО", counterparty_id=w.b.id, account_id=account.id, closing_debit=amount ) ], ) db_session.add(ConsolidationDocument(consolidation_version_id=w.version.id, document_id=broad.id)) await db_session.flush() base = f"/consolidations/versions/{w.version.id}" assert (await client.get(base + "/goodwill")).json()["totals"]["financial_investment_amount"] == "539994600.00" child = ChartOfAccount( code="58.01.99", name="Тестовая детализация", parent_id=account.id, row_type=account.row_type, kind=account.kind ) db_session.add(child) await db_session.flush() detail = await make_confirmed_osv( db_session, entity=w.a, account="58.01.99", year=2024, quarter=PeriodQuarter.Q3, is_cumulative=True, rows=[ dict(name="СЕМИКАРАКОРСКИЙ ЭЛЕВАТОР ООО", counterparty_id=w.b.id, account_id=child.id, closing_debit=amount) ], ) db_session.add(ConsolidationDocument(consolidation_version_id=w.version.id, document_id=detail.id)) await db_session.flush() assert (await client.get(base + "/goodwill")).json()["totals"]["financial_investment_amount"] == "809991900.00" async def test_view_and_edit_permissions(auth_client): client = await auth_client("admin@armka.local") base = f"/consolidations/versions/{uuid4()}" for path in ("/goodwill", "/goodwill/export"): assert (await client.get(base + path)).status_code == HTTP_FORBIDDEN response = await client.put(base + "/goodwill-settings", json={"goodwill_enabled": True}) assert response.status_code == HTTP_FORBIDDEN async def test_real_balance_and_ambiguous_forms(db_session, auth_client, fixtures_dir): w = await goodwill_world(db_session, fixtures_dir) consolidation = await db_session.get(Consolidation, w.version.consolidation_id) consolidation.year = 2025 # Реальный ББ НПЗ, строки и суммы не меняются. path = next((fixtures_dir / GOLDEN_STATEMENTS / "npz/2025-Q3").glob("*.json")) documents = await map_golden_snapshot( db_session, entity=w.b, path=path, uploaded_by=None, result_builder=golden_result_v1 ) previous = ( await db_session.scalars( select(Document) .join(ConsolidationDocument) .where( ConsolidationDocument.consolidation_version_id == w.version.id, Document.document_type == DocumentType.BALANCE, ) ) ).one() await db_session.execute( delete(ConsolidationDocument).where( ConsolidationDocument.consolidation_version_id == w.version.id, ConsolidationDocument.document_id == previous.id, ) ) for doc in documents: if doc.document_type == DocumentType.BALANCE: doc.status = DocumentStatus.CONFIRMED db_session.add(ConsolidationDocument(consolidation_version_id=w.version.id, document_id=doc.id)) for doc in ( await db_session.scalars( select(Document) .join(ConsolidationDocument) .where( ConsolidationDocument.consolidation_version_id == w.version.id, Document.document_type == DocumentType.OSV, ) ) ).all(): doc.period_year = 2025 await db_session.flush() client = await auth_client(KA_RF) base = f"/consolidations/versions/{w.version.id}/goodwill" row = (await client.get(base)).json()["rows"][0] assert row["charter_capital_amount"] == "100000.00" db_session.add(ConsolidationDocument(consolidation_version_id=w.version.id, document_id=previous.id)) await db_session.flush() row = (await client.get(base)).json()["rows"][0] assert row["charter_capital_amount"] is None assert row["warnings"][0]["code"] == "CAPITAL_SOURCE_AMBIGUOUS" async def test_quarterly_bu_and_sum_are_preserved(db_session, auth_client, fixtures_dir): w = await goodwill_world(db_session, fixtures_dir) # Заменяем нарастающий 58.01 тремя квартальными источниками с двумя видами учёта. await db_session.execute( delete(ConsolidationDocument).where( ConsolidationDocument.consolidation_version_id == w.version.id, ConsolidationDocument.document_id.in_(select(Document.id).where(Document.osv_account_code == "58.01")), ) ) account = (await db_session.scalars(select(ChartOfAccount).where(ChartOfAccount.code == "58.01"))).one() raw = investment_rows(fixtures_dir)[0] for quarter in (PeriodQuarter.Q1, PeriodQuarter.Q2, PeriodQuarter.Q3): doc = await make_confirmed_osv( db_session, entity=w.a, account="58.01", year=2024, quarter=quarter, is_cumulative=False, rows=[ dict( name=raw["counterparty_name"], counterparty_id=w.b.id, account_id=account.id, closing_debit=Decimal(raw["closing_debit"]), data_type=kind, ) for kind in (OsvDataType.BU, OsvDataType.SUM) ], ) db_session.add(ConsolidationDocument(consolidation_version_id=w.version.id, document_id=doc.id)) await db_session.flush() client = await auth_client(KA_RF) table = (await client.get(f"/consolidations/versions/{w.version.id}/goodwill")).json() assert table["totals"]["financial_investment_amount"] == "539994600.00" async def test_table_loads_osv_rows_once(db_session, auth_client, fixtures_dir, monkeypatch): world = await goodwill_world(db_session, fixtures_dir) original = consolidation_repo.load_osv_rows calls = 0 async def counted(*args, **kwargs): nonlocal calls calls += 1 return await original(*args, **kwargs) monkeypatch.setattr(consolidation_repo, "load_osv_rows", counted) client = await auth_client(KA_RF) response = await client.get(f"/consolidations/versions/{world.version.id}/goodwill") assert response.status_code == HTTP_OK, response.text assert response.json()["rows"][0]["financial_investment_amount"] == "269997300.00" assert calls == 1 async def test_null_closing_debit_does_not_block_pair_calculation(db_session, auth_client, fixtures_dir): world = await goodwill_world(db_session, fixtures_dir) entry = ( await db_session.scalars( select(OsvEntry).where(OsvEntry.owner_legal_entity_id == world.a.id, OsvEntry.account_code == "58.01") ) ).one() db_session.add( OsvEntry( document_id=entry.document_id, recognition_job_id=entry.recognition_job_id, owner_legal_entity_id=entry.owner_legal_entity_id, row_number=entry.row_number + 1, account_id=entry.account_id, account_code=entry.account_code, account_root=entry.account_root, group_type=entry.group_type, data_type=entry.data_type, counterparty_id=entry.counterparty_id, counterparty_name=entry.counterparty_name, closing_debit=None, ) ) await db_session.flush() client = await auth_client(KA_RF) base = f"/consolidations/versions/{world.version.id}" assert (await client.post(base + "/lock")).status_code == HTTP_CREATED response = await client.put( base + f"/goodwill/{world.a.id}/{world.b.id}", json=dict(ownership_share="100", include_in_kfo=True, comment="Подтверждено"), ) assert response.status_code == HTTP_OK, response.text row = response.json() assert row["financial_investment_amount"] == "269997300.00" assert row["participation_amount"] == "100000000.00" assert row["goodwill_amount"] == "169997300.00" assert row["warnings"] == []