"""Office-held mode lifecycle pins (migration 183, 2026-05-19). Pin the invariants and lifecycle transitions for the new ``holder_kind='office'`` mode so a refactor cannot silently regress them. Mirror format of ``test_held_by_leaver_2026_05_12.py``. T1 — ``recycle_to_office`` transitions a personal-held asset to ``status='held_by_office'`` + ``holder_kind='office'`` + ``holder_office_id=`` and clears ``assigned_to``. Operation + history rows written. T2 — ``release_from_office`` is the inverse: flips an office-held asset back to ``idle`` / ``personal``, clears ``holder_office_id``. T3 — Claim accepts ``held_by_office`` as a starting status and clears holder_kind + holder_office_id on transition. T4 — Return with ``return_target='office'`` lands the asset at ``held_by_office`` under the returner's home office. T5 — Dispose clears ``holder_kind`` + ``holder_office_id`` on transition so the disposed row carries no stale office attribution. T6 — Invariant: creating an asset with ``holder_kind='office'`` and no ``holder_office_id`` raises ``ASSET_OFFICE_REQUIRED``. T7 — ``recycle_to_office`` rejects ``borrowed`` and ``held_by_leaver`` source states (must clear those first). T8 — Idempotent no-op: calling ``recycle_to_office`` on an asset already at the target office returns success without mutating fields beyond the operation log. """ from __future__ import annotations import uuid from datetime import UTC, datetime import pytest from app.core.exceptions import AppException from app.domains.asset.models import ( ASSET_HOLDER_KIND_OFFICE, ASSET_HOLDER_KIND_PERSONAL, ASSET_STATUS_HELD_BY_LEAVER, ASSET_STATUS_HELD_BY_OFFICE, AssetItem, AssetOfficeAddress, ) from app.domains.asset.schemas import ( AssetBulkRecycleToOfficeRequest, AssetClaimRequest, AssetCreateRequest, AssetDisposeRequest, AssetReturnRequest, ) from app.domains.asset.service import AssetService from app.identity.person_models import IdentityPerson # ───────────────────────── helpers ───────────────────────── async def _make_person( db_session, *, status: str = "active", work_location_name: str | None = None, ) -> IdentityPerson: p = IdentityPerson( id=uuid.uuid4(), source_system="test", hris_person_id=f"P{uuid.uuid4().hex[:8]}", employee_no=f"E{uuid.uuid4().hex[:8]}", work_email=f"u{uuid.uuid4().hex[:8]}@x.test", display_name="Test Person", employment_status=status, work_location_name=work_location_name, ) db_session.add(p) await db_session.flush() return p async def _make_office( db_session, *, city: str = "Shanghai", ) -> AssetOfficeAddress: o = AssetOfficeAddress( id=uuid.uuid4(), city=city, office_name=f"Office {city}", country_or_region="Test", is_active=True, ) db_session.add(o) await db_session.flush() return o async def _make_asset( db_session, *, assigned_to=None, status: str = "claimed", holder_kind: str = ASSET_HOLDER_KIND_PERSONAL, holder_office_id=None, ) -> AssetItem: a = AssetItem( id=uuid.uuid4(), asset_tag=f"AT-{uuid.uuid4().hex[:6]}", asset_name="Test laptop", category="laptop", status=status, assigned_to=assigned_to, assigned_at=datetime.now(UTC) if assigned_to else None, holder_kind=holder_kind, holder_office_id=holder_office_id, ) db_session.add(a) await db_session.flush() return a # ───────────────────────── T1 ───────────────────────── @pytest.mark.asyncio async def test_recycle_to_office_transitions_personal_to_office(db_session): """T1 — recycle a claimed personal-held asset into office mode.""" holder = await _make_person(db_session) operator = await _make_person(db_session) office = await _make_office(db_session, city="Beijing") asset = await _make_asset( db_session, assigned_to=holder.id, status="claimed", ) svc = AssetService(db_session) updated, op = await svc.recycle_to_office( asset.id, office_id=office.id, reason="Holder changed roles", operator_id=operator.id, ) await db_session.refresh(updated) assert updated.status == ASSET_STATUS_HELD_BY_OFFICE assert updated.holder_kind == ASSET_HOLDER_KIND_OFFICE assert updated.holder_office_id == office.id assert updated.assigned_to is None assert updated.assigned_at is None assert op.operation_type == "recycle_to_office" assert op.from_user_id == holder.id # previous holder captured # ───────────────────────── T2 ───────────────────────── @pytest.mark.asyncio async def test_release_from_office_returns_to_idle(db_session): """T2 — release flips office-held back to idle / personal.""" operator = await _make_person(db_session) office = await _make_office(db_session, city="Shanghai") asset = await _make_asset( db_session, status=ASSET_STATUS_HELD_BY_OFFICE, holder_kind=ASSET_HOLDER_KIND_OFFICE, holder_office_id=office.id, ) svc = AssetService(db_session) updated, op = await svc.release_from_office( asset.id, notes="Shipping to remote new hire", operator_id=operator.id, ) await db_session.refresh(updated) assert updated.status == "idle" assert updated.holder_kind == ASSET_HOLDER_KIND_PERSONAL assert updated.holder_office_id is None assert op.operation_type == "release_from_office" # ───────────────────────── T3 ───────────────────────── @pytest.mark.asyncio async def test_claim_from_held_by_office_clears_holder(db_session): """T3 — Claim accepts held_by_office and clears holder fields.""" new_holder = await _make_person(db_session) operator = await _make_person(db_session) office = await _make_office(db_session, city="Hangzhou") asset = await _make_asset( db_session, status=ASSET_STATUS_HELD_BY_OFFICE, holder_kind=ASSET_HOLDER_KIND_OFFICE, holder_office_id=office.id, ) svc = AssetService(db_session) updated, _op = await svc.claim_asset( asset.id, AssetClaimRequest(to_user_id=new_holder.id), operator_id=operator.id, ) await db_session.refresh(updated) assert updated.status == "claimed" assert updated.assigned_to == new_holder.id assert updated.holder_kind == ASSET_HOLDER_KIND_PERSONAL assert updated.holder_office_id is None # ───────────────────────── T4 ───────────────────────── @pytest.mark.asyncio async def test_return_with_office_target_lands_held_by_office(db_session): """T4 — return_target='office' + explicit office routes through held_by_office instead of idle.""" holder = await _make_person(db_session) operator = await _make_person(db_session) office = await _make_office(db_session, city="Shenzhen") asset = await _make_asset( db_session, assigned_to=holder.id, status="claimed", ) svc = AssetService(db_session) req = AssetReturnRequest( return_target="office", target_office_id=office.id, ) updated, op = await svc.return_asset( asset.id, req, operator_id=operator.id, ) await db_session.refresh(updated) assert updated.status == ASSET_STATUS_HELD_BY_OFFICE assert updated.holder_kind == ASSET_HOLDER_KIND_OFFICE assert updated.holder_office_id == office.id assert updated.assigned_to is None assert op.detail is not None assert op.detail.get("return_target") == "office" assert op.detail.get("target_office_id") == str(office.id) @pytest.mark.asyncio async def test_return_office_falls_back_to_idle_when_no_office(db_session): """T4b — return_target='office' with no resolvable office degrades to idle (the schema rejects an explicit non-office id, but the home-office-inferred path can legitimately yield None).""" holder = await _make_person(db_session) # no work_location_name operator = await _make_person(db_session) asset = await _make_asset( db_session, assigned_to=holder.id, status="claimed", ) svc = AssetService(db_session) req = AssetReturnRequest(return_target="office") # no target_office_id updated, _op = await svc.return_asset( asset.id, req, operator_id=operator.id, ) await db_session.refresh(updated) assert updated.status == "idle" assert updated.holder_kind == ASSET_HOLDER_KIND_PERSONAL # ───────────────────────── T5 ───────────────────────── @pytest.mark.asyncio async def test_dispose_clears_holder_fields(db_session): """T5 — disposing an office-held asset clears office attribution.""" operator = await _make_person(db_session) office = await _make_office(db_session, city="Tokyo") asset = await _make_asset( db_session, status=ASSET_STATUS_HELD_BY_OFFICE, holder_kind=ASSET_HOLDER_KIND_OFFICE, holder_office_id=office.id, ) # Dispose service-layer guard rejects HELD_BY_OFFICE without # ``force=True`` because the status isn't in the (assigned, claimed, # borrowed) IN_USE list. Pass force=True to mirror the # offboarding-buyback / write-off path. svc = AssetService(db_session) updated, _op = await svc.dispose_asset( asset.id, AssetDisposeRequest(disposal_type="scrapped", notes=None), operator_id=operator.id, force=True, ) await db_session.refresh(updated) assert updated.status == "disposed" assert updated.holder_kind == ASSET_HOLDER_KIND_PERSONAL assert updated.holder_office_id is None # ───────────────────────── T6 ───────────────────────── @pytest.mark.asyncio async def test_create_office_held_requires_office_id(db_session): """T6 — invariant: holder_kind='office' without holder_office_id is rejected at create time.""" operator = await _make_person(db_session) svc = AssetService(db_session) with pytest.raises(AppException) as exc: await svc.create_asset( AssetCreateRequest( asset_tag=f"AT-{uuid.uuid4().hex[:6]}", category="laptop", holder_kind="office", ), operator_id=operator.id, ) assert exc.value.code == "ASSET_OFFICE_REQUIRED" @pytest.mark.asyncio async def test_create_office_held_sets_office_status(db_session): """T6b — happy path: holder_kind='office' + valid office_id lands the row at held_by_office with no assignee.""" operator = await _make_person(db_session) office = await _make_office(db_session, city="Guangzhou") svc = AssetService(db_session) created = await svc.create_asset( AssetCreateRequest( asset_tag=f"AT-{uuid.uuid4().hex[:6]}", category="laptop", holder_kind="office", holder_office_id=office.id, ), operator_id=operator.id, ) await db_session.refresh(created) assert created.status == ASSET_STATUS_HELD_BY_OFFICE assert created.holder_kind == ASSET_HOLDER_KIND_OFFICE assert created.holder_office_id == office.id assert created.assigned_to is None # ───────────────────────── T7 ───────────────────────── @pytest.mark.asyncio async def test_recycle_to_office_rejects_held_by_leaver(db_session): """T7 — held_by_leaver must be recovered before recycle.""" leaver = await _make_person(db_session, status="left") operator = await _make_person(db_session) office = await _make_office(db_session, city="Chengdu") asset = await _make_asset( db_session, assigned_to=leaver.id, status=ASSET_STATUS_HELD_BY_LEAVER, ) svc = AssetService(db_session) with pytest.raises(AppException) as exc: await svc.recycle_to_office( asset.id, office_id=office.id, reason="Move to office", operator_id=operator.id, ) assert exc.value.code == "ASSET_HELD_BY_LEAVER" # ───────────────────────── T8 ───────────────────────── @pytest.mark.asyncio async def test_recycle_to_office_is_idempotent_for_same_office(db_session): """T8 — recycling to the same office twice returns a no-op operation marked idempotent (no field churn beyond the audit row).""" operator = await _make_person(db_session) office = await _make_office(db_session, city="Kuala Lumpur") asset = await _make_asset( db_session, status=ASSET_STATUS_HELD_BY_OFFICE, holder_kind=ASSET_HOLDER_KIND_OFFICE, holder_office_id=office.id, ) svc = AssetService(db_session) _updated, op = await svc.recycle_to_office( asset.id, office_id=office.id, reason="Repeat", operator_id=operator.id, ) assert op is not None # ───────────────────────── T9 (PR-#406 review regression) ───── # Pin the production-bug fixes surfaced by the code-review pass on # 2026-05-19. Each test below maps 1:1 to a bug code in the review # (P0-#N / P1-#N) so a future refactor that regresses the bug also # fails its named test. @pytest.mark.asyncio async def test_resolve_office_for_person_with_home_office(db_session): """P0-1 regression — exercises the previously-broken ``app.domains.identity.models`` import path. Pre-fix this raised ImportError every time ``resolve_office_for_person`` was called with a person who has a work_location_name. Test was missing in the original PR because the only test that called this method used work_location_name=None, which short-circuits before the import. """ leaver = await _make_person( db_session, work_location_name="Beijing", ) office = await _make_office(db_session, city="Beijing") svc = AssetService(db_session) resolved = await svc.resolve_office_for_person(leaver.id) assert resolved is not None assert resolved.id == office.id @pytest.mark.asyncio async def test_recycle_to_office_rejects_pending_status(db_session): """P2-10 regression — explicit whitelist must reject in-flight pending statuses, not silently accept them via the prior denylist. """ holder = await _make_person(db_session) operator = await _make_person(db_session) office = await _make_office(db_session, city="Beijing") asset = await _make_asset( db_session, assigned_to=holder.id, status="claim_pending", ) svc = AssetService(db_session) with pytest.raises(AppException) as exc: await svc.recycle_to_office( asset.id, office_id=office.id, reason="Try while pending", operator_id=operator.id, ) assert exc.value.code == "ASSET_NOT_RECYCLABLE" @pytest.mark.asyncio async def test_recycle_to_office_noop_does_not_insert_duplicate_operation( db_session, ): """P2-12 regression — repeat noop calls must NOT each insert a fresh AssetOperation row; the second call should return the operation from the first. """ from sqlalchemy import select as _select # noqa: PLC0415 from app.domains.asset.models import AssetOperation # noqa: PLC0415 operator = await _make_person(db_session) office = await _make_office(db_session, city="Kuala Lumpur") asset = await _make_asset( db_session, status=ASSET_STATUS_HELD_BY_OFFICE, holder_kind=ASSET_HOLDER_KIND_OFFICE, holder_office_id=office.id, ) svc = AssetService(db_session) # First call writes a synthesised noop row (no prior recycle). _a1, op1 = await svc.recycle_to_office( asset.id, office_id=office.id, reason="First", operator_id=operator.id, ) # Second call must return op1, NOT create a new operation row. _a2, op2 = await svc.recycle_to_office( asset.id, office_id=office.id, reason="Second", operator_id=operator.id, ) assert op2.id == op1.id rows = ( await db_session.execute( _select(AssetOperation).where(AssetOperation.asset_id == asset.id) ) ).scalars().all() assert len(rows) == 1, ( f"expected single recycle op row, got {len(rows)} — " "noop path is inserting duplicates" ) @pytest.mark.asyncio async def test_recycle_to_office_syncs_location_to_office_city(db_session): """Recycling to an office MUST overwrite the free-text ``location`` with the target office's city so inventory group-bys, location filters, and the legacy office-routing fallback don't keep pointing at the prior holder's city after the device physically moved. The PR #597 managed-location resolver routes ``office.city`` through the asset_locations master before the sync, so the test uses the canonical Chinese names (which the prod master carries) rather than English placeholders. """ # Seed the managed-location master so the resolver gate added # in PR #597 (server/app/domains/asset/location_resolver.py) # recognises both the prior and the new location values. from app.domains.asset.models import AssetLocation for code, name in [("P-CN03", "北京"), ("P-CN07", "上海")]: db_session.add(AssetLocation(code=code, name=name, is_active=True)) await db_session.flush() holder = await _make_person(db_session) operator = await _make_person(db_session) office = await _make_office(db_session, city="上海") asset = await _make_asset( db_session, assigned_to=holder.id, status="claimed", ) asset.location = "北京" # prior holder's city await db_session.flush() svc = AssetService(db_session) updated, _op = await svc.recycle_to_office( asset.id, office_id=office.id, reason="Relocated to Shanghai office", operator_id=operator.id, ) await db_session.refresh(updated) assert updated.holder_office_id == office.id assert updated.location == "上海", ( f"location must follow holder_office_id; got {updated.location!r}" ) # ───────────────────────── T9 (regression) ───────────────────────── @pytest.mark.asyncio async def test_bulk_recycle_to_office_reaches_redis_lock(db_session): """T9 — bulk recycle must reach the Redis idempotency lock via the correct module path and transition every row. Regression for a prod bug: ``_acquire_bulk_idempotency_lock`` imported ``app.core.redis`` (real module is ``app.core.redis_client``). The broad ``except Exception`` swallowed the ``ModuleNotFoundError`` and re-raised it as a 503 ``IDEMPOTENCY_LOCK_UNAVAILABLE``, so EVERY bulk recycle 503'd in prod while single recycle (no lock) worked. A broken import here would raise inside this call instead of completing the batch. """ h1 = await _make_person(db_session) h2 = await _make_person(db_session) operator = await _make_person(db_session) office = await _make_office(db_session, city="Beijing") a1 = await _make_asset(db_session, assigned_to=h1.id, status="claimed") a2 = await _make_asset(db_session, assigned_to=h2.id, status="assigned") svc = AssetService(db_session) resp = await svc.bulk_recycle_to_office( AssetBulkRecycleToOfficeRequest( asset_ids=[a1.id, a2.id], office_id=office.id, reason="变更成公共资产", idempotency_key=f"recycle_test_{uuid.uuid4().hex}", ), operator_id=operator.id, ) assert resp.total_rows == 2 assert resp.successful_rows == 2 assert resp.failed_rows == 0 for a in (a1, a2): await db_session.refresh(a) assert a.status == ASSET_STATUS_HELD_BY_OFFICE assert a.holder_office_id == office.id assert a.assigned_to is None