diff --git a/api/alembic/versions/0012_vgo_rule_not_excluded.py b/api/alembic/versions/0012_vgo_rule_not_excluded.py new file mode 100644 index 0000000..260969f --- /dev/null +++ b/api/alembic/versions/0012_vgo_rule_not_excluded.py @@ -0,0 +1,29 @@ +"""Запретить исключение из КФО строк с назначенным правилом ВГО.""" + +import sqlalchemy as sa + +from alembic import op +from src.models.base import SCHEMA + +revision = "0012_vgo_rule_not_excluded" +down_revision = "0011_vgo_assignment" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.execute(sa.text(f'UPDATE "{SCHEMA}".vgo_assignment SET is_exclude = false WHERE rule_id IS NOT NULL')) + op.alter_column("vgo_assignment", "is_exclude", server_default=sa.text("true"), schema=SCHEMA) + op.create_check_constraint( + "ck_vgo_assignment_rule_not_excluded", + "vgo_assignment", + "rule_id IS NULL OR NOT is_exclude", + schema=SCHEMA, + ) + + +def downgrade() -> None: + op.alter_column("vgo_assignment", "is_exclude", server_default=sa.text("false"), schema=SCHEMA) + op.drop_constraint( + "ck_vgo_assignment_rule_not_excluded", "vgo_assignment", type_="check", schema=SCHEMA + ) diff --git a/api/src/api/dependencies/consolidations.py b/api/src/api/dependencies/consolidations.py index cc61291..36cc74f 100644 --- a/api/src/api/dependencies/consolidations.py +++ b/api/src/api/dependencies/consolidations.py @@ -16,7 +16,7 @@ from src.services.consolidations import ConsolidationService from src.services.counterparty_mapping import CounterpartyMappingService from src.services.elimination_rules import EliminationRuleService from src.services.intercompany_matrix import IntercompanyMatrixService -from src.services.vgo import VgoService +from src.services.vgo import VgoAssignmentCrudService, VgoService def get_consolidation_service( @@ -108,3 +108,16 @@ def get_vgo_service( VgoServiceDep = Annotated[VgoService, Depends(get_vgo_service)] + + +def get_vgo_assignment_crud_service( + session: SessionDep, + audit: Annotated[RequestAudit, Depends(get_request_audit)], + scope: Annotated[VisibilityScope, Depends(get_scope)], + osv_service: ConsolidatedOsvServiceDep, +) -> VgoAssignmentCrudService: + """Сервис ручной правки назначений ВГО.""" + return VgoAssignmentCrudService(session, audit, scope, osv_service) + + +VgoAssignmentCrudServiceDep = Annotated[VgoAssignmentCrudService, Depends(get_vgo_assignment_crud_service)] diff --git a/api/src/api/endpoints/consolidations/vgo_assignment.py b/api/src/api/endpoints/consolidations/vgo_assignment.py index 8de7ccc..3057fc1 100644 --- a/api/src/api/endpoints/consolidations/vgo_assignment.py +++ b/api/src/api/endpoints/consolidations/vgo_assignment.py @@ -5,8 +5,14 @@ import uuid from fastapi import APIRouter, Depends from src.api.dependencies.auth import require -from src.api.dependencies.consolidations import VgoServiceDep -from src.schemas.consolidations import VgoAssignmentCandidatesOut +from src.api.dependencies.consolidations import VgoAssignmentCrudServiceDep, VgoServiceDep +from src.schemas.consolidations import ( + VgoAssignmentAttachRuleIn, + VgoAssignmentCandidatesOut, + VgoAssignmentDetachRuleIn, + VgoAssignmentExcludeIn, + VgoAssignmentOut, +) router = APIRouter(prefix="/consolidations/versions") @@ -20,3 +26,42 @@ router = APIRouter(prefix="/consolidations/versions") async def distribute_articles(version_id: uuid.UUID, service: VgoServiceDep) -> VgoAssignmentCandidatesOut: """Возвращает агрегаты и ручные назначения по группам, а также строки без назначения.""" return await service.assignment_candidates(version_id) + + +@router.put( + "/{version_id}/vgo-assignments/rule", + dependencies=[Depends(require("CONSOLIDATION_EDIT"))], + response_model=list[VgoAssignmentOut], + summary="Привязать выбранные строки ОСВ к правилу ВГО", +) +async def attach_rule( + version_id: uuid.UUID, data: VgoAssignmentAttachRuleIn, service: VgoAssignmentCrudServiceDep +) -> list[VgoAssignmentOut]: + """Обновляет назначения для указанных ID и создаёт недостающие записи.""" + return await service.attach_rule(version_id, data) + + +@router.delete( + "/{version_id}/vgo-assignments/rule", + dependencies=[Depends(require("CONSOLIDATION_EDIT"))], + response_model=list[VgoAssignmentOut], + summary="Снять назначение правила ВГО с выбранных строк ОСВ", +) +async def detach_rule( + version_id: uuid.UUID, data: VgoAssignmentDetachRuleIn, service: VgoAssignmentCrudServiceDep +) -> list[VgoAssignmentOut]: + """Оставляет или создаёт записи без правила с is_exclude=true.""" + return await service.detach_rule(version_id, data) + + +@router.put( + "/{version_id}/vgo-assignments/exclude", + dependencies=[Depends(require("CONSOLIDATION_EDIT"))], + response_model=list[VgoAssignmentOut], + summary="Исключить или вернуть выбранные строки ОСВ в КФО", +) +async def set_exclusion( + version_id: uuid.UUID, data: VgoAssignmentExcludeIn, service: VgoAssignmentCrudServiceDep +) -> list[VgoAssignmentOut]: + """Сохраняет is_exclude для выбранных ID; строки с правилом исключать нельзя.""" + return await service.set_exclusion(version_id, data) diff --git a/api/src/models/kfo.py b/api/src/models/kfo.py index 50e4651..8d15cb5 100644 --- a/api/src/models/kfo.py +++ b/api/src/models/kfo.py @@ -34,6 +34,7 @@ class VgoAssignment(UUIDPrimaryMixin, Base): "(action = 'ASSIGN' AND rule_id IS NOT NULL) OR (action = 'UNASSIGNED' AND rule_id IS NULL)", name="action_rule_consistency", ), + CheckConstraint("rule_id IS NULL OR NOT is_exclude", name="rule_not_excluded"), ) consolidation_version_id: Mapped[uuid.UUID] = mapped_column( @@ -47,7 +48,7 @@ class VgoAssignment(UUIDPrimaryMixin, Base): UUID(as_uuid=True), ForeignKey("elimination_rule.id", name="fk_vgo_assignment_rule") ) action: Mapped[str] = mapped_column(String, nullable=False) - is_exclude: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default=text("false")) + is_exclude: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default=text("true")) created_by: Mapped[uuid.UUID | None] = mapped_column( UUID(as_uuid=True), ForeignKey("app_user.id", name="fk_vgo_assignment_created_by") ) diff --git a/api/src/repositories/vgo_assignments.py b/api/src/repositories/vgo_assignments.py index f0c3132..f56abca 100644 --- a/api/src/repositories/vgo_assignments.py +++ b/api/src/repositories/vgo_assignments.py @@ -10,15 +10,23 @@ from src.models.kfo import VgoAssignment async def list_assignments( - session: AsyncSession, version_id: uuid.UUID, statement_type: StatementType + session: AsyncSession, version_id: uuid.UUID, statement_type: StatementType | None = None ) -> list[VgoAssignment]: - """Все ручные решения для версии и типа отчётности.""" - stmt = ( - select(VgoAssignment) - .where( - VgoAssignment.consolidation_version_id == version_id, - VgoAssignment.statement_type == statement_type, - ) - .order_by(VgoAssignment.created_at, VgoAssignment.id) - ) + """Все ручные решения версии, при необходимости ограниченные типом отчётности.""" + stmt = select(VgoAssignment).where(VgoAssignment.consolidation_version_id == version_id) + if statement_type is not None: + stmt = stmt.where(VgoAssignment.statement_type == statement_type) + stmt = stmt.order_by(VgoAssignment.created_at, VgoAssignment.id) return list((await session.scalars(stmt)).all()) + + +async def for_entries( + session: AsyncSession, version_id: uuid.UUID, statement_type: StatementType, entry_ids: list[uuid.UUID] +) -> dict[uuid.UUID, VgoAssignment]: + """Решения для конкретных исходных строк ОСВ.""" + stmt = select(VgoAssignment).where( + VgoAssignment.consolidation_version_id == version_id, + VgoAssignment.statement_type == statement_type, + VgoAssignment.osv_entry_id.in_(entry_ids), + ) + return {item.osv_entry_id: item for item in (await session.scalars(stmt)).all()} diff --git a/api/src/schemas/consolidations.py b/api/src/schemas/consolidations.py index b20e047..15e1729 100644 --- a/api/src/schemas/consolidations.py +++ b/api/src/schemas/consolidations.py @@ -430,7 +430,7 @@ class VgoAssignmentRowOut(ConsolidatedRowOut): osv_entry_id: uuid.UUID | None = None action: Literal["ASSIGN", "UNASSIGNED"] | None = None - is_exclude: bool | None = None + is_exclude: bool = True class VgoAssignmentGroupOut(BaseModel): @@ -466,6 +466,65 @@ class VgoAssignmentCandidatesOut(BaseModel): unassigned: list[VgoAssignmentRowOut] +class VgoAssignmentAttachRuleIn(BaseModel): + """Привязать все исходные ID выбранных строк сводной ОСВ к правилу.""" + + entry_ids: list[uuid.UUID] = Field(min_length=1) + rule_id: uuid.UUID + + @field_validator("entry_ids") + @classmethod + def unique_entries(cls, value: list[uuid.UUID]) -> list[uuid.UUID]: + if len(value) != len(set(value)): + raise ValueError("entry_ids не должны повторяться") + return value + + +class VgoAssignmentExcludeIn(BaseModel): + """Включить или исключить выбранные строки сводной ОСВ из КФО.""" + + statement_type: StatementType + entry_ids: list[uuid.UUID] = Field(min_length=1) + is_exclude: bool + + @field_validator("entry_ids") + @classmethod + def unique_entries(cls, value: list[uuid.UUID]) -> list[uuid.UUID]: + if len(value) != len(set(value)): + raise ValueError("entry_ids не должны повторяться") + return value + + +class VgoAssignmentDetachRuleIn(BaseModel): + """Исходные ID выбранных строк сводной ОСВ для снятия правила.""" + + statement_type: StatementType + entry_ids: list[uuid.UUID] = Field(min_length=1) + + @field_validator("entry_ids") + @classmethod + def unique_entries(cls, value: list[uuid.UUID]) -> list[uuid.UUID]: + if len(value) != len(set(value)): + raise ValueError("entry_ids не должны повторяться") + return value + + +class VgoAssignmentOut(BaseModel): + """Сохранённое решение для одной исходной строки ОСВ.""" + + id: uuid.UUID + consolidation_version_id: uuid.UUID + statement_type: StatementType + osv_entry_id: uuid.UUID + rule_id: uuid.UUID | None + action: Literal["ASSIGN", "UNASSIGNED"] + is_exclude: bool + created_by: uuid.UUID | None + created_at: datetime.datetime + updated_by: uuid.UUID | None + updated_at: datetime.datetime | None + + EliminationTotalStatus = Literal["AUTO", "DISCREPANCY", "ONE_SIDED", "MANUAL"] diff --git a/api/src/services/vgo.py b/api/src/services/vgo.py index d3e259e..3e5dfd2 100644 --- a/api/src/services/vgo.py +++ b/api/src/services/vgo.py @@ -8,21 +8,29 @@ from decimal import Decimal from sqlalchemy.ext.asyncio import AsyncSession +from src.api.dependencies.audit import RequestAudit +from src.common.clock import utc_now from src.common.exceptions import DomainValidationError from src.domain.consolidated_osv import ConsolidatedRow from src.domain.elimination_packing import calculate_rule_totals +from src.domain.policy import VisibilityScope from src.models.accounting import EliminationRule -from src.models.enums import StatementType +from src.models.enums import AuditAction, AuditSection, StatementType from src.models.kfo import VgoAssignment from src.repositories import elimination_rules as elimination_rules_repo from src.repositories import vgo_assignments as vgo_assignments_repo from src.schemas.consolidations import ( EliminationMatchedPairOut, + VgoAssignmentAttachRuleIn, VgoAssignmentCandidatesOut, + VgoAssignmentDetachRuleIn, + VgoAssignmentExcludeIn, VgoAssignmentGroupOut, + VgoAssignmentOut, VgoAssignmentRowOut, ) from src.services.consolidated_osv import ConsolidatedOsvBuildResult, ConsolidatedOsvService +from src.services.consolidation_guard import lock_editable_version from src.services.intercompany_matrix import IntercompanyMatrixService ZERO = Decimal("0") @@ -269,3 +277,151 @@ class VgoService: groups = _build_groups(search, diff_aggregates, rules, manual.groups, names) unassigned = _unassigned_rows(osv_result.consolidated.rows, rows, manual, search.used_ids) return VgoAssignmentCandidatesOut(groups=groups, unassigned=unassigned) + + +class VgoAssignmentCrudService: + """Ручная правка назначений исходных строк ОСВ одной версии.""" + + def __init__( + self, session: AsyncSession, audit: RequestAudit, scope: VisibilityScope, osv_service: ConsolidatedOsvService + ) -> None: + self.session = session + self.audit = audit + self.scope = scope + self.osv_service = osv_service + + async def _check_rows(self, version_id: uuid.UUID, entry_ids: list[uuid.UUID]) -> None: + """Проверить, что ID образуют одну или несколько полных строк свода.""" + osv_result = await self.osv_service.build_with_members(version_id) + requested = set(entry_ids) + matched_rows = [row for row in osv_result.consolidated.rows if requested.intersection(row.entry_ids)] + matched_ids = {entry_id for row in matched_rows for entry_id in row.entry_ids} + if requested != matched_ids: + raise DomainValidationError( + "INVALID_OSV_ENTRY_IDS", "entry_ids должны содержать все исходные ID выбранных строк сводной ОСВ" + ) + + async def attach_rule( + self, version_id: uuid.UUID, data: VgoAssignmentAttachRuleIn + ) -> list[VgoAssignmentOut]: + """Привязать выбранные полные строки сводной ОСВ версии к правилу.""" + version, _ = await lock_editable_version(self.session, self.scope, version_id, self.audit.user_id) + rule = await self.session.get(EliminationRule, data.rule_id) + if rule is None or rule.version_no != version.elimination_rules_version: + raise DomainValidationError("INVALID_VGO_RULE", "правило не относится к версии консолидации") + await self._check_rows(version_id, data.entry_ids) + existing = await vgo_assignments_repo.for_entries( + self.session, version_id, rule.statement_type, data.entry_ids + ) + return await self._persist_rule(version_id, rule, data.entry_ids, existing) + + async def set_exclusion( + self, version_id: uuid.UUID, data: VgoAssignmentExcludeIn + ) -> list[VgoAssignmentOut]: + """Установить признак исключения, не меняя назначенное правило.""" + await lock_editable_version(self.session, self.scope, version_id, self.audit.user_id) + await self._check_rows(version_id, data.entry_ids) + existing = await vgo_assignments_repo.for_entries(self.session, version_id, data.statement_type, data.entry_ids) + if data.is_exclude and any(item.rule_id is not None for item in existing.values()): + raise DomainValidationError("VGO_RULE_ASSIGNED", "строку с правилом нельзя исключить из КФО") + + before = {str(entry_id): _assignment_snapshot(item) for entry_id, item in existing.items()} + result = [] + for entry_id in data.entry_ids: + item = existing.get(entry_id) + if item is None: + item = VgoAssignment( + consolidation_version_id=version_id, + statement_type=data.statement_type, + osv_entry_id=entry_id, + action="UNASSIGNED", + is_exclude=data.is_exclude, + created_by=self.audit.user_id, + ) + self.session.add(item) + else: + item.is_exclude = data.is_exclude + item.updated_by = self.audit.user_id + item.updated_at = utc_now() + result.append(item) + await self.session.flush() + await self._audit(version_id, before, {str(item.osv_entry_id): _assignment_snapshot(item) for item in result}) + return [VgoAssignmentOut.model_validate(item, from_attributes=True) for item in result] + + async def detach_rule( + self, version_id: uuid.UUID, data: VgoAssignmentDetachRuleIn + ) -> list[VgoAssignmentOut]: + """Снять правило и создать отсутствующие записи как исключённые.""" + await lock_editable_version(self.session, self.scope, version_id, self.audit.user_id) + await self._check_rows(version_id, data.entry_ids) + existing = await vgo_assignments_repo.for_entries(self.session, version_id, data.statement_type, data.entry_ids) + + before = {str(entry_id): _assignment_snapshot(item) for entry_id, item in existing.items()} + result = [] + for entry_id in data.entry_ids: + item = existing.get(entry_id) + if item is None: + item = VgoAssignment( + consolidation_version_id=version_id, + statement_type=data.statement_type, + osv_entry_id=entry_id, + created_by=self.audit.user_id, + ) + self.session.add(item) + else: + item.updated_by = self.audit.user_id + item.updated_at = utc_now() + item.action = "UNASSIGNED" + item.rule_id = None + item.is_exclude = True + result.append(item) + await self.session.flush() + await self._audit(version_id, before, {str(item.osv_entry_id): _assignment_snapshot(item) for item in result}) + return [VgoAssignmentOut.model_validate(item, from_attributes=True) for item in result] + + async def _persist_rule( + self, + version_id: uuid.UUID, + rule: EliminationRule, + entry_ids: list[uuid.UUID], + existing: dict[uuid.UUID, VgoAssignment], + ) -> list[VgoAssignmentOut]: + """Сохранить привязку к правилу и аудит для проверенных ID.""" + + before = {str(entry_id): _assignment_snapshot(item) for entry_id, item in existing.items()} + result = [] + for entry_id in entry_ids: + item = existing.get(entry_id) + if item is None: + item = VgoAssignment( + consolidation_version_id=version_id, + statement_type=rule.statement_type, + osv_entry_id=entry_id, + created_by=self.audit.user_id, + ) + self.session.add(item) + else: + item.updated_by = self.audit.user_id + item.updated_at = utc_now() + item.action = "ASSIGN" + item.rule_id = rule.id + item.is_exclude = False + result.append(item) + await self.session.flush() + await self._audit(version_id, before, {str(item.osv_entry_id): _assignment_snapshot(item) for item in result}) + return [VgoAssignmentOut.model_validate(item, from_attributes=True) for item in result] + + async def _audit(self, version_id: uuid.UUID, before: Mapping[str, object], after: Mapping[str, object]) -> None: + await self.audit.record( + action=AuditAction.ASSIGNMENT_CHANGE, + section=AuditSection.KFO, + object_type="vgo_assignment", + object_id=version_id, + consolidation_version_id=version_id, + before_value={"entries": before}, + after_value={"entries": after}, + ) + + +def _assignment_snapshot(item: VgoAssignment) -> dict[str, object]: + return {"action": item.action, "rule_id": item.rule_id, "is_exclude": item.is_exclude} diff --git a/api/tests/test_api_vgo_assignments.py b/api/tests/test_api_vgo_assignments.py new file mode 100644 index 0000000..141cae3 --- /dev/null +++ b/api/tests/test_api_vgo_assignments.py @@ -0,0 +1,250 @@ +"""Ручные операции над назначениями ВГО для полных строк сводной ОСВ.""" + +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 diff --git a/api/tests/test_vgo.py b/api/tests/test_vgo.py index 1ffdcf8..eefc949 100644 --- a/api/tests/test_vgo.py +++ b/api/tests/test_vgo.py @@ -153,7 +153,7 @@ async def test_pairs_are_grouped_and_entry_ids_are_used_once(monkeypatch: pytest ] -async def test_manual_assignments_override_auto_and_keep_exclusion_flag(monkeypatch: pytest.MonkeyPatch) -> None: +async def test_manual_assignments_override_auto_and_show_exclusion_flag(monkeypatch: pytest.MonkeyPatch) -> None: version_id, company_a, company_b, rule_id = (uuid.uuid4() for _ in range(4)) entry_ids = [uuid.uuid4() for _ in range(4)] rows = [_row(entry_id, company_a, company_b, index) for index, entry_id in enumerate(entry_ids)] @@ -170,7 +170,7 @@ async def test_manual_assignments_override_auto_and_keep_exclusion_flag(monkeypa ) assignments = [ SimpleNamespace(osv_entry_id=entry_ids[0], rule_id=rule_id, action="ASSIGN", is_exclude=False), - SimpleNamespace(osv_entry_id=entry_ids[1], rule_id=rule_id, action="ASSIGN", is_exclude=True), + SimpleNamespace(osv_entry_id=entry_ids[1], rule_id=rule_id, action="ASSIGN", is_exclude=False), SimpleNamespace(osv_entry_id=entry_ids[2], rule_id=None, action="UNASSIGNED", is_exclude=True), ] monkeypatch.setattr(vgo.elimination_rules_repo, "list_rules", AsyncMock(return_value=[rule])) @@ -195,15 +195,15 @@ async def test_manual_assignments_override_auto_and_keep_exclusion_flag(monkeypa company_a, company_b, ) - assert (response.groups[0].amount_a, response.groups[0].amount_b) == (Decimal("10.00"), Decimal("0.00")) + assert (response.groups[0].amount_a, response.groups[0].amount_b) == (Decimal("30.00"), Decimal("0.00")) assert [(row.osv_entry_id, row.is_exclude) for row in response.groups[0].manual] == [ (entry_ids[0], False), - (entry_ids[1], True), + (entry_ids[1], False), ] assert "manual" in response.model_dump()["groups"][0] assert [(row.entry_ids, row.action, row.is_exclude) for row in response.unassigned] == [ ([entry_ids[2]], "UNASSIGNED", True), - ([entry_ids[3]], None, None), + ([entry_ids[3]], None, True), ]