228 lines
10 KiB
Python
228 lines
10 KiB
Python
"""Гудвилл изменяет актив и пассив на одинаковую сумму, не смешивает пары."""
|
||
|
||
from decimal import Decimal
|
||
from http import HTTPStatus
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
from unittest.mock import AsyncMock
|
||
from uuid import UUID
|
||
|
||
import pytest
|
||
from sqlalchemy import func, select
|
||
|
||
from src.common.exceptions import ConflictError
|
||
from src.domain.goodwill import Decision, Investment, build, contributions
|
||
from src.domain.policy import VisibilityScope
|
||
from src.models.access import AppUser
|
||
from src.models.accounting import ChartOfAccount, EliminationRule
|
||
from src.models.consolidation import ConsolidationDocument
|
||
from src.models.enums import DocumentType, PeriodQuarter, StatementType
|
||
from src.models.files import StoredFile
|
||
from src.models.kfo import Correction, Goodwill, KfoResult
|
||
from src.models.osv import OsvEntry
|
||
from src.models.statements import FinancialStatementMeta
|
||
from src.repositories import consolidations as version_repo
|
||
from src.repositories import goodwill as goodwill_repo
|
||
from src.services.consolidated_osv import ConsolidatedOsvService
|
||
from src.services.goodwill import GoodwillService
|
||
from src.services.goodwill_source import GoodwillSource
|
||
from tests.factories import make_confirmed_osv
|
||
from tests.test_api_consolidated_osv import KA_RF
|
||
from tests.test_api_goodwill import goodwill_world
|
||
from tests.test_goodwill_golden import investment_rows
|
||
|
||
BB_RULE_NO = 2
|
||
|
||
pytestmark = pytest.mark.fixtures
|
||
|
||
|
||
def test_contribution_balance(fixtures_dir: Path) -> None:
|
||
a, b, version, rule = (UUID(int=i) for i in range(1, 5))
|
||
fv = Decimal(investment_rows(fixtures_dir)[0]["closing_debit"])
|
||
saved = {(a, b): Decision(Decimal("80"), True, "Подтверждено")}
|
||
rows = build([Investment(a, b, fv)], saved, {b: Decimal("100000000")})
|
||
item = contributions(version, rule, rows, True)[0]
|
||
assert item.rule_id == rule and item.head_id == a and item.subsidiary_id == b
|
||
assert sum(v for _, code, v in item.deltas if code in ("1170", "GOODWILL")) == -item.participation_amount
|
||
assert dict((code, value) for _, code, value in item.deltas)["1310"] == -item.participation_amount
|
||
assert contributions(version, rule, rows, False) == []
|
||
assert contributions(version, rule, build([Investment(a, b, fv)], saved, {}), True) == []
|
||
assert contributions(version, rule, build([Investment(a, b, fv)], saved, {b: fv * 2}), True) == []
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_service_uses_pinned_rule_without_writing_kfo(db_session, auth_client, fixtures_dir):
|
||
world = await goodwill_world(db_session, fixtures_dir)
|
||
client = await auth_client(KA_RF)
|
||
base = f"/consolidations/versions/{world.version.id}"
|
||
assert (await client.post(base + "/lock")).status_code == HTTPStatus.CREATED
|
||
assert (
|
||
await client.put(
|
||
base + f"/goodwill/{world.a.id}/{world.b.id}",
|
||
json=dict(ownership_share="80", include_in_kfo=True, comment="Да"),
|
||
)
|
||
).status_code == HTTPStatus.OK
|
||
assert (await client.put(base + "/goodwill-settings", json={"goodwill_enabled": True})).status_code == HTTPStatus.OK
|
||
pinned = (
|
||
await db_session.scalars(
|
||
select(EliminationRule).where(
|
||
EliminationRule.version_no == 1,
|
||
EliminationRule.rule_no == BB_RULE_NO,
|
||
EliminationRule.statement_type == StatementType.balance,
|
||
)
|
||
)
|
||
).one()
|
||
other = EliminationRule(
|
||
**{
|
||
col.name: getattr(pinned, col.name)
|
||
for col in EliminationRule.__table__.columns
|
||
if col.name not in ("id", "version_no")
|
||
},
|
||
version_no=2,
|
||
)
|
||
db_session.add(other)
|
||
await db_session.flush()
|
||
principal_user = (await db_session.scalars(select(AppUser).where(AppUser.email == KA_RF))).one()
|
||
scope = VisibilityScope(office_id=None)
|
||
service = GoodwillService(
|
||
db_session,
|
||
scope,
|
||
SimpleNamespace(user=principal_user, permissions={"CONSOLIDATION_EDIT"}),
|
||
GoodwillSource(db_session, ConsolidatedOsvService(db_session, scope)),
|
||
)
|
||
result = await service.contributions(world.version.id)
|
||
assert len(result) == 1 and result[0].rule_id == pinned.id
|
||
assert await db_session.scalar(select(func.count()).select_from(KfoResult)) == 0
|
||
assert await db_session.scalar(select(func.count()).select_from(Correction)) == 0
|
||
# Дорезолв на чтение сохраняет те же выбранные пары и суммы.
|
||
entry = (
|
||
await db_session.scalars(
|
||
select(OsvEntry).where(OsvEntry.owner_legal_entity_id == world.a.id, OsvEntry.account_code == "58.01")
|
||
)
|
||
).one()
|
||
entry.counterparty_id = None
|
||
entry.counterparty_name = world.b.name
|
||
await db_session.flush()
|
||
assert await service.contributions(world.version.id) == result
|
||
pinned.rule_no = 99
|
||
await db_session.flush()
|
||
with pytest.raises(ConflictError) as error:
|
||
await service.contributions(world.version.id)
|
||
assert error.value.code == "GOODWILL_RULE_MISSING"
|
||
|
||
world.version.goodwill_enabled = False
|
||
await db_session.flush()
|
||
assert await service.contributions(world.version.id) == []
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_kfo_loads_only_saved_selected_pairs(db_session, auth_client, fixtures_dir, monkeypatch):
|
||
world = await goodwill_world(db_session, fixtures_dir)
|
||
client = await auth_client(KA_RF)
|
||
base = f"/consolidations/versions/{world.version.id}"
|
||
await client.post(base + "/lock")
|
||
await client.put(
|
||
base + f"/goodwill/{world.a.id}/{world.b.id}",
|
||
json=dict(ownership_share="80", include_in_kfo=True, comment="Да"),
|
||
)
|
||
await client.put(base + "/goodwill-settings", json={"goodwill_enabled": True})
|
||
db_session.add(
|
||
Goodwill(
|
||
consolidation_version_id=world.version.id,
|
||
head_legal_entity_id=world.a.id,
|
||
subsidiary_legal_entity_id=world.chkf.id,
|
||
ownership_share=Decimal("100"),
|
||
include_in_kfo=False,
|
||
)
|
||
)
|
||
account = (await db_session.scalars(select(ChartOfAccount).where(ChartOfAccount.code == "58.01"))).one()
|
||
for owner, other in ((world.a, world.chkf), (world.b, world.a)):
|
||
doc = await make_confirmed_osv(
|
||
db_session,
|
||
entity=owner,
|
||
account="58.01",
|
||
year=2024,
|
||
quarter=PeriodQuarter.Q3,
|
||
is_cumulative=True,
|
||
rows=[
|
||
dict(
|
||
name=other.name,
|
||
counterparty_id=other.id,
|
||
account_id=account.id,
|
||
closing_debit=Decimal("269997300.00"),
|
||
)
|
||
],
|
||
)
|
||
db_session.add(ConsolidationDocument(consolidation_version_id=world.version.id, document_id=doc.id))
|
||
extra = await make_confirmed_osv(
|
||
db_session, entity=world.a, account="80", year=2024, quarter=PeriodQuarter.Q3, is_cumulative=True, rows=[]
|
||
)
|
||
extra.document_type = DocumentType.BALANCE
|
||
extra.osv_account_code = None
|
||
file = await db_session.get(StoredFile, extra.stored_file_id)
|
||
meta = FinancialStatementMeta(
|
||
document_id=extra.id,
|
||
recognition_job_id=file.current_recognition_job_id,
|
||
statement_type=StatementType.balance,
|
||
units="тыс. руб.",
|
||
)
|
||
db_session.add(meta)
|
||
db_session.add(ConsolidationDocument(consolidation_version_id=world.version.id, document_id=extra.id))
|
||
await db_session.flush()
|
||
original_rows, original_forms = version_repo.load_osv_rows, goodwill_repo.forms
|
||
loaded_pairs, loaded_form_owners = [], []
|
||
|
||
async def capture_rows(*args, **kwargs):
|
||
assert kwargs["company_pairs"] == {(world.a.id, world.b.id)}
|
||
assert account.id in kwargs["account_ids"]
|
||
rows = await original_rows(*args, **kwargs)
|
||
loaded_pairs.extend((e.owner_legal_entity_id, e.counterparty_id) for e, _, _ in rows)
|
||
return rows
|
||
|
||
async def capture_forms(*args, **kwargs):
|
||
rows = await original_forms(*args, **kwargs)
|
||
loaded_form_owners.extend(doc.legal_entity_id for doc, _ in rows)
|
||
return rows
|
||
|
||
monkeypatch.setattr(version_repo, "load_osv_rows", capture_rows)
|
||
monkeypatch.setattr(goodwill_repo, "forms", capture_forms)
|
||
user = (await db_session.scalars(select(AppUser).where(AppUser.email == KA_RF))).one()
|
||
scope = VisibilityScope(office_id=None)
|
||
source = GoodwillSource(db_session, ConsolidatedOsvService(db_session, scope))
|
||
source.load = AsyncMock(side_effect=AssertionError("Полная таблица для КФО не нужна"))
|
||
service = GoodwillService(db_session, scope, SimpleNamespace(user=user), source)
|
||
result = await service.contributions(world.version.id)
|
||
assert len(result) == 1
|
||
assert {(r.head_id, r.subsidiary_id) for r in result} == {(world.a.id, world.b.id)}
|
||
assert loaded_pairs and set(loaded_pairs) == {(world.a.id, world.b.id)}
|
||
assert loaded_form_owners == [world.b.id]
|
||
source.load.assert_not_awaited()
|
||
|
||
|
||
@pytest.mark.integration
|
||
async def test_kfo_without_selected_rows_does_not_load_sources(db_session, auth_client, fixtures_dir):
|
||
world = await goodwill_world(db_session, fixtures_dir)
|
||
world.version.goodwill_enabled = True
|
||
await db_session.flush()
|
||
user = (await db_session.scalars(select(AppUser).where(AppUser.email == KA_RF))).one()
|
||
scope = VisibilityScope(office_id=None)
|
||
source = GoodwillSource(db_session, ConsolidatedOsvService(db_session, scope))
|
||
source.load_for_kfo = AsyncMock(side_effect=AssertionError("Нет выбранных пар — нет источников"))
|
||
service = GoodwillService(db_session, scope, SimpleNamespace(user=user), source)
|
||
assert await service.contributions(world.version.id) == []
|
||
source.load_for_kfo.assert_not_awaited()
|
||
db_session.add(
|
||
Goodwill(
|
||
consolidation_version_id=world.version.id,
|
||
head_legal_entity_id=world.a.id,
|
||
subsidiary_legal_entity_id=world.b.id,
|
||
ownership_share=Decimal("100"),
|
||
include_in_kfo=True,
|
||
comment="Да",
|
||
)
|
||
)
|
||
world.version.goodwill_enabled = False
|
||
await db_session.flush()
|
||
assert await service.contributions(world.version.id) == []
|
||
source.load_for_kfo.assert_not_awaited()
|