251 lines
12 KiB
Python
251 lines
12 KiB
Python
"""Ручные операции над назначениями ВГО для полных строк сводной ОСВ."""
|
||
|
||
import uuid
|
||
from http import HTTPStatus
|
||
from types import SimpleNamespace
|
||
from unittest.mock import AsyncMock, Mock
|
||
|
||
import pytest
|
||
from pydantic import ValidationError
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from src.api.endpoints.consolidations.vgo_assignment import router
|
||
from src.common.clock import utc_now
|
||
from src.common.exceptions import DomainValidationError
|
||
from src.models.access import AppUser
|
||
from src.models.accounting import EliminationRule
|
||
from src.models.consolidation import ConsolidationDocument
|
||
from src.models.directories import RegionalOffice
|
||
from src.models.enums import ConsolidationStatus, PeriodQuarter
|
||
from src.models.kfo import VgoAssignment
|
||
from src.models.osv import OsvEntry
|
||
from src.schemas.consolidations import VgoAssignmentAttachRuleIn, VgoAssignmentDetachRuleIn, VgoAssignmentExcludeIn
|
||
from src.services import vgo
|
||
from src.services.vgo import VgoAssignmentCrudService
|
||
from tests.conftest import AuthClientFactory
|
||
from tests.factories import make_company_group, make_confirmed_osv, make_consolidation, make_legal_entity, make_lock
|
||
|
||
USER_EMAIL = "ka_rf@armka.local"
|
||
|
||
|
||
def test_assignment_routes_are_limited_to_requested_actions() -> None:
|
||
routes = {(method, route.path) for route in router.routes for method in route.methods}
|
||
prefix = "/consolidations/versions/{version_id}/vgo-assignments"
|
||
assert routes == {
|
||
("POST", f"{prefix}/distribute"),
|
||
("PUT", f"{prefix}/rule"),
|
||
("DELETE", f"{prefix}/rule"),
|
||
("PUT", f"{prefix}/exclude"),
|
||
}
|
||
|
||
|
||
def test_inputs_require_unique_ids() -> None:
|
||
entry_id = uuid.uuid4()
|
||
with pytest.raises(ValidationError, match="entry_ids не должны повторяться"):
|
||
VgoAssignmentAttachRuleIn(entry_ids=[entry_id, entry_id], rule_id=uuid.uuid4())
|
||
with pytest.raises(ValidationError, match="entry_ids не должны повторяться"):
|
||
VgoAssignmentDetachRuleIn(statement_type="balance", entry_ids=[entry_id, entry_id])
|
||
with pytest.raises(ValidationError, match="entry_ids не должны повторяться"):
|
||
VgoAssignmentExcludeIn(statement_type="balance", entry_ids=[entry_id, entry_id], is_exclude=True)
|
||
|
||
|
||
async def test_write_requires_complete_consolidated_row() -> None:
|
||
entry_a, entry_b = uuid.uuid4(), uuid.uuid4()
|
||
row = SimpleNamespace(entry_ids=(entry_a, entry_b))
|
||
osv_result = SimpleNamespace(consolidated=SimpleNamespace(rows=[row]))
|
||
|
||
service = VgoAssignmentCrudService(
|
||
session=AsyncMock(), audit=AsyncMock(), scope=SimpleNamespace(),
|
||
osv_service=SimpleNamespace(build_with_members=AsyncMock(return_value=osv_result)),
|
||
)
|
||
version_id = uuid.uuid4()
|
||
with pytest.raises(DomainValidationError, match="все исходные ID"):
|
||
await service._check_rows(version_id, [entry_a])
|
||
await service._check_rows(version_id, [entry_a, entry_b])
|
||
|
||
|
||
async def test_cannot_exclude_entries_with_rule(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
entry_id = uuid.uuid4()
|
||
session = AsyncMock()
|
||
service = VgoAssignmentCrudService(
|
||
session=session, audit=SimpleNamespace(user_id=uuid.uuid4()), scope=SimpleNamespace(),
|
||
osv_service=SimpleNamespace(build_with_members=AsyncMock(return_value=SimpleNamespace(
|
||
consolidated=SimpleNamespace(rows=[SimpleNamespace(entry_ids=(entry_id,))])
|
||
))),
|
||
)
|
||
monkeypatch.setattr(vgo, "lock_editable_version", AsyncMock())
|
||
monkeypatch.setattr(
|
||
vgo.vgo_assignments_repo, "for_entries",
|
||
AsyncMock(return_value={entry_id: SimpleNamespace(rule_id=uuid.uuid4())}),
|
||
)
|
||
data = VgoAssignmentExcludeIn(statement_type="balance", entry_ids=[entry_id], is_exclude=True)
|
||
with pytest.raises(DomainValidationError, match="нельзя исключить из КФО"):
|
||
await service.set_exclusion(uuid.uuid4(), data)
|
||
session.flush.assert_not_awaited()
|
||
|
||
|
||
async def test_detach_rule_keeps_assignment(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
version_id, entry_id, user_id = uuid.uuid4(), uuid.uuid4(), uuid.uuid4()
|
||
item = VgoAssignment(
|
||
id=uuid.uuid4(), consolidation_version_id=version_id, statement_type="balance",
|
||
osv_entry_id=entry_id, action="ASSIGN", rule_id=uuid.uuid4(),
|
||
is_exclude=False, created_by=user_id, created_at=utc_now(),
|
||
)
|
||
session = AsyncMock()
|
||
service = VgoAssignmentCrudService(
|
||
session=session, audit=SimpleNamespace(user_id=user_id), scope=SimpleNamespace(),
|
||
osv_service=SimpleNamespace(build_with_members=AsyncMock(return_value=SimpleNamespace(
|
||
consolidated=SimpleNamespace(rows=[SimpleNamespace(entry_ids=(entry_id,))])
|
||
))),
|
||
)
|
||
service._audit = AsyncMock()
|
||
monkeypatch.setattr(vgo, "lock_editable_version", AsyncMock())
|
||
monkeypatch.setattr(vgo.vgo_assignments_repo, "for_entries", AsyncMock(return_value={entry_id: item}))
|
||
result = await service.detach_rule(
|
||
version_id, VgoAssignmentDetachRuleIn(statement_type="balance", entry_ids=[entry_id])
|
||
)
|
||
assert result[0].id == item.id
|
||
assert result[0].action == "UNASSIGNED"
|
||
assert result[0].rule_id is None
|
||
assert result[0].is_exclude is True
|
||
session.delete.assert_not_awaited()
|
||
session.flush.assert_awaited_once()
|
||
|
||
|
||
async def test_detach_rule_creates_missing_assignment(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
version_id, entry_id, user_id = uuid.uuid4(), uuid.uuid4(), uuid.uuid4()
|
||
session = SimpleNamespace(add=Mock(), flush=AsyncMock())
|
||
service = VgoAssignmentCrudService(
|
||
session=session, audit=SimpleNamespace(user_id=user_id), scope=SimpleNamespace(),
|
||
osv_service=SimpleNamespace(build_with_members=AsyncMock(return_value=SimpleNamespace(
|
||
consolidated=SimpleNamespace(rows=[SimpleNamespace(entry_ids=(entry_id,))])
|
||
))),
|
||
)
|
||
service._audit = AsyncMock()
|
||
monkeypatch.setattr(vgo, "lock_editable_version", AsyncMock())
|
||
monkeypatch.setattr(vgo.vgo_assignments_repo, "for_entries", AsyncMock(return_value={}))
|
||
monkeypatch.setattr(vgo.VgoAssignmentOut, "model_validate", lambda item, from_attributes: item)
|
||
|
||
result = await service.detach_rule(
|
||
version_id, VgoAssignmentDetachRuleIn(statement_type="balance", entry_ids=[entry_id])
|
||
)
|
||
item = result[0]
|
||
session.add.assert_called_once_with(item)
|
||
assert item.osv_entry_id == entry_id
|
||
assert item.rule_id is None
|
||
assert item.action == "UNASSIGNED"
|
||
assert item.is_exclude is True
|
||
session.flush.assert_awaited_once()
|
||
|
||
|
||
async def test_exclusion_creates_missing_assignment(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
version_id, entry_id, user_id = uuid.uuid4(), uuid.uuid4(), uuid.uuid4()
|
||
session = SimpleNamespace(add=Mock(), flush=AsyncMock())
|
||
service = VgoAssignmentCrudService(
|
||
session=session, audit=SimpleNamespace(user_id=user_id), scope=SimpleNamespace(),
|
||
osv_service=SimpleNamespace(build_with_members=AsyncMock(return_value=SimpleNamespace(
|
||
consolidated=SimpleNamespace(rows=[SimpleNamespace(entry_ids=(entry_id,))])
|
||
))),
|
||
)
|
||
service._audit = AsyncMock()
|
||
monkeypatch.setattr(vgo, "lock_editable_version", AsyncMock())
|
||
monkeypatch.setattr(vgo.vgo_assignments_repo, "for_entries", AsyncMock(return_value={}))
|
||
monkeypatch.setattr(vgo.VgoAssignmentOut, "model_validate", lambda item, from_attributes: item)
|
||
|
||
result = await service.set_exclusion(
|
||
version_id, VgoAssignmentExcludeIn(statement_type="balance", entry_ids=[entry_id], is_exclude=True)
|
||
)
|
||
item = result[0]
|
||
session.add.assert_called_once_with(item)
|
||
assert item.osv_entry_id == entry_id
|
||
assert item.rule_id is None
|
||
assert item.action == "UNASSIGNED"
|
||
assert item.is_exclude is True
|
||
session.flush.assert_awaited_once()
|
||
|
||
|
||
async def test_vgo_assignment_integration(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
|
||
user = (await seeded.scalars(select(AppUser).where(AppUser.email == USER_EMAIL))).one()
|
||
office = (await seeded.scalars(select(RegionalOffice).where(RegionalOffice.id == user.regional_office_id))).one()
|
||
group = await make_company_group(seeded, office=office)
|
||
company_a = await make_legal_entity(seeded, name="Компания A", group=group)
|
||
company_b = await make_legal_entity(seeded, name="Компания B", group=group)
|
||
_, version = await make_consolidation(
|
||
seeded, group=group, year=2024, quarter=PeriodQuarter.Q2,
|
||
members=[company_a, company_b], initiator=user, status=ConsolidationStatus.DATA_CORRECTION,
|
||
)
|
||
await make_lock(seeded, version=version, user=user)
|
||
documents = []
|
||
for quarter in (PeriodQuarter.Q1, PeriodQuarter.Q2):
|
||
document = await make_confirmed_osv(
|
||
seeded, entity=company_a, account="60", year=2024, quarter=quarter,
|
||
is_cumulative=False, rows=[{"name": "Компания B", "counterparty_id": company_b.id, "closing_debit": 10}],
|
||
)
|
||
documents.append(document)
|
||
seeded.add(ConsolidationDocument(consolidation_version_id=version.id, document_id=document.id))
|
||
await seeded.flush()
|
||
entry_ids = list((await seeded.scalars(
|
||
select(OsvEntry.id).where(OsvEntry.document_id.in_([document.id for document in documents]))
|
||
)).all())
|
||
client = await auth_client(USER_EMAIL)
|
||
await _check_assignment_actions(client, seeded, version, entry_ids)
|
||
|
||
|
||
async def _check_assignment_actions(client, seeded, version, entry_ids) -> None:
|
||
url = f"/consolidations/versions/{version.id}/vgo-assignments"
|
||
entry_id_strings = [str(entry_id) for entry_id in entry_ids]
|
||
rule = (await seeded.scalars(select(EliminationRule).where(
|
||
EliminationRule.version_no == version.elimination_rules_version,
|
||
EliminationRule.statement_type == "balance",
|
||
))).first()
|
||
assert rule is not None
|
||
attach_url = f"{url}/rule"
|
||
attach_body = {"entry_ids": entry_id_strings, "rule_id": str(rule.id)}
|
||
exclude_url = f"{url}/exclude"
|
||
exclude_body = {"statement_type": "balance", "entry_ids": entry_id_strings, "is_exclude": True}
|
||
|
||
partial = await client.put(exclude_url, json={**exclude_body, "entry_ids": entry_id_strings[:1]})
|
||
assert partial.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
|
||
excluded = await client.put(exclude_url, json=exclude_body)
|
||
assert excluded.status_code == HTTPStatus.OK, excluded.text
|
||
assert {item["osv_entry_id"] for item in excluded.json()} == set(entry_id_strings)
|
||
assert all(item["is_exclude"] and item["rule_id"] is None for item in excluded.json())
|
||
included = await client.put(exclude_url, json={**exclude_body, "is_exclude": False})
|
||
assert included.status_code == HTTPStatus.OK, included.text
|
||
assert all(not item["is_exclude"] for item in included.json())
|
||
|
||
attached = await client.put(attach_url, json=attach_body)
|
||
assert attached.status_code == HTTPStatus.OK, attached.text
|
||
assert attached.json()[0]["action"] == "ASSIGN"
|
||
assert attached.json()[0]["rule_id"] == str(rule.id)
|
||
assert attached.json()[0]["is_exclude"] is False
|
||
assert {item["id"] for item in attached.json()} == {item["id"] for item in excluded.json()}
|
||
assert len(attached.json()) == len(entry_ids)
|
||
|
||
detached = await client.request(
|
||
"DELETE", attach_url, json={"statement_type": "balance", "entry_ids": entry_id_strings}
|
||
)
|
||
assert detached.status_code == HTTPStatus.OK, detached.text
|
||
assert {item["id"] for item in detached.json()} == {item["id"] for item in attached.json()}
|
||
assert all(item["action"] == "UNASSIGNED" and item["rule_id"] is None for item in detached.json())
|
||
assert all(item["is_exclude"] for item in detached.json())
|
||
excluded_after_detach = await client.put(exclude_url, json=exclude_body)
|
||
assert excluded_after_detach.status_code == HTTPStatus.OK, excluded_after_detach.text
|
||
attached = await client.put(attach_url, json=attach_body)
|
||
assert attached.status_code == HTTPStatus.OK, attached.text
|
||
assert all(not item["is_exclude"] for item in attached.json())
|
||
|
||
forbidden = await client.put(exclude_url, json=exclude_body)
|
||
assert forbidden.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
|
||
assert forbidden.json()["detail"]["code"] == "VGO_RULE_ASSIGNED"
|
||
partial = await client.put(
|
||
attach_url, json={"entry_ids": [str(entry_ids[0])], "rule_id": str(rule.id)}
|
||
)
|
||
assert partial.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
|
||
|
||
invalid_id = await client.put(
|
||
attach_url, json={"entry_ids": [str(uuid.uuid4())], "rule_id": str(rule.id)}
|
||
)
|
||
assert invalid_id.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
|