"""POST /api/v1/assets/bulk-restore — recycle-bin batch restore. Covers the platform batch-endpoint convention plus the ``IdempotencyService`` replay contract: * all rows succeed → assets undeleted, ``restored`` history rows written * mixed batch → not-deleted row is ``skipped``, missing id is ``failed``, the good row is ``ok`` (row isolation) * same idempotency_key replay → identical response body, NO re-execution (history rows / audit rows not duplicated) * employee → 403 (require_asset_admin gate) * exactly ONE aggregate audit row (``asset.bulk_restore``) per call """ from __future__ import annotations import uuid from datetime import UTC, datetime import pytest from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.models import AuditLog from app.domains.asset.models import AssetHistory, AssetItem async def _mk_deleted_asset( db: AsyncSession, *, tag: str, deleted: bool = True, ) -> AssetItem: asset = AssetItem( asset_tag=tag, asset_name=f"Bulk restore test {tag}", category="laptop", status="idle", deleted_at=datetime.now(UTC) if deleted else None, ) db.add(asset) await db.commit() return asset async def _post_bulk_restore(client, headers, asset_ids, key): return await client.post( "/api/v1/assets/bulk-restore", json={"asset_ids": asset_ids, "idempotency_key": key}, headers=headers, ) async def _restore_history_count(db: AsyncSession, asset_id) -> int: return ( await db.execute( select(func.count()).select_from(AssetHistory).where( AssetHistory.asset_id == asset_id, AssetHistory.action == "restored", ) ) ).scalar() or 0 @pytest.mark.asyncio async def test_bulk_restore_all_success( client, admin_headers, db_session: AsyncSession, ): a = await _mk_deleted_asset(db_session, tag="BR-OK-001") b = await _mk_deleted_asset(db_session, tag="BR-OK-002") resp = await _post_bulk_restore( client, admin_headers, [str(a.id), str(b.id)], str(uuid.uuid4()), ) assert resp.status_code == 200, resp.text data = resp.json()["data"] assert data["succeeded"] == 2 assert data["skipped"] == 0 assert data["failed"] == 0 assert {r["status"] for r in data["results"]} == {"ok"} await db_session.refresh(a) await db_session.refresh(b) assert a.deleted_at is None assert b.deleted_at is None assert await _restore_history_count(db_session, a.id) == 1 assert await _restore_history_count(db_session, b.id) == 1 @pytest.mark.asyncio async def test_bulk_restore_mixed_results( client, admin_headers, db_session: AsyncSession, ): """Not-in-recycle-bin → skipped; missing id → failed; good row → ok.""" live = await _mk_deleted_asset(db_session, tag="BR-LIVE-001", deleted=False) good = await _mk_deleted_asset(db_session, tag="BR-GOOD-001") missing_id = str(uuid.uuid4()) resp = await _post_bulk_restore( client, admin_headers, [str(live.id), missing_id, str(good.id)], str(uuid.uuid4()), ) assert resp.status_code == 200, resp.text data = resp.json()["data"] assert data["succeeded"] == 1 assert data["skipped"] == 1 assert data["failed"] == 1 by_id = {r["id"]: r for r in data["results"]} assert by_id[str(live.id)]["status"] == "skipped" assert by_id[missing_id]["status"] == "failed" assert by_id[missing_id]["error"] assert by_id[str(good.id)]["status"] == "ok" await db_session.refresh(good) assert good.deleted_at is None # The skipped row got no history entry — nothing was executed for it. assert await _restore_history_count(db_session, live.id) == 0 @pytest.mark.asyncio async def test_bulk_restore_idempotency_replay( client, admin_headers, db_session: AsyncSession, ): """Same key replay returns the cached body and re-executes nothing.""" a = await _mk_deleted_asset(db_session, tag="BR-IDEM-001") key = str(uuid.uuid4()) first = await _post_bulk_restore(client, admin_headers, [str(a.id)], key) assert first.status_code == 200, first.text assert first.json()["data"]["succeeded"] == 1 second = await _post_bulk_restore(client, admin_headers, [str(a.id)], key) assert second.status_code == 200, second.text # Cached replay: identical body (a fresh run would report skipped=1, # succeeded=0 since the asset is already restored). assert second.json()["data"] == first.json()["data"] # No second execution: exactly one restore-history row and one # aggregate audit row despite two HTTP calls. assert await _restore_history_count(db_session, a.id) == 1 audit_count = ( await db_session.execute( select(func.count()).select_from(AuditLog).where( AuditLog.action == "asset.bulk_restore", AuditLog.detail["idempotency_key"].as_string() == key, ) ) ).scalar() assert audit_count == 1 @pytest.mark.asyncio async def test_bulk_restore_employee_forbidden( client, employee_headers, db_session: AsyncSession, ): a = await _mk_deleted_asset(db_session, tag="BR-EMP-001") resp = await _post_bulk_restore( client, employee_headers, [str(a.id)], str(uuid.uuid4()), ) assert resp.status_code == 403 await db_session.refresh(a) assert a.deleted_at is not None @pytest.mark.asyncio async def test_bulk_restore_writes_single_aggregate_audit_row( client, admin_headers, admin_user, db_session: AsyncSession, ): a = await _mk_deleted_asset(db_session, tag="BR-AUD-001") live = await _mk_deleted_asset(db_session, tag="BR-AUD-002", deleted=False) missing_id = str(uuid.uuid4()) key = str(uuid.uuid4()) resp = await _post_bulk_restore( client, admin_headers, [str(a.id), str(live.id), missing_id], key, ) assert resp.status_code == 200, resp.text audit_user, _ = admin_user audits = ( await db_session.execute( select(AuditLog).where( AuditLog.action == "asset.bulk_restore", AuditLog.actor_user_id == audit_user.id, AuditLog.detail["idempotency_key"].as_string() == key, ) ) ).scalars().all() assert len(audits) == 1 # ONE aggregate row, not one per id detail = audits[0].detail assert detail["total"] == 3 assert detail["succeeded"] == 1 assert detail["skipped"] == 1 assert detail["failed"] == 1 assert detail["ids"] == [str(a.id), str(live.id), missing_id]