"""Round-56 regression: budget matching + order-item delete snapshot. * ``find_matching_budget`` must not raise ``MultipleResultsFound`` when two NULL-scoped budgets coexist (the DB UNIQUE is loose on NULLs) — before the fix it crashed inside the create_order best-effort wrapper, silently killing every budget lock for that scope. * ``delete_order_item`` must return a full content snapshot so a deleted PO line (incl. receipt history) is reconstructable from the audit row. """ from __future__ import annotations import uuid from decimal import Decimal import pytest from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.domains.procurement import service from app.domains.procurement.budget_service import find_matching_budget from app.domains.procurement.models import ( ProcurementBudget, ProcurementBudgetConsumption, ProcurementOrderItem, ) from app.domains.procurement.schemas import ( OrderCreate, OrderItemCreate, OrderUpdate, ) from app.identity.person_models import IdentityPerson pytestmark = pytest.mark.asyncio async def _person(db: AsyncSession) -> IdentityPerson: p = IdentityPerson( id=uuid.uuid4(), work_email=f"a-{uuid.uuid4().hex[:8]}@x.com", display_name="A", employment_status="active", source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", ) db.add(p) await db.flush() return p async def _budget(db: AsyncSession, *, dept: str, currency: str, total: str) -> ProcurementBudget: b = ProcurementBudget( id=str(uuid.uuid4()), year=2026, department=dept, currency=currency, total_amount=Decimal(total), is_active=True, ) db.add(b) await db.flush() return b async def _releases_for(db: AsyncSession, po_id) -> list[ProcurementBudgetConsumption]: return (await db.execute( select(ProcurementBudgetConsumption).where( ProcurementBudgetConsumption.po_id == str(po_id), ProcurementBudgetConsumption.event == "release", ) )).scalars().all() async def test_create_order_survives_budget_statement_error( db_session: AsyncSession, monkeypatch, ): """Regression (#1644): a statement-level failure inside ``lock_budget_for_order`` must NOT poison the outer session and silently drop the whole PO. The SAVEPOINT rolls back only the nested budget scope; the order persists and a subsequent commit succeeds (no PendingRollbackError).""" from sqlalchemy import text from app.domains.procurement import budget_service async def _boom(db, *_a, **_kw): # Genuine statement error: without a SAVEPOINT this leaves the # async session in pending-rollback state. await db.execute(text("SELECT * FROM __helix_no_such_table__")) monkeypatch.setattr(budget_service, "lock_budget_for_order", _boom) applicant = await _person(db_session) order = await service.create_order( db_session, OrderCreate( order_type="purchase", currency="CNY", total_amount=Decimal("500.00"), items=[OrderItemCreate(description="Dock", quantity=Decimal("1"))], ), applicant_id=applicant.id, actor=applicant, ) # The order object came back intact despite the budget lock blowing up. assert order.id is not None # The session is NOT poisoned — the router's commit would succeed. await db_session.commit() # The order row genuinely persisted (not silently rolled back). from app.domains.procurement.models import ProcurementOrder fresh = await db_session.get(ProcurementOrder, str(order.id)) assert fresh is not None assert fresh.total_amount == Decimal("500.00") async def test_find_matching_budget_tolerates_null_scoped_duplicates( db_session: AsyncSession, ): # Two active budgets, identical (year, dept, currency), both with # NULL cost_center + NULL category — only possible because the DB # UNIQUE treats NULLs as distinct. Inserted directly to bypass the # create_budget dedup guard, reproducing legacy/imported data. dept = f"DEPT-DUP-{uuid.uuid4().hex[:6]}" for total in ("1000.00", "2000.00"): db_session.add(ProcurementBudget( id=str(uuid.uuid4()), year=2026, department=dept, currency="CNY", total_amount=Decimal(total), is_active=True, )) await db_session.flush() # Must return exactly one (deterministic newest), NOT raise. match = await find_matching_budget( db_session, year=2026, department=dept, cost_center=None, category_id=None, currency="CNY", ) assert match is not None assert match.department == dept async def test_delete_order_item_returns_content_snapshot( db_session: AsyncSession, ): applicant_obj = IdentityPerson( id=uuid.uuid4(), work_email=f"applicant-{uuid.uuid4().hex[:8]}@x.com", display_name="Applicant", employment_status="active", source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", ) db_session.add(applicant_obj) await db_session.flush() applicant = applicant_obj.id order = await service.create_order( db_session, OrderCreate( order_type="purchase", currency="CNY", total_amount=Decimal("100.00"), items=[OrderItemCreate( description="Laptop dock", quantity=Decimal("2"), unit_price=Decimal("50.00"), )], ), applicant_id=applicant, actor=applicant_obj, ) await db_session.flush() item = (await db_session.execute( select(ProcurementOrderItem) .where(ProcurementOrderItem.order_id == str(order.id)) )).scalars().first() assert item is not None snapshot = await service.delete_order_item( db_session, item.id, actor=applicant_obj, ) assert snapshot["description"] == "Laptop dock" assert snapshot["quantity"] == "2.00" assert snapshot["unit_price"] == "50.00" assert "receipt_status" in snapshot assert snapshot["id"] == str(item.id) async def _po(db: AsyncSession, *, dept: str, currency: str, amount: str): applicant = await _person(db) order = await service.create_order( db, OrderCreate( order_type="purchase", department=dept, currency=currency, total_amount=Decimal(amount), items=[OrderItemCreate(description="x", quantity=Decimal("1"))], ), applicant_id=applicant.id, actor=applicant, ) await db.flush() return order async def test_scope_currency_change_release_keeps_old_currency( db_session: AsyncSession, ): """When a PATCH changes the PO currency (a scope move), the release on the OLD pool must be tagged with the lock currency (CNY), not the new currency — otherwise the pool's per-currency ledger is corrupted.""" dept = f"DEPT-CCY-{uuid.uuid4().hex[:6]}" await _budget(db_session, dept=dept, currency="CNY", total="10000") order = await _po(db_session, dept=dept, currency="CNY", amount="1200") assert order.budget_id is not None # locked against the CNY pool # Flip CNY → USD (no USD pool exists) → old pool released, detached. await service.update_order( db_session, order.id, OrderUpdate(currency="USD"), actor=await _person(db_session), ) releases = await _releases_for(db_session, order.id) assert len(releases) == 1 assert releases[0].currency == "CNY" # NOT 'USD' async def test_cancel_with_scope_change_keeps_budget_trace( db_session: AsyncSession, ): """A PATCH that cancels AND changes scope must NOT migrate pools: it releases the current pool, keeps budget_id, and stamps budget_released_at so the on-PO trace survives (and the release stays in the lock currency).""" dept = f"DEPT-CXL-{uuid.uuid4().hex[:6]}" await _budget(db_session, dept=dept, currency="CNY", total="5000") order = await _po(db_session, dept=dept, currency="CNY", amount="1200") locked_budget_id = order.budget_id assert locked_budget_id is not None await service.update_order( db_session, order.id, OrderUpdate(status="cancelled", department=f"{dept}-MOVED"), actor=await _person(db_session), ) assert order.budget_id == locked_budget_id # link preserved assert order.budget_released_at is not None # release stamped releases = await _releases_for(db_session, order.id) assert len(releases) == 1 assert releases[0].currency == "CNY"