"""Tests for asset management API endpoints.""" import io from datetime import UTC, date, datetime, timedelta from decimal import Decimal from uuid import uuid4 from zoneinfo import ZoneInfo import pytest from httpx import AsyncClient from sqlalchemy import select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.auth.models import AuthUser from app.domains.asset.models import ( ASSET_HOLDER_KIND_OFFICE, ASSET_HOLDER_KIND_PERSONAL, AssetBrand, AssetCategoryConfig, AssetHistory, AssetItem, AssetLegalEntity, AssetModel, AssetOfficeAddress, AssetOperation, ) from app.domains.asset.service import _asset_duplicate_unique_column class TestAssetAPI: """Test suite for asset management endpoints.""" @pytest.mark.asyncio async def test_admin_can_create_asset( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test that IT Admin can create a new asset.""" payload = { "asset_tag": "LAP-2024-001", "category": "laptop", "brand": "Dell", "model": "Latitude 5520", "serial_number": "SN123456", "status": "available", "purchase_date": "2024-01-15", "purchase_cost": "1299.99", "warranty_until": "2027-01-15", "notes": "Developer laptop", } response = await client.post( "/api/v1/assets", json=payload, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["asset_tag"] == "LAP-2024-001" assert data["category"] == "laptop" assert data["brand"] == "Dell" assert data["status"] == "available" # Verify in database result = await db_session.execute( select(AssetItem).where(AssetItem.asset_tag == "LAP-2024-001") ) asset = result.scalar_one() assert asset.brand == "Dell" assert asset.purchase_cost == Decimal("1299.99") # Verify history entry created history_result = await db_session.execute( select(AssetHistory).where(AssetHistory.asset_id == asset.id) ) history = history_result.scalar_one() assert history.action == "created" @pytest.mark.asyncio async def test_employee_cannot_create_asset( self, client: AsyncClient, employee_headers: dict, ): """Test that regular employee cannot create assets.""" payload = { "asset_tag": "LAP-2024-002", "category": "laptop", "brand": "HP", "model": "EliteBook", } response = await client.post( "/api/v1/assets", json=payload, headers=employee_headers, ) assert response.status_code == 403 assert response.json()["errors"][0]["code"] == "FORBIDDEN" @pytest.mark.asyncio async def test_list_assets_with_pagination( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test listing assets with pagination.""" # Create test assets for i in range(25): asset = AssetItem( asset_tag=f"TEST-{i:03d}", category="laptop" if i % 2 == 0 else "monitor", status="available", brand="TestBrand", ) db_session.add(asset) await db_session.commit() # Get first page response = await client.get( "/api/v1/assets?page=1&page_size=10", headers=admin_headers, ) assert response.status_code == 200 data = response.json() assert len(data["data"]) == 10 assert data["meta"]["total"] >= 25 assert data["meta"]["page"] == 1 assert data["meta"]["page_size"] == 10 # Get second page response = await client.get( "/api/v1/assets?page=2&page_size=10", headers=admin_headers, ) assert response.status_code == 200 data = response.json() assert len(data["data"]) == 10 assert data["meta"]["page"] == 2 @pytest.mark.asyncio async def test_filter_assets_by_category( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test filtering assets by category.""" # Create test assets laptop = AssetItem( asset_tag="LAP-001", category="laptop", status="available", ) monitor = AssetItem( asset_tag="MON-001", category="monitor", status="available", ) db_session.add_all([laptop, monitor]) await db_session.commit() # Filter by laptop response = await client.get( "/api/v1/assets?category=laptop", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert all(item["category"] == "laptop" for item in data) @pytest.mark.asyncio async def test_filter_assets_by_location_and_usage_company( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test filtering assets by operational location and usage company.""" beijing_asset = AssetItem( asset_tag="LOC-001", category="laptop", status="available", location="北京", usage_company="PingCAP", ) singapore_asset = AssetItem( asset_tag="LOC-002", category="laptop", status="available", location="新加坡", usage_company="PingCAP Pte. Ltd.", ) db_session.add_all([beijing_asset, singapore_asset]) await db_session.commit() response = await client.get( "/api/v1/assets?location=北京&usage_company=PingCAP", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert len(data) >= 1 assert all(item["location"] == "北京" for item in data) assert all(item["usage_company"] == "PingCAP" for item in data) @pytest.mark.asyncio async def test_filter_assets_by_multiple_statuses( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test filtering assets by multiple statuses (multi-select).""" assets = [ AssetItem(asset_tag="MS-IDLE", category="laptop", status="idle"), AssetItem(asset_tag="MS-CLAIMED", category="laptop", status="claimed"), AssetItem(asset_tag="MS-BORROWED", category="laptop", status="borrowed"), ] db_session.add_all(assets) await db_session.commit() # Multi-select: idle + claimed response = await client.get( "/api/v1/assets?status=idle&status=claimed", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] returned_statuses = {item["status"] for item in data} assert "idle" in returned_statuses or "claimed" in returned_statuses assert "borrowed" not in returned_statuses @pytest.mark.asyncio async def test_filter_assets_by_multiple_categories( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test filtering assets by multiple categories (multi-select).""" assets = [ AssetItem(asset_tag="MC-LAP", category="laptop", status="idle"), AssetItem(asset_tag="MC-MON", category="monitor", status="idle"), AssetItem(asset_tag="MC-SRV", category="server", status="idle"), ] db_session.add_all(assets) await db_session.commit() # Multi-select: laptop + monitor response = await client.get( "/api/v1/assets?category=laptop&category=monitor", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] returned_categories = {item["category"] for item in data} assert returned_categories <= {"laptop", "monitor"} assert len(data) >= 2 @pytest.mark.asyncio async def test_filter_assets_by_multiple_locations( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test filtering assets by multiple locations (multi-select).""" assets = [ AssetItem(asset_tag="ML-BJ", category="laptop", status="idle", location="北京"), AssetItem(asset_tag="ML-SH", category="laptop", status="idle", location="上海"), AssetItem(asset_tag="ML-SG", category="laptop", status="idle", location="新加坡"), ] db_session.add_all(assets) await db_session.commit() # Multi-select: 北京 + 上海 response = await client.get( "/api/v1/assets?location=北京&location=上海", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] returned_locations = {item["location"] for item in data} assert returned_locations <= {"北京", "上海"} assert len(data) >= 2 @pytest.mark.asyncio async def test_filter_assets_by_multiple_usage_companies( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test filtering assets by multiple usage companies (multi-select).""" assets = [ AssetItem(asset_tag="MU-PC", category="laptop", status="idle", usage_company="PingCAP"), AssetItem(asset_tag="MU-SG", category="laptop", status="idle", usage_company="PingCAP Pte. Ltd."), AssetItem(asset_tag="MU-US", category="laptop", status="idle", usage_company="PingCAP(US) Inc"), ] db_session.add_all(assets) await db_session.commit() # Multi-select: PingCAP + PingCAP Pte. Ltd. response = await client.get( "/api/v1/assets?usage_company=PingCAP&usage_company=PingCAP+Pte.+Ltd.", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] returned_companies = {item["usage_company"] for item in data} assert returned_companies <= {"PingCAP", "PingCAP Pte. Ltd."} assert len(data) >= 2 @pytest.mark.asyncio async def test_filter_assets_by_purchase_entity( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """采购主体 filter — independent from usage_company, multi-value.""" assets = [ AssetItem(asset_tag="PE-PC", category="laptop", status="idle", purchase_entity="PingCAP", usage_company="PingCAP Pte. Ltd."), AssetItem(asset_tag="PE-SG", category="laptop", status="idle", purchase_entity="PingCAP Pte. Ltd.", usage_company="PingCAP"), AssetItem(asset_tag="PE-US", category="laptop", status="idle", purchase_entity="PINGCAP (US), INC.", usage_company="PingCAP"), ] db_session.add_all(assets) await db_session.commit() # Single-value response = await client.get( "/api/v1/assets?purchase_entity=PingCAP", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] tags = {item["asset_tag"] for item in data} assert "PE-PC" in tags assert "PE-SG" not in tags assert "PE-US" not in tags # Multi-value response = await client.get( "/api/v1/assets?purchase_entity=PingCAP&purchase_entity=PingCAP+Pte.+Ltd.", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] returned_entities = {item["purchase_entity"] for item in data} assert returned_entities <= {"PingCAP", "PingCAP Pte. Ltd."} assert "PE-US" not in {item["asset_tag"] for item in data} @pytest.mark.asyncio async def test_multi_filter_combined( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test combining multiple multi-select filters.""" assets = [ AssetItem(asset_tag="CMB-1", category="laptop", status="idle", location="北京", usage_company="PingCAP"), AssetItem(asset_tag="CMB-2", category="monitor", status="claimed", location="北京", usage_company="PingCAP"), AssetItem(asset_tag="CMB-3", category="laptop", status="idle", location="上海", usage_company="PingCAP Pte. Ltd."), AssetItem(asset_tag="CMB-4", category="server", status="borrowed", location="新加坡", usage_company="PingCAP Pte. Ltd."), ] db_session.add_all(assets) await db_session.commit() # status=[idle,claimed] + category=[laptop,monitor] + location=[北京] response = await client.get( "/api/v1/assets?status=idle&status=claimed&category=laptop&category=monitor&location=北京", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] for item in data: assert item["status"] in ("idle", "claimed") assert item["category"] in ("laptop", "monitor") assert item["location"] == "北京" @pytest.mark.asyncio async def test_single_value_filter_still_works( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test that single-value filters remain backward compatible.""" asset = AssetItem(asset_tag="COMPAT-001", category="laptop", status="idle") db_session.add(asset) await db_session.commit() response = await client.get( "/api/v1/assets?status=idle&category=laptop", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert all(item["status"] == "idle" for item in data) assert all(item["category"] == "laptop" for item in data) @pytest.mark.asyncio async def test_filter_assets_by_holder_kind_office( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """holder_kind=office returns only office-held rows; multi-value works; unknown value rejected. """ office = AssetOfficeAddress( id=uuid4(), city="Shanghai", office_name="Office Shanghai", country_or_region="CN", is_active=True, ) db_session.add(office) await db_session.flush() personal = AssetItem( asset_tag="HK-P-1", category="laptop", status="idle", holder_kind=ASSET_HOLDER_KIND_PERSONAL, ) office_held = AssetItem( asset_tag="HK-O-1", category="laptop", status="held_by_office", holder_kind=ASSET_HOLDER_KIND_OFFICE, holder_office_id=office.id, ) db_session.add_all([personal, office_held]) await db_session.commit() # Only office-held r = await client.get( "/api/v1/assets?holder_kind=office", headers=admin_headers, ) assert r.status_code == 200 tags = {item["asset_tag"] for item in r.json()["data"]} assert "HK-O-1" in tags and "HK-P-1" not in tags # Both values selected = no narrowing on kind r = await client.get( "/api/v1/assets?holder_kind=office&holder_kind=personal", headers=admin_headers, ) assert r.status_code == 200 tags = {item["asset_tag"] for item in r.json()["data"]} assert {"HK-O-1", "HK-P-1"}.issubset(tags) # Unknown holder_kind rejected r = await client.get( "/api/v1/assets?holder_kind=bogus", headers=admin_headers, ) assert r.status_code in (400, 422) @pytest.mark.asyncio async def test_filter_assets_by_holder_office_id( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """holder_office_id narrows to specific office(s).""" office_sh = AssetOfficeAddress( id=uuid4(), city="Shanghai", office_name="Office SH", country_or_region="CN", is_active=True, ) office_bj = AssetOfficeAddress( id=uuid4(), city="Beijing", office_name="Office BJ", country_or_region="CN", is_active=True, ) db_session.add_all([office_sh, office_bj]) await db_session.flush() a_sh = AssetItem( asset_tag="HO-SH-1", category="laptop", status="held_by_office", holder_kind=ASSET_HOLDER_KIND_OFFICE, holder_office_id=office_sh.id, ) a_bj = AssetItem( asset_tag="HO-BJ-1", category="laptop", status="held_by_office", holder_kind=ASSET_HOLDER_KIND_OFFICE, holder_office_id=office_bj.id, ) a_personal = AssetItem( asset_tag="HO-P-1", category="laptop", status="idle", holder_kind=ASSET_HOLDER_KIND_PERSONAL, ) db_session.add_all([a_sh, a_bj, a_personal]) await db_session.commit() # Scope to Shanghai office only r = await client.get( f"/api/v1/assets?holder_office_id={office_sh.id}", headers=admin_headers, ) assert r.status_code == 200 tags = {item["asset_tag"] for item in r.json()["data"]} assert "HO-SH-1" in tags assert "HO-BJ-1" not in tags assert "HO-P-1" not in tags # Combine with holder_kind — same office, redundant but valid r = await client.get( f"/api/v1/assets?holder_kind=office&holder_office_id={office_bj.id}", headers=admin_headers, ) assert r.status_code == 200 tags = {item["asset_tag"] for item in r.json()["data"]} assert tags == {"HO-BJ-1"} @pytest.mark.asyncio async def test_search_assets( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test searching assets by tag, brand, model, serial.""" asset = AssetItem( asset_tag="UNIQUE-TAG-999", category="laptop", brand="Apple", model="MacBook Pro", serial_number="UNIQUE-SERIAL", status="available", ) db_session.add(asset) await db_session.commit() # Search by asset tag response = await client.get( "/api/v1/assets?search=UNIQUE-TAG", headers=admin_headers, ) assert response.status_code == 200 assert len(response.json()["data"]) >= 1 # Search by brand response = await client.get( "/api/v1/assets?search=Apple", headers=admin_headers, ) assert response.status_code == 200 assert len(response.json()["data"]) >= 1 # Search by serial response = await client.get( "/api/v1/assets?search=UNIQUE-SERIAL", headers=admin_headers, ) assert response.status_code == 200 assert len(response.json()["data"]) >= 1 @pytest.mark.asyncio async def test_update_asset( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test updating asset details.""" # Create asset asset = AssetItem( asset_tag="UPDATE-TEST", category="laptop", brand="Dell", model="Old Model", status="available", ) db_session.add(asset) await db_session.commit() # Update asset payload = { "model": "New Model", "notes": "Updated notes", } response = await client.patch( f"/api/v1/assets/{asset.id}", json=payload, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["model"] == "New Model" assert data["notes"] == "Updated notes" assert data["brand"] == "Dell" # Unchanged # Verify history await db_session.refresh(asset) history_result = await db_session.execute( select(AssetHistory) .where(AssetHistory.asset_id == asset.id) .where(AssetHistory.action == "updated") ) history = history_result.scalar_one() assert "model:" in history.notes @pytest.mark.asyncio async def test_update_purchase_entity_canonicalizes_legal_suffix_punctuation( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """A pasted ``LTD`` label must persist the registered ``LTD.`` name.""" entity = AssetLegalEntity( name="PINGCAP PTE. LTD.", code="SG", is_active=True, ) asset = AssetItem(asset_tag="ENTITY-CANONICAL-TEST", category="laptop") db_session.add_all([entity, asset]) await db_session.commit() response = await client.patch( f"/api/v1/assets/{asset.id}", json={"purchase_entity": "PINGCAP PTE LTD"}, headers=admin_headers, ) assert response.status_code == 200 assert response.json()["data"]["purchase_entity"] == "PINGCAP PTE. LTD." @pytest.mark.asyncio async def test_update_purchase_entity_rejects_unknown_label( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): asset = AssetItem(asset_tag="ENTITY-UNKNOWN-TEST", category="laptop") db_session.add(asset) await db_session.commit() response = await client.patch( f"/api/v1/assets/{asset.id}", json={"purchase_entity": "Unknown Legal Entity"}, headers=admin_headers, ) assert response.status_code == 400 assert response.json()["errors"][0]["code"] == "PURCHASE_ENTITY_NOT_REGISTERED" @pytest.mark.asyncio async def test_update_asset_brand_and_model_text_persists_when_curated_ids_set( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Regression: editing both ``brand`` and ``model`` text on an asset that already has curated ``brand_id`` / ``model_id`` rows must persist the new text (and rebind to the new brand/model rows). Prior bug: the update path inherited the asset's ``brand_id`` / ``model_id`` into ``resolve_brand_and_model``; that helper's "id wins over name" rule then silently overwrote the typed values with the curated rows' names, the field-level diff saw no change, and the API returned 200 without flushing — "save success" but the dialog reopened with the old values. """ old_brand = AssetBrand(name="Apple") db_session.add(old_brand) await db_session.flush() old_model = AssetModel(brand_id=old_brand.id, name="MacBook Pro 14") db_session.add(old_model) await db_session.flush() asset = AssetItem( asset_tag="P-NB-CURATED-EDIT", category="laptop", brand=old_brand.name, brand_id=old_brand.id, model=old_model.name, model_id=old_model.id, status="available", ) db_session.add(asset) await db_session.commit() response = await client.patch( f"/api/v1/assets/{asset.id}", json={"brand": "MacBook", "model": "Mac Pro"}, headers=admin_headers, ) assert response.status_code == 200, response.text data = response.json()["data"] assert data["brand"] == "MacBook" assert data["model"] == "Mac Pro" # Re-fetch from DB to confirm persistence (the original bug # returned 200 but never flushed). await db_session.refresh(asset) assert asset.brand == "MacBook" assert asset.model == "Mac Pro" assert asset.brand_id is not None assert asset.brand_id != old_brand.id assert asset.model_id is not None assert asset.model_id != old_model.id @pytest.mark.asyncio async def test_update_asset_with_decimal_purchase_cost( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Regression: editing an asset with a Decimal field (purchase_cost) must not 500 on audit-log JSON write. The audit detail column is JSON; Python's json encoder rejects Decimal, so the router has to serialize the payload via ``mode='json'`` before storing. """ asset = AssetItem( asset_tag="DECIMAL-AUDIT", category="laptop", brand="Apple", status="available", ) db_session.add(asset) await db_session.commit() response = await client.patch( f"/api/v1/assets/{asset.id}", json={"purchase_cost": "184800.00", "model": "Macbook Air"}, headers=admin_headers, ) assert response.status_code == 200, response.text data = response.json()["data"] assert data["model"] == "Macbook Air" @pytest.mark.asyncio async def test_assign_asset_to_user( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, db_session: AsyncSession, ): """Test assigning an asset to a user.""" employee_user, _ = employee_user # Extract user from tuple # Create available asset asset = AssetItem( asset_tag="ASSIGN-TEST", category="laptop", status="available", ) db_session.add(asset) await db_session.commit() # Assign to employee payload = { "user_id": str(employee_user.id), "notes": "Assigned for development work", } response = await client.post( f"/api/v1/assets/{asset.id}/assign", json=payload, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["status"] == "assigned" assert data["assigned_to"] == str(employee_user.id) assert data["assigned_to_email"] == employee_user.email # Verify in database await db_session.refresh(asset) assert asset.status == "assigned" assert asset.assigned_to == employee_user.id assert asset.assigned_at is not None # Verify history history_result = await db_session.execute( select(AssetHistory) .where(AssetHistory.asset_id == asset.id) .where(AssetHistory.action == "assigned") ) history = history_result.scalar_one() assert history.to_user == employee_user.id @pytest.mark.asyncio async def test_cannot_assign_already_assigned_asset( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, admin_user: tuple, db_session: AsyncSession, ): """Test that already assigned assets cannot be assigned again.""" employee_user, _ = employee_user # Extract user from tuple admin_user, _ = admin_user # Extract user from tuple # Create assigned asset (assigned to admin) asset = AssetItem( asset_tag="ALREADY-ASSIGNED", category="laptop", status="assigned", assigned_to=admin_user.id, # Assigned to admin user who exists ) db_session.add(asset) await db_session.commit() # Try to assign payload = {"user_id": str(employee_user.id)} response = await client.post( f"/api/v1/assets/{asset.id}/assign", json=payload, headers=admin_headers, ) assert response.status_code == 400 assert "cannot assign" in response.json()["errors"][0]["message"].lower() @pytest.mark.asyncio async def test_unassign_asset( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, db_session: AsyncSession, ): """Test unassigning (reclaiming) an asset.""" employee_user, _ = employee_user # Extract user from tuple # Create assigned asset asset = AssetItem( asset_tag="UNASSIGN-TEST", category="laptop", status="assigned", assigned_to=employee_user.id, assigned_at=date.today(), ) db_session.add(asset) await db_session.commit() # Unassign response = await client.post( f"/api/v1/assets/{asset.id}/unassign", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] # Canonical stock status is ``idle`` — the legacy ``available`` # synonym is accepted on read but no longer written (see the # comment in ``unassign_asset`` / migration 454). assert data["status"] == "idle" assert data["assigned_to"] is None # Verify in database await db_session.refresh(asset) assert asset.status == "idle" assert asset.assigned_to is None assert asset.assigned_at is None # Verify history history_result = await db_session.execute( select(AssetHistory) .where(AssetHistory.asset_id == asset.id) .where(AssetHistory.action == "unassigned") ) history = history_result.scalar_one() assert history.from_user == employee_user.id @pytest.mark.asyncio async def test_transfer_asset_between_users( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, admin_user: tuple, db_session: AsyncSession, ): """Test transferring asset from one user to another.""" employee_user, _ = employee_user # Extract user from tuple admin_user, _ = admin_user # Extract user from tuple # Create assigned asset asset = AssetItem( asset_tag="TRANSFER-TEST", category="laptop", status="assigned", assigned_to=employee_user.id, ) db_session.add(asset) await db_session.commit() # Transfer to admin payload = { "to_user_id": str(admin_user.id), "notes": "Transfer for admin use", } response = await client.post( f"/api/v1/assets/{asset.id}/transfer", json=payload, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["status"] == "assigned" assert data["assigned_to"] == str(admin_user.id) # Verify history history_result = await db_session.execute( select(AssetHistory) .where(AssetHistory.asset_id == asset.id) .where(AssetHistory.action == "transferred") ) history = history_result.scalar_one() assert history.from_user == employee_user.id assert history.to_user == admin_user.id @pytest.mark.asyncio async def test_hard_delete_asset( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """DELETE soft-deletes an asset that nobody is using. PR #164 (2026-05-04) replaced hard-DELETE with soft-delete via ``deleted_at`` so the audit chain on ``asset_history`` / ``asset_operations`` is preserved (CLAUDE.md rule 6 — audit by design). The HTTP contract is unchanged (200 + ``deleted: True``); the DB row remains with a non-NULL ``deleted_at`` and is filtered out of every active read. """ # Create available asset asset = AssetItem( asset_tag="DELETE-TEST", category="laptop", status="available", ) db_session.add(asset) await db_session.commit() asset_id = asset.id response = await client.delete( f"/api/v1/assets/{asset_id}", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["deleted"] is True assert data["id"] == str(asset_id) # Row remains in the database but has ``deleted_at`` set and # is invisible to active reads (``get_asset`` filters it out). from sqlalchemy import select result = await db_session.execute( select(AssetItem).where(AssetItem.id == asset_id) ) row = result.scalar_one_or_none() assert row is not None, "soft-deleted asset row should still exist" assert row.deleted_at is not None, "deleted_at should be set" @pytest.mark.asyncio async def test_cannot_delete_asset_in_use( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, db_session: AsyncSession, ): """DELETE refuses when the asset still has a holder.""" employee_user, _ = employee_user # Extract user from tuple # Create assigned asset asset = AssetItem( asset_tag="ASSIGNED-DELETE", category="laptop", status="assigned", assigned_to=employee_user.id, ) db_session.add(asset) await db_session.commit() response = await client.delete( f"/api/v1/assets/{asset.id}", headers=admin_headers, ) assert response.status_code == 400 errors = response.json()["errors"] assert errors[0]["code"] == "ASSET_IN_USE" # Asset must still exist. await db_session.refresh(asset) assert asset.assigned_to == employee_user.id @pytest.mark.asyncio async def test_employee_can_view_own_assets( self, client: AsyncClient, employee_headers: dict, employee_user: tuple, admin_user: tuple, db_session: AsyncSession, ): """Test that employees can view their own assigned assets.""" employee_user, _ = employee_user # Extract user from tuple admin_user, _ = admin_user # Extract user from tuple # Create assets my_asset = AssetItem( asset_tag="MY-ASSET", category="laptop", status="assigned", assigned_to=employee_user.id, ) other_asset = AssetItem( asset_tag="OTHER-ASSET", category="laptop", status="assigned", assigned_to=admin_user.id, # Assigned to admin who exists ) db_session.add_all([my_asset, other_asset]) await db_session.commit() # Get my assets response = await client.get( "/api/v1/assets/my", headers=employee_headers, ) assert response.status_code == 200 data = response.json()["data"] assert len(data) == 1 assert data[0]["asset_tag"] == "MY-ASSET" @pytest.mark.asyncio async def test_manager_can_view_subordinate_assets( self, client: AsyncClient, db_session: AsyncSession, ): """Test that managers can view their subordinates' assets.""" # Create manager and subordinate manager = AuthUser( work_email="manager@test.com", display_name="Manager User", employment_status="active", source_system="successfactors", hris_person_id="sf-mgr-asset-001", ) db_session.add(manager) await db_session.flush() subordinate = AuthUser( work_email="subordinate@test.com", display_name="Subordinate User", manager_person_id=manager.id, employment_status="active", source_system="successfactors", hris_person_id="sf-sub-asset-001", ) db_session.add(subordinate) await db_session.flush() await db_session.commit() # Create asset for subordinate asset = AssetItem( asset_tag="SUB-ASSET", category="laptop", status="assigned", assigned_to=subordinate.id, ) db_session.add(asset) await db_session.commit() # Manager login manager_token = "mock_manager_token" # In real test, generate proper token manager_headers = {"Authorization": f"Bearer {manager_token}"} # Get subordinate's assets await client.get( f"/api/v1/assets?assigned_to={subordinate.id}", headers=manager_headers, ) # This would work if we had proper token generation # assert response.status_code == 200 @pytest.mark.asyncio async def test_asset_history_tracking( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, db_session: AsyncSession, ): """Test that all asset actions are tracked in history.""" employee_user, _ = employee_user # Extract user from tuple # Create asset create_response = await client.post( "/api/v1/assets", json={ "asset_tag": "HISTORY-TEST", "category": "laptop", }, headers=admin_headers, ) asset_id = create_response.json()["data"]["id"] # Update asset await client.patch( f"/api/v1/assets/{asset_id}", json={"brand": "HP"}, headers=admin_headers, ) # Assign asset await client.post( f"/api/v1/assets/{asset_id}/assign", json={"user_id": str(employee_user.id)}, headers=admin_headers, ) # Get asset history response = await client.get( f"/api/v1/assets/history?asset_id={asset_id}", headers=admin_headers, ) assert response.status_code == 200 history = response.json()["data"] assert len(history) >= 3 # Check history actions actions = [h["action"] for h in history] assert "created" in actions assert "updated" in actions assert "assigned" in actions @pytest.mark.asyncio async def test_asset_stats_endpoint( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, db_session: AsyncSession, ): """Test asset statistics endpoint.""" employee_user, _ = employee_user # Extract user from tuple # Create test assets today = date.today() assets = [ AssetItem( asset_tag="STAT-1", category="laptop", status="available", purchase_cost=Decimal("1000.00"), warranty_until=today + timedelta(days=15), ), AssetItem( asset_tag="STAT-2", category="monitor", status="assigned", assigned_to=employee_user.id, purchase_cost=Decimal("500.00"), ), AssetItem( asset_tag="STAT-3", category="laptop", status="idle", purchase_cost=Decimal("1500.00"), ), ] db_session.add_all(assets) await db_session.commit() # Get stats response = await client.get( "/api/v1/assets/stats", headers=admin_headers, ) assert response.status_code == 200 stats = response.json()["data"] assert stats["total_assets"] >= 3 assert stats["by_category"]["laptop"] >= 2 assert stats["by_category"]["monitor"] >= 1 assert stats["by_status"]["available"] >= 1 assert stats["by_status"]["assigned"] >= 1 assert float(stats["total_value"]) >= 3000.00 assert stats["warranty_expiring_soon"] >= 1 @pytest.mark.asyncio async def test_asset_stats_excludes_soft_deleted( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, db_session: AsyncSession, ): """Regression (#1664): soft-deleted assets must not inflate total_value or the assigned/available/warranty/recent counts — those omitted the deleted_at filter that every sibling stat already applied.""" employee_user, _ = employee_user today = date.today() # One live asset + one soft-deleted asset in each counted bucket. db_session.add_all([ AssetItem(asset_tag="SD-LIVE", category="laptop", status="available", purchase_cost=Decimal("1000.00"), warranty_until=today + timedelta(days=10)), AssetItem(asset_tag="SD-DEL-VALUE", category="laptop", status="available", purchase_cost=Decimal("9999.00"), warranty_until=today + timedelta(days=10), deleted_at=datetime.now(UTC)), AssetItem(asset_tag="SD-DEL-ASSIGNED", category="monitor", status="assigned", assigned_to=employee_user.id, purchase_cost=Decimal("50.00"), deleted_at=datetime.now(UTC)), ]) await db_session.commit() response = await client.get("/api/v1/assets/stats", headers=admin_headers) assert response.status_code == 200 stats = response.json()["data"] # The 9999 soft-deleted asset must not be in total_value. assert Decimal(str(stats["total_value"])) < Decimal("9999.00") + Decimal("5000.00") # A crisp per-tag check: recompute from live-only expectation. from sqlalchemy import func, select live_value = (await db_session.execute( select(func.sum(AssetItem.purchase_cost)).where(AssetItem.deleted_at.is_(None)) )).scalar() assert Decimal(str(stats["total_value"])) == live_value @pytest.mark.asyncio async def test_duplicate_asset_tag_rejected( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test that duplicate asset tags are rejected.""" # Create first asset asset = AssetItem( asset_tag="DUPLICATE-TAG", category="laptop", status="available", ) db_session.add(asset) await db_session.commit() # Try to create with same tag response = await client.post( "/api/v1/assets", json={ "asset_tag": "DUPLICATE-TAG", "category": "monitor", }, headers=admin_headers, ) assert response.status_code == 400 assert "already exists" in response.json()["errors"][0]["message"] @pytest.mark.asyncio async def test_duplicate_serial_number_rejected( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test that duplicate serial numbers are rejected.""" # Create first asset asset = AssetItem( asset_tag="SERIAL-1", category="laptop", serial_number="DUPLICATE-SERIAL", status="available", ) db_session.add(asset) await db_session.commit() # Try to create with same serial response = await client.post( "/api/v1/assets", json={ "asset_tag": "SERIAL-2", "category": "laptop", "serial_number": "DUPLICATE-SERIAL", }, headers=admin_headers, ) assert response.status_code == 400 assert "Serial number" in response.json()["errors"][0]["message"] @pytest.mark.asyncio async def test_invalid_category_rejected( self, client: AsyncClient, admin_headers: dict, ): """Test that invalid categories are rejected.""" response = await client.post( "/api/v1/assets", json={ "asset_tag": "INVALID-CAT", "category": "invalid_category", }, headers=admin_headers, ) assert response.status_code == 422 assert "Category must be one of" in str(response.json()) @pytest.mark.asyncio async def test_invalid_status_rejected( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """PATCH /assets/{id} no longer accepts ``status`` at all. PR #164 (2026-05-04) removed ``status`` from ``AssetUpdateRequest`` so the admin update path can't bypass the lifecycle state machine (claim / return / borrow / dispose are the only ways to change status). Pydantic's default ``extra='ignore'`` silently drops the field. Verify the original status survives the no-op patch. """ # Create asset asset = AssetItem( asset_tag="STATUS-TEST", category="laptop", status="available", ) db_session.add(asset) await db_session.commit() original_status = asset.status # Status is silently dropped by AssetUpdateRequest now. response = await client.patch( f"/api/v1/assets/{asset.id}", json={"status": "invalid_status"}, headers=admin_headers, ) # 200 because the field is dropped, not a validation error. assert response.status_code == 200 # Verify status was NOT changed. await db_session.refresh(asset) assert asset.status == original_status class TestAssetDuplicateUniqueColumn: """Unit tests for the 1062 duplicate-key classifier. Faithful regression guard for the prod 500 on asset intake (2026-05-27): prod's UNIQUE indexes are named ``uk_asset_items_serial_number`` / ``uk_asset_items_asset_tag`` (built by alembic migrations), while the test DB's ``create_all`` names them after the bare column. The old handler matched only ``asset_items.serial_number`` (the test name), so a real soft-deleted-serial collision in prod fell through to a raw 500. These cases pin BOTH index-name shapes so the test/prod divergence can't hide the bug again. """ @staticmethod def _err(key: str, value: str = "X") -> IntegrityError: orig = Exception() orig.args = (1062, f"Duplicate entry '{value}' for key '{key}'") return IntegrityError("INSERT INTO asset_items ...", {}, orig) def test_prod_serial_index_name(self): exc = self._err( "asset_items.uk_asset_items_serial_number", "K27CY61C37" ) assert _asset_duplicate_unique_column(exc) == "serial_number" def test_prod_asset_tag_index_name(self): exc = self._err( "asset_items.uk_asset_items_asset_tag", "IT-JP-DT-20260527042922" ) assert _asset_duplicate_unique_column(exc) == "asset_tag" def test_create_all_column_index_name(self): assert ( _asset_duplicate_unique_column(self._err("asset_items.serial_number")) == "serial_number" ) assert ( _asset_duplicate_unique_column(self._err("asset_items.asset_tag")) == "asset_tag" ) def test_non_1062_returns_none(self): orig = Exception() orig.args = (1452, "foreign key constraint fails on asset_tag column") exc = IntegrityError("...", {}, orig) assert _asset_duplicate_unique_column(exc) is None def test_value_containing_column_name_not_misclassified(self): # The colliding *value* mentions another column, but only the key # name (after ``for key``) should drive classification. exc = self._err( "asset_items.uk_asset_items_serial_number", "asset_tag-lookalike" ) assert _asset_duplicate_unique_column(exc) == "serial_number" class TestSoftDeletedCollisionIntake: """End-to-end: intake colliding with a soft-deleted ghost returns a friendly 409, never a raw 500 (the prod symptom).""" @pytest.mark.asyncio async def test_intake_serial_of_soft_deleted_asset_returns_409( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): ghost = AssetItem( asset_tag="GHOST-TAG-SN", category="laptop", serial_number="GHOST-SN-1", status="idle", deleted_at=datetime.now(UTC), ) db_session.add(ghost) await db_session.commit() response = await client.post( "/api/v1/assets/intake", json={ "asset_tag": "NEW-TAG-SN", "category": "laptop", "serial_number": "GHOST-SN-1", }, headers=admin_headers, ) assert response.status_code == 409 assert ( response.json()["errors"][0]["code"] == "SERIAL_NUMBER_EXISTS_SOFT_DELETED" ) @pytest.mark.asyncio async def test_intake_tag_of_soft_deleted_asset_returns_409( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): ghost = AssetItem( asset_tag="GHOST-TAG-1", category="laptop", status="idle", deleted_at=datetime.now(UTC), ) db_session.add(ghost) await db_session.commit() response = await client.post( "/api/v1/assets/intake", json={"asset_tag": "GHOST-TAG-1", "category": "laptop"}, headers=admin_headers, ) assert response.status_code == 409 assert ( response.json()["errors"][0]["code"] == "ASSET_TAG_EXISTS_SOFT_DELETED" ) class TestAssetRecycleBin: """Recycle bin: admins list soft-deleted assets and restore them.""" async def _make_deleted_asset( self, db_session: AsyncSession, tag: str, serial: str, ) -> AssetItem: asset = AssetItem( asset_tag=tag, category="laptop", serial_number=serial, status="idle", deleted_at=datetime.now(UTC), ) db_session.add(asset) await db_session.commit() return asset @pytest.mark.asyncio async def test_admin_lists_recycle_bin( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): await self._make_deleted_asset(db_session, "RB-DEL-A", "RB-SN-A") active = AssetItem(asset_tag="RB-ACTIVE", category="laptop", status="idle") db_session.add(active) await db_session.commit() resp = await client.get( "/api/v1/assets/recycle-bin", headers=admin_headers, ) assert resp.status_code == 200 rows = resp.json()["data"] tags = [a["asset_tag"] for a in rows] assert "RB-DEL-A" in tags # soft-deleted shows up assert "RB-ACTIVE" not in tags # active never shows up deleted_row = next(a for a in rows if a["asset_tag"] == "RB-DEL-A") assert deleted_row["deleted_at"] is not None @pytest.mark.asyncio async def test_admin_restore_brings_asset_back( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): asset = await self._make_deleted_asset(db_session, "RB-DEL-B", "RB-SN-B") resp = await client.post( f"/api/v1/assets/{asset.id}/restore", headers=admin_headers, ) assert resp.status_code == 200 assert resp.json()["data"]["restored"] is True # Back in the normal list, gone from the recycle bin. listed = await client.get("/api/v1/assets", headers=admin_headers) assert "RB-DEL-B" in [a["asset_tag"] for a in listed.json()["data"]] bin_resp = await client.get( "/api/v1/assets/recycle-bin", headers=admin_headers, ) assert "RB-DEL-B" not in [a["asset_tag"] for a in bin_resp.json()["data"]] # A 'restored' history row was written. hist = (await db_session.execute( select(AssetHistory).where( AssetHistory.asset_id == asset.id, AssetHistory.action == "restored", ) )).scalars().all() assert len(hist) == 1 @pytest.mark.asyncio async def test_restore_non_deleted_returns_409( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): active = AssetItem(asset_tag="RB-NOTDEL", category="laptop", status="idle") db_session.add(active) await db_session.commit() resp = await client.post( f"/api/v1/assets/{active.id}/restore", headers=admin_headers, ) assert resp.status_code == 409 assert resp.json()["errors"][0]["code"] == "ASSET_NOT_DELETED" @pytest.mark.asyncio async def test_employee_cannot_access_recycle_bin( self, client: AsyncClient, employee_headers: dict, db_session: AsyncSession, ): asset = await self._make_deleted_asset(db_session, "RB-DEL-C", "RB-SN-C") list_resp = await client.get( "/api/v1/assets/recycle-bin", headers=employee_headers, ) assert list_resp.status_code == 403 restore_resp = await client.post( f"/api/v1/assets/{asset.id}/restore", headers=employee_headers, ) assert restore_resp.status_code == 403 class TestAssetLifecycleOperations: """Test suite for asset lifecycle operations (intake, claim, return, etc.).""" @pytest.mark.asyncio async def test_intake_asset( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test registering a new asset via intake.""" payload = { "asset_tag": "INTAKE-001", "category": "laptop", "brand": "Lenovo", "model": "ThinkPad X1", "serial_number": "INTAKE-SN-001", "source_type": "purchased", "supplier_name": "Lenovo Direct", "notes": "New intake test", } response = await client.post( "/api/v1/assets/intake", json=payload, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert "asset" in data assert "operation" in data assert data["asset"]["asset_tag"] == "INTAKE-001" assert data["asset"]["status"] == "idle" assert data["operation"]["operation_type"] == "intake" assert data["operation"]["status"] == "confirmed" # Verify operation in DB result = await db_session.execute( select(AssetOperation).where( AssetOperation.operation_type == "intake" ) ) op = result.scalar_one() assert op.operation_number.startswith("OP-") @pytest.mark.asyncio async def test_intake_duplicate_tag_rejected( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test that intake with duplicate asset tag is rejected.""" asset = AssetItem( asset_tag="INTAKE-DUP", category="laptop", status="idle", ) db_session.add(asset) await db_session.commit() response = await client.post( "/api/v1/assets/intake", json={"asset_tag": "INTAKE-DUP", "category": "laptop"}, headers=admin_headers, ) assert response.status_code == 400 assert "already exists" in response.json()["errors"][0]["message"] @pytest.mark.asyncio async def test_claim_asset( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, db_session: AsyncSession, ): """Test claiming (dispatching) an asset to a user.""" employee_user_obj, _ = employee_user asset = AssetItem( asset_tag="CLAIM-001", category="laptop", status="idle", ) db_session.add(asset) await db_session.commit() response = await client.post( f"/api/v1/assets/{asset.id}/claim", json={"to_user_id": str(employee_user_obj.id), "notes": "Claim test"}, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["asset"]["status"] == "claimed" assert data["asset"]["assigned_to"] == str(employee_user_obj.id) assert data["operation"]["operation_type"] == "claim" assert data["operation"]["status"] == "confirmed" @pytest.mark.asyncio async def test_claim_non_idle_asset_rejected( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, db_session: AsyncSession, ): """Test that claiming a non-idle asset is rejected.""" employee_user_obj, _ = employee_user asset = AssetItem( asset_tag="CLAIM-BUSY", category="laptop", status="claimed", assigned_to=employee_user_obj.id, ) db_session.add(asset) await db_session.commit() response = await client.post( f"/api/v1/assets/{asset.id}/claim", json={"to_user_id": str(employee_user_obj.id)}, headers=admin_headers, ) assert response.status_code == 400 assert "cannot claim" in response.json()["errors"][0]["message"].lower() @pytest.mark.asyncio async def test_return_claimed_asset( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, db_session: AsyncSession, ): """Test returning a claimed asset to stock.""" employee_user_obj, _ = employee_user asset = AssetItem( asset_tag="RETURN-001", category="laptop", status="claimed", assigned_to=employee_user_obj.id, ) db_session.add(asset) await db_session.commit() response = await client.post( f"/api/v1/assets/{asset.id}/return", json={"notes": "Return test"}, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["asset"]["status"] == "idle" assert data["asset"]["assigned_to"] is None assert data["operation"]["operation_type"] == "return" @pytest.mark.asyncio async def test_return_idle_asset_rejected( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test that returning an idle asset is rejected.""" asset = AssetItem( asset_tag="RETURN-IDLE", category="laptop", status="idle", ) db_session.add(asset) await db_session.commit() response = await client.post( f"/api/v1/assets/{asset.id}/return", json={}, headers=admin_headers, ) assert response.status_code == 400 assert "cannot return" in response.json()["errors"][0]["message"].lower() @pytest.mark.asyncio async def test_borrow_asset( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, db_session: AsyncSession, ): """Test temporarily lending an asset. PR #164 (2026-05-04): * ``expected_return_date`` is REQUIRED on the borrow request (must be a future date) — the new ``asset.borrow_overdue`` cron uses this signal. * Borrow now writes the borrower to ``borrowed_by`` and PRESERVES ``assigned_to`` as the original owner. For borrow-from-idle, ``assigned_to`` stays NULL. """ from datetime import date, timedelta employee_user_obj, _ = employee_user asset = AssetItem( asset_tag="BORROW-001", category="laptop", status="idle", ) db_session.add(asset) await db_session.commit() future = (date.today() + timedelta(days=7)).isoformat() response = await client.post( f"/api/v1/assets/{asset.id}/borrow", json={ "to_user_id": str(employee_user_obj.id), "notes": "Borrow test", "expected_return_date": future, }, headers=admin_headers, ) assert response.status_code == 200, response.text data = response.json()["data"] assert data["asset"]["status"] == "borrowed" # New model: borrower lives in borrowed_by, NOT assigned_to. assert data["asset"]["borrowed_by"] == str(employee_user_obj.id) # Borrowed-from-idle: no original owner. assert data["asset"]["assigned_to"] is None assert data["operation"]["operation_type"] == "borrow" @pytest.mark.asyncio async def test_borrow_return_asset( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, db_session: AsyncSession, ): """Test returning a borrowed asset.""" employee_user_obj, _ = employee_user asset = AssetItem( asset_tag="BORROW-RET-001", category="laptop", status="borrowed", assigned_to=employee_user_obj.id, ) db_session.add(asset) await db_session.commit() response = await client.post( f"/api/v1/assets/{asset.id}/borrow-return", json={"notes": "Borrow return test"}, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["asset"]["status"] == "idle" assert data["asset"]["assigned_to"] is None assert data["operation"]["operation_type"] == "borrow_return" @pytest.mark.asyncio async def test_borrow_return_non_borrowed_rejected( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test that returning a non-borrowed asset is rejected.""" asset = AssetItem( asset_tag="BORROW-RET-BAD", category="laptop", status="idle", ) db_session.add(asset) await db_session.commit() response = await client.post( f"/api/v1/assets/{asset.id}/borrow-return", json={}, headers=admin_headers, ) assert response.status_code == 400 assert "not borrowed" in response.json()["errors"][0]["message"].lower() @pytest.mark.asyncio async def test_change_assignee( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, admin_user: tuple, db_session: AsyncSession, ): """Test changing asset assignee (transfer).""" employee_user_obj, _ = employee_user admin_user_obj, _ = admin_user asset = AssetItem( asset_tag="CHANGE-001", category="laptop", status="claimed", assigned_to=employee_user_obj.id, ) db_session.add(asset) await db_session.commit() response = await client.post( f"/api/v1/assets/{asset.id}/change-assignee", json={"to_user_id": str(admin_user_obj.id), "notes": "Transfer test"}, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["asset"]["assigned_to"] == str(admin_user_obj.id) assert data["operation"]["operation_type"] == "transfer" @pytest.mark.asyncio async def test_change_assignee_same_user_rejected( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, db_session: AsyncSession, ): """Test that transferring to the same user is rejected.""" employee_user_obj, _ = employee_user asset = AssetItem( asset_tag="CHANGE-SAME", category="laptop", status="claimed", assigned_to=employee_user_obj.id, ) db_session.add(asset) await db_session.commit() response = await client.post( f"/api/v1/assets/{asset.id}/change-assignee", json={"to_user_id": str(employee_user_obj.id)}, headers=admin_headers, ) assert response.status_code == 400 assert "already assigned" in response.json()["errors"][0]["message"].lower() @pytest.mark.asyncio async def test_repair_asset( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test sending an asset to repair.""" asset = AssetItem( asset_tag="REPAIR-001", category="laptop", status="idle", device_status="normal", ) db_session.add(asset) await db_session.commit() response = await client.post( f"/api/v1/assets/{asset.id}/repair", json={"notes": "Screen cracked", "detail": {"issue": "cracked_screen"}}, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["asset"]["device_status"] == "under_repair" assert data["operation"]["operation_type"] == "repair" @pytest.mark.asyncio async def test_repair_complete( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test marking repair as done.""" asset = AssetItem( asset_tag="REPAIR-DONE-001", category="laptop", status="idle", device_status="under_repair", ) db_session.add(asset) await db_session.commit() response = await client.post( f"/api/v1/assets/{asset.id}/repair-complete", json={"notes": "Screen replaced", "detail": {"cost": 200}}, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["asset"]["device_status"] == "normal" assert data["operation"]["operation_type"] == "repair_complete" @pytest.mark.asyncio async def test_repair_complete_not_under_repair_rejected( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test that completing repair on a non-under-repair asset is rejected.""" asset = AssetItem( asset_tag="REPAIR-BAD", category="laptop", status="idle", device_status="normal", ) db_session.add(asset) await db_session.commit() response = await client.post( f"/api/v1/assets/{asset.id}/repair-complete", json={}, headers=admin_headers, ) assert response.status_code == 400 assert "not under_repair" in response.json()["errors"][0]["message"].lower() @pytest.mark.asyncio async def test_dispose_asset( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test disposing/retiring an asset.""" asset = AssetItem( asset_tag="DISPOSE-001", category="laptop", status="idle", ) db_session.add(asset) await db_session.commit() response = await client.post( f"/api/v1/assets/{asset.id}/dispose", json={"disposal_type": "scrapped", "notes": "End of life"}, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["asset"]["status"] == "disposed" assert data["asset"]["disposal_type"] == "scrapped" assert data["asset"]["disposed_at"] is not None assert data["operation"]["operation_type"] == "disposal" @pytest.mark.asyncio async def test_dispose_already_disposed_rejected( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test that disposing an already disposed asset is rejected.""" asset = AssetItem( asset_tag="DISPOSE-DUP", category="laptop", status="disposed", ) db_session.add(asset) await db_session.commit() response = await client.post( f"/api/v1/assets/{asset.id}/dispose", json={"disposal_type": "sold"}, headers=admin_headers, ) assert response.status_code == 400 assert "already" in response.json()["errors"][0]["message"].lower() @pytest.mark.asyncio async def test_dispose_claimed_asset_requires_return_for_physical_disposal( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, db_session: AsyncSession, ): """Physical disposal still requires a held asset to be returned.""" employee_user_obj, _ = employee_user asset = AssetItem( asset_tag="DISPOSE-ASSIGNED", category="laptop", status="claimed", assigned_to=employee_user_obj.id, ) db_session.add(asset) await db_session.commit() response = await client.post( f"/api/v1/assets/{asset.id}/dispose", json={"disposal_type": "scrapped"}, headers=admin_headers, ) assert response.status_code == 400 assert "return it first" in response.json()["errors"][0]["message"].lower() @pytest.mark.asyncio @pytest.mark.parametrize("disposal_type", ["sold", "lost"]) async def test_dispose_claimed_asset_directly_for_sale_or_loss( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, db_session: AsyncSession, disposal_type: str, ): """Buyback and loss are terminal even when the employee still holds it.""" employee_user_obj, _ = employee_user asset = AssetItem( asset_tag=f"DISPOSE-DIRECT-{disposal_type.upper()}", category="laptop", status="claimed", assigned_to=employee_user_obj.id, ) db_session.add(asset) await db_session.commit() response = await client.post( f"/api/v1/assets/{asset.id}/dispose", json={"disposal_type": disposal_type}, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["asset"]["status"] == "disposed" assert data["asset"]["assigned_to"] is None assert data["operation"]["from_user_id"] == str(employee_user_obj.id) @pytest.mark.asyncio async def test_asset_admin_cannot_dispose_claimed_asset_directly( self, client: AsyncClient, asset_admin_headers: dict, employee_user: tuple, db_session: AsyncSession, ): """Only global_admin may bypass return-first for sale or loss.""" employee_user_obj, _ = employee_user asset = AssetItem( asset_tag="DISPOSE-ASSET-ADMIN-DIRECT", category="laptop", status="claimed", assigned_to=employee_user_obj.id, location="TestCity", ) db_session.add(asset) await db_session.commit() response = await client.post( f"/api/v1/assets/{asset.id}/dispose", json={"disposal_type": "lost"}, headers=asset_admin_headers, ) assert response.status_code == 400 assert "return it first" in response.json()["errors"][0]["message"].lower() @pytest.mark.asyncio async def test_employee_cannot_intake( self, client: AsyncClient, employee_headers: dict, ): """Test that regular employee cannot intake assets.""" response = await client.post( "/api/v1/assets/intake", json={"asset_tag": "NOPE", "category": "laptop"}, headers=employee_headers, ) assert response.status_code == 403 @pytest.mark.asyncio async def test_list_operations( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test listing operations with filters.""" # Intake an asset to create an operation response = await client.post( "/api/v1/assets/intake", json={"asset_tag": "OP-LIST-001", "category": "monitor"}, headers=admin_headers, ) assert response.status_code == 200 # List operations response = await client.get( "/api/v1/assets/operations?operation_type=intake", headers=admin_headers, ) assert response.status_code == 200 data = response.json() assert data["meta"]["total"] >= 1 assert all( op["operation_type"] == "intake" for op in data["data"] ) @pytest.mark.asyncio async def test_confirm_operation( self, client: AsyncClient, admin_headers: dict, admin_user: tuple, employee_user: tuple, db_session: AsyncSession, ): """Test confirming a pending operation.""" admin_user_obj, _ = admin_user employee_user_obj, _ = employee_user # Create asset and a pending operation asset = AssetItem( asset_tag="CONFIRM-OP-001", category="laptop", status="idle", ) db_session.add(asset) await db_session.flush() operation = AssetOperation( operation_number="OP-TEST-CONFIRM-001", asset_id=asset.id, operation_type="claim", status="pending", to_user_id=employee_user_obj.id, operator_id=admin_user_obj.id, notes="Pending claim", ) db_session.add(operation) await db_session.commit() # Confirm it response = await client.post( f"/api/v1/assets/operations/{operation.id}/confirm", json={"action": "confirm", "notes": "Approved"}, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["status"] == "confirmed" assert data["confirmed_at"] is not None @pytest.mark.asyncio async def test_reject_operation( self, client: AsyncClient, admin_headers: dict, admin_user: tuple, db_session: AsyncSession, ): """Test rejecting a pending operation.""" admin_user_obj, _ = admin_user asset = AssetItem( asset_tag="REJECT-OP-001", category="laptop", status="idle", ) db_session.add(asset) await db_session.flush() operation = AssetOperation( operation_number="OP-TEST-REJECT-001", asset_id=asset.id, operation_type="claim", status="pending", operator_id=admin_user_obj.id, notes="Pending claim to reject", ) db_session.add(operation) await db_session.commit() response = await client.post( f"/api/v1/assets/operations/{operation.id}/confirm", json={"action": "reject", "notes": "Not needed"}, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["status"] == "rejected" @pytest.mark.asyncio async def test_confirm_non_pending_rejected( self, client: AsyncClient, admin_headers: dict, admin_user: tuple, db_session: AsyncSession, ): """Test that confirming a non-pending operation is rejected.""" admin_user_obj, _ = admin_user asset = AssetItem( asset_tag="CONFIRM-NP-001", category="laptop", status="idle", ) db_session.add(asset) await db_session.flush() operation = AssetOperation( operation_number="OP-TEST-NP-001", asset_id=asset.id, operation_type="intake", status="confirmed", operator_id=admin_user_obj.id, ) db_session.add(operation) await db_session.commit() response = await client.post( f"/api/v1/assets/operations/{operation.id}/confirm", json={"action": "confirm"}, headers=admin_headers, ) assert response.status_code == 400 assert "not pending" in response.json()["errors"][0]["message"].lower() @pytest.mark.asyncio async def test_confirm_operation_enforces_asset_region( self, client: AsyncClient, asset_admin_headers: dict, asset_admin_user: tuple, employee_user: tuple, db_session: AsyncSession, ): """Regression: a region-scoped asset_admin must NOT be able to confirm/reject a pending operation for an asset in an office they don't manage. Every other operation/asset mutation runs ``_enforce_asset_region``; ``/operations/{id}/confirm`` was the sole one that omitted it, letting a TestCity admin confirm an out-of-region asset's pending operation. """ admin_user_obj, _ = asset_admin_user employee_user_obj, _ = employee_user # Asset held by an office the TestCity-scoped admin does NOT manage. other_office = AssetOfficeAddress( id=uuid4(), city="OtherCity", office_name="Other Office", ) db_session.add(other_office) await db_session.flush() asset = AssetItem( asset_tag=f"REGION-OP-{uuid4().hex[:6]}", category="laptop", status="held_by_office", holder_kind=ASSET_HOLDER_KIND_OFFICE, holder_office_id=other_office.id, ) db_session.add(asset) await db_session.flush() operation = AssetOperation( operation_number=f"OP-REGION-{uuid4().hex[:6]}", asset_id=asset.id, operation_type="claim", status="pending", to_user_id=employee_user_obj.id, operator_id=admin_user_obj.id, ) db_session.add(operation) await db_session.commit() response = await client.post( f"/api/v1/assets/operations/{operation.id}/confirm", json={"action": "confirm", "notes": "should be blocked"}, headers=asset_admin_headers, ) assert response.status_code == 403, response.text # And the operation stays pending — the mutation never ran. await db_session.refresh(operation) assert operation.status == "pending" @pytest.mark.asyncio async def test_full_lifecycle_flow( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, admin_user: tuple, db_session: AsyncSession, ): """Test a full asset lifecycle: intake -> claim -> return -> borrow -> borrow_return -> dispose.""" employee_user_obj, _ = employee_user admin_user_obj, _ = admin_user # 1. Intake resp = await client.post( "/api/v1/assets/intake", json={ "asset_tag": "LIFECYCLE-001", "category": "laptop", "brand": "Dell", "model": "Latitude 5540", }, headers=admin_headers, ) assert resp.status_code == 200 asset_id = resp.json()["data"]["asset"]["id"] # 2. Claim (dispatch to employee) resp = await client.post( f"/api/v1/assets/{asset_id}/claim", json={"to_user_id": str(employee_user_obj.id)}, headers=admin_headers, ) assert resp.status_code == 200 assert resp.json()["data"]["asset"]["status"] == "claimed" # 3. Return to stock resp = await client.post( f"/api/v1/assets/{asset_id}/return", json={"notes": "Employee left"}, headers=admin_headers, ) assert resp.status_code == 200 assert resp.json()["data"]["asset"]["status"] == "idle" # 4. Borrow — PR #164 requires expected_return_date. from datetime import date, timedelta future = (date.today() + timedelta(days=7)).isoformat() resp = await client.post( f"/api/v1/assets/{asset_id}/borrow", json={ "to_user_id": str(admin_user_obj.id), "expected_return_date": future, }, headers=admin_headers, ) assert resp.status_code == 200, resp.text assert resp.json()["data"]["asset"]["status"] == "borrowed" # 5. Borrow return resp = await client.post( f"/api/v1/assets/{asset_id}/borrow-return", json={}, headers=admin_headers, ) assert resp.status_code == 200 assert resp.json()["data"]["asset"]["status"] == "idle" # 6. Dispose resp = await client.post( f"/api/v1/assets/{asset_id}/dispose", json={"disposal_type": "scrapped"}, headers=admin_headers, ) assert resp.status_code == 200 assert resp.json()["data"]["asset"]["status"] == "disposed" # Verify operations were created for all steps resp = await client.get( f"/api/v1/assets/operations?asset_id={asset_id}", headers=admin_headers, ) assert resp.status_code == 200 ops = resp.json()["data"] op_types = [o["operation_type"] for o in ops] assert "intake" in op_types assert "claim" in op_types assert "return" in op_types assert "borrow" in op_types assert "borrow_return" in op_types assert "disposal" in op_types class TestAssetDetailOperations: """Test that asset detail API returns operations with enriched user info.""" @pytest.mark.asyncio async def test_detail_includes_operations( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, db_session: AsyncSession, ): """Test that GET /assets/{id} returns operations list.""" employee_user_obj, _ = employee_user # Intake creates an operation resp = await client.post( "/api/v1/assets/intake", json={"asset_tag": "DETAIL-OP-001", "category": "laptop"}, headers=admin_headers, ) assert resp.status_code == 200 asset_id = resp.json()["data"]["asset"]["id"] # Claim creates another operation await client.post( f"/api/v1/assets/{asset_id}/claim", json={"to_user_id": str(employee_user_obj.id), "notes": "Claim for test"}, headers=admin_headers, ) # Fetch detail resp = await client.get( f"/api/v1/assets/{asset_id}", headers=admin_headers, ) assert resp.status_code == 200 data = resp.json()["data"] # Must contain operations key assert "operations" in data assert len(data["operations"]) >= 2 # Operations should have required fields op = data["operations"][0] assert "operation_number" in op assert "operation_type" in op assert "status" in op assert "operator_id" in op assert "created_at" in op @pytest.mark.asyncio async def test_detail_operations_have_user_names( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, db_session: AsyncSession, ): """Test that operations in detail include enriched operator/user names.""" employee_user_obj, _ = employee_user # Intake + claim to get operations with from/to users resp = await client.post( "/api/v1/assets/intake", json={"asset_tag": "DETAIL-OP-002", "category": "monitor"}, headers=admin_headers, ) asset_id = resp.json()["data"]["asset"]["id"] await client.post( f"/api/v1/assets/{asset_id}/claim", json={"to_user_id": str(employee_user_obj.id)}, headers=admin_headers, ) resp = await client.get( f"/api/v1/assets/{asset_id}", headers=admin_headers, ) data = resp.json()["data"] # Find the claim operation (most recent first) claim_ops = [o for o in data["operations"] if o["operation_type"] == "claim"] assert len(claim_ops) >= 1 claim_op = claim_ops[0] # operator_name should be enriched assert "operator_name" in claim_op assert claim_op["operator_name"] is not None # to_user_name should be enriched for claim assert "to_user_name" in claim_op assert claim_op["to_user_name"] is not None @pytest.mark.asyncio async def test_operations_surface_leaver_employee_no_and_status( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Regression (2026-06-18): the 操作记录 must carry the per-person ``*_employee_no`` + ``*_employment_status`` so the UI can render a departed assignee as 姓名(工号) + 已离职 instead of a bare 工号. A person who held an asset and has since left (employment_status == 'left') must surface their 工号 + 离职状态 on the ``to_user`` of the historical claim operation and as the current assignee — the exact case from the 资产详情 screenshot that motivated this change. We build the end-state directly: flipping an active person to 'left' is guarded (LeaverBlockGuard), but inserting a row that is already 'left' is allowed.""" holder = AuthUser( work_email="leaver-op@test.com", display_name="Departed Holder", employee_no="P0343", employment_status="left", source_system="successfactors", hris_person_id="sf-leaver-op-001", ) db_session.add(holder) await db_session.flush() asset = AssetItem( asset_tag="LEAVER-OP-001", category="laptop", status="held_by_leaver", assigned_to=holder.id, ) db_session.add(asset) await db_session.flush() operation = AssetOperation( operation_number="OP-TEST-LEAVER-001", asset_id=asset.id, operation_type="claim", status="confirmed", to_user_id=holder.id, operator_id=holder.id, notes="Claimed while active, holder later left", ) db_session.add(operation) await db_session.commit() resp = await client.get( f"/api/v1/assets/{asset.id}", headers=admin_headers, ) data = resp.json()["data"] claim_op = next( o for o in data["operations"] if o["operation_type"] == "claim" ) assert claim_op["to_user_employee_no"] == "P0343" assert claim_op["to_user_employment_status"] == "left" # And the same person as the assigned holder on the asset itself. assert data["assigned_to_employee_no"] == "P0343" assert data["assigned_to_employment_status"] == "left" @pytest.mark.asyncio async def test_detail_operations_ordered_desc( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, db_session: AsyncSession, ): """Test that operations are returned newest-first.""" employee_user_obj, _ = employee_user resp = await client.post( "/api/v1/assets/intake", json={"asset_tag": "DETAIL-OP-003", "category": "laptop"}, headers=admin_headers, ) asset_id = resp.json()["data"]["asset"]["id"] await client.post( f"/api/v1/assets/{asset_id}/claim", json={"to_user_id": str(employee_user_obj.id)}, headers=admin_headers, ) resp = await client.get( f"/api/v1/assets/{asset_id}", headers=admin_headers, ) ops = resp.json()["data"]["operations"] assert len(ops) >= 2 # First operation should be newer (claim) than second (intake) assert ops[0]["operation_type"] == "claim" assert ops[1]["operation_type"] == "intake" @pytest.mark.asyncio async def test_detail_still_includes_history( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test that history is still returned alongside operations.""" resp = await client.post( "/api/v1/assets", json={"asset_tag": "DETAIL-OP-004", "category": "laptop"}, headers=admin_headers, ) asset_id = resp.json()["data"]["id"] resp = await client.get( f"/api/v1/assets/{asset_id}", headers=admin_headers, ) data = resp.json()["data"] # Both should be present assert "history" in data assert "operations" in data # Legacy create always adds history assert len(data["history"]) >= 1 @pytest.mark.asyncio async def test_asset_with_no_operations_returns_empty_list( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Test that an asset with no operations returns empty operations list.""" # Legacy create does NOT create an operation (only intake does) asset = AssetItem( asset_tag="DETAIL-OP-005", category="laptop", status="idle", ) db_session.add(asset) await db_session.commit() resp = await client.get( f"/api/v1/assets/{asset.id}", headers=admin_headers, ) data = resp.json()["data"] assert "operations" in data assert data["operations"] == [] class TestAssetEntityConnectivity: """Test purchase/usage entity connectivity on asset detail and lifecycle ops.""" @pytest.mark.asyncio async def test_detail_includes_purchase_and_usage_entity( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, db_session: AsyncSession, ): employee_user_obj, _ = employee_user employee_user_obj.company_name = "PingCAP Pte. Ltd." await db_session.commit() asset = AssetItem( asset_tag="ENTITY-DETAIL-001", category="laptop", status="claimed", asset_name="Entity Detail Test", assigned_to=employee_user_obj.id, purchase_entity="PingCAP Holdings", ) db_session.add(asset) await db_session.commit() response = await client.get( f"/api/v1/assets/{asset.id}", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["purchase_entity"] == "PingCAP Holdings" assert data["user_entity"] == "PingCAP Pte. Ltd." @pytest.mark.asyncio async def test_claim_syncs_usage_company_from_person_entity_when_missing( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, db_session: AsyncSession, ): employee_user_obj, _ = employee_user employee_user_obj.company_name = "PingCAP (US), Inc." await db_session.commit() asset = AssetItem( asset_tag="ENTITY-CLAIM-001", category="laptop", status="idle", ) db_session.add(asset) await db_session.commit() response = await client.post( f"/api/v1/assets/{asset.id}/claim", json={"to_user_id": str(employee_user_obj.id)}, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["asset"]["usage_company"] == "PingCAP (US), Inc." await db_session.refresh(asset) assert asset.usage_company == "PingCAP (US), Inc." @pytest.mark.asyncio async def test_borrow_syncs_usage_company_from_person_entity_when_missing( self, client: AsyncClient, admin_headers: dict, employee_user: tuple, db_session: AsyncSession, ): employee_user_obj, _ = employee_user employee_user_obj.company_name = "平凯星辰（北京）科技有限公司" await db_session.commit() asset = AssetItem( asset_tag="ENTITY-BORROW-001", category="monitor", status="idle", ) db_session.add(asset) await db_session.commit() # PR #164 requires expected_return_date on borrow. from datetime import date, timedelta future = (date.today() + timedelta(days=7)).isoformat() response = await client.post( f"/api/v1/assets/{asset.id}/borrow", json={ "to_user_id": str(employee_user_obj.id), "expected_return_date": future, }, headers=admin_headers, ) assert response.status_code == 200, response.text data = response.json()["data"] assert data["asset"]["usage_company"] == "平凯星辰（北京）科技有限公司" await db_session.refresh(asset) assert asset.usage_company == "平凯星辰（北京）科技有限公司" @pytest.mark.asyncio async def test_change_assignee_syncs_usage_company_from_person_entity_when_missing( self, client: AsyncClient, admin_headers: dict, admin_user: tuple, employee_user: tuple, db_session: AsyncSession, ): admin_user_obj, _ = admin_user employee_user_obj, _ = employee_user employee_user_obj.company_name = "北京星云深图软件科技有限公司" await db_session.commit() asset = AssetItem( asset_tag="ENTITY-TRANSFER-001", category="laptop", status="claimed", assigned_to=admin_user_obj.id, ) db_session.add(asset) await db_session.commit() response = await client.post( f"/api/v1/assets/{asset.id}/change-assignee", json={"to_user_id": str(employee_user_obj.id)}, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["asset"]["usage_company"] == "北京星云深图软件科技有限公司" await db_session.refresh(asset) assert asset.usage_company == "北京星云深图软件科技有限公司" class TestAssetBulkMutations: """Test suite for bulk intake and bulk update.""" @pytest.mark.asyncio async def test_bulk_intake_csv_creates_assets( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): # ``purchase_entity`` is validated against the legal-entity # master (``_resolve_canonical_purchase_entity_name``), so the # row has to exist or every line fails with # PURCHASE_ENTITY_NOT_REGISTERED. db_session.add(AssetLegalEntity( name="PingCAP Holdings", code="PH", is_active=True, )) await db_session.commit() csv_payload = ( "asset_tag,category,asset_name,purchase_cost,purchase_entity\n" "BULK-INT-001,laptop,Bulk Laptop,1200,PingCAP Holdings\n" "BULK-INT-002,monitor,Bulk Monitor,800,PingCAP Holdings\n" ) response = await client.post( "/api/v1/assets/intake/bulk", headers=admin_headers, files={"file": ("bulk-intake.csv", io.BytesIO(csv_payload.encode("utf-8")), "text/csv")}, ) assert response.status_code == 200 data = response.json()["data"] assert data["total_rows"] == 2 assert data["successful_rows"] == 2 assert data["failed_rows"] == 0 result = await db_session.execute( select(AssetItem).where(AssetItem.asset_tag.in_(["BULK-INT-001", "BULK-INT-002"])) ) assets = result.scalars().all() assert len(assets) == 2 @pytest.mark.asyncio async def test_bulk_update_csv_updates_assets( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): # Seed the managed-location master so the resolver gate the # ``update_asset`` service applies (every PATCH routes # ``location`` through ``ManagedLocationResolver`` — see # ``app/domains/asset/location_resolver.py``) recognises # the CSV values. Without these rows, ``上海`` / ``深圳`` # would NOT exist in the active master, the resolver would # return ``None``, and ``update_asset`` would silently drop # the field from the patch (Round-1 review regression # finding 2026-05-30) — the test would then see locations # stay as their original values and the assertions below # would fail. Mirrors ``location_master`` in # ``test_location_resolver.py``. from app.domains.asset.models import AssetLocation for code, name, order in [ ("P-CN03", "北京", 1), ("P-CN07", "上海", 5), ("P-CN08", "深圳", 6), ]: db_session.add(AssetLocation( code=code, name=name, sort_order=order, is_active=True, )) # ``purchase_entity`` has its own master gate — the CSV names # PingCAP Pte. Ltd., so that legal entity has to be registered. db_session.add(AssetLegalEntity( name="PingCAP Pte. Ltd.", code="SG", is_active=True, )) await db_session.flush() asset_one = AssetItem(asset_tag="BULK-UPD-001", category="laptop", status="idle", asset_name="Old Name") asset_two = AssetItem(asset_tag="BULK-UPD-002", category="monitor", status="idle", location="北京") db_session.add_all([asset_one, asset_two]) await db_session.commit() csv_payload = ( "asset_tag,asset_name,location,purchase_entity\n" "BULK-UPD-001,Updated Name,上海,PingCAP Pte. Ltd.\n" "BULK-UPD-002,,深圳,\n" ) response = await client.post( "/api/v1/assets/bulk-update", headers=admin_headers, files={"file": ("bulk-update.csv", io.BytesIO(csv_payload.encode("utf-8")), "text/csv")}, ) assert response.status_code == 200 data = response.json()["data"] assert data["total_rows"] == 2 assert data["successful_rows"] == 2 assert data["failed_rows"] == 0 await db_session.refresh(asset_one) await db_session.refresh(asset_two) assert asset_one.asset_name == "Updated Name" assert asset_one.location == "上海" assert asset_one.purchase_entity == "PingCAP Pte. Ltd." assert asset_two.location == "深圳" @pytest.mark.asyncio async def test_bulk_update_missing_asset_reports_failure( self, client: AsyncClient, admin_headers: dict, ): csv_payload = "asset_tag,asset_name\nMISSING-ASSET,Nope\n" response = await client.post( "/api/v1/assets/bulk-update", headers=admin_headers, files={"file": ("bulk-update.csv", io.BytesIO(csv_payload.encode("utf-8")), "text/csv")}, ) assert response.status_code == 200 data = response.json()["data"] assert data["successful_rows"] == 0 assert data["failed_rows"] == 1 assert data["results"][0]["status"] == "failed" class TestAssetRelationshipModeling: """Test supplier / procurement / software / payment linkage.""" @pytest.mark.asyncio async def test_procurement_order_can_link_supplier( self, client: AsyncClient, admin_headers: dict, ): supplier_resp = await client.post( "/api/v1/assets/suppliers", json={ "name": "Dell China", "contact_name": "Alice", "contact_phone": "13800000000", }, headers=admin_headers, ) assert supplier_resp.status_code == 200 supplier_id = supplier_resp.json()["data"]["id"] response = await client.post( "/api/v1/assets/procurement", json={ "order_type": "purchase", "supplier_id": supplier_id, "notes": "Supplier linked order", }, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["supplier_id"] == supplier_id assert data["supplier_name"] == "Dell China" @pytest.mark.asyncio async def test_business_payment_can_link_software_procurement_and_supplier( self, client: AsyncClient, admin_headers: dict, ): software_resp = await client.post( "/api/v1/assets/software-catalog", json={"name": "Notion", "status": "active"}, headers=admin_headers, ) assert software_resp.status_code == 200 software_id = software_resp.json()["data"]["id"] supplier_resp = await client.post( "/api/v1/assets/suppliers", json={"name": "Acme Supplier"}, headers=admin_headers, ) assert supplier_resp.status_code == 200 supplier_id = supplier_resp.json()["data"]["id"] procurement_resp = await client.post( "/api/v1/assets/procurement", json={ "order_type": "purchase", "supplier_id": supplier_id, "notes": "Linked procurement", }, headers=admin_headers, ) assert procurement_resp.status_code == 200 procurement = procurement_resp.json()["data"] response = await client.post( "/api/v1/assets/business-payments", json={ "software_catalog_id": software_id, "procurement_order_id": procurement["id"], "supplier_id": supplier_id, "amount": "100.00", }, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["software_catalog_id"] == software_id assert data["procurement_order_id"] == procurement["id"] assert data["supplier_id"] == supplier_id assert data["software_name"] == "Notion" assert data["po_number"] == procurement["order_number"] assert data["signing_entity"] == "Acme Supplier" @pytest.mark.asyncio async def test_business_payment_requester_reference_data_returns_people( self, client: AsyncClient, admin_headers: dict, employee_user, ): employee, _ = employee_user response = await client.get( "/api/v1/assets/reference-data/business-payment-requesters", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert any( item["id"] == str(employee.id) and item["display_name"] == employee.display_name for item in data ) class TestProcurementOrders: """Test suite for procurement order CRUD.""" @pytest.mark.asyncio async def test_create_procurement_order( self, client: AsyncClient, admin_headers: dict, ): """Test creating a procurement order.""" payload = { "order_type": "purchase", "department": "Engineering", "supplier_name": "Dell Direct", "total_amount": "15000.00", "items": [ { "name": "Latitude 5540", "category": "laptop", "brand": "Dell", "quantity": 10, "unit_price": 1500, } ], "notes": "Q2 laptop refresh", } response = await client.post( "/api/v1/assets/procurement", json=payload, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["order_type"] == "purchase" assert data["status"] == "draft" assert data["order_number"].startswith("PO-") assert data["department"] == "Engineering" assert float(data["total_amount"]) == 15000.00 @pytest.mark.asyncio async def test_list_procurement_orders( self, client: AsyncClient, admin_headers: dict, ): """Test listing procurement orders.""" # Create two orders for i in range(2): await client.post( "/api/v1/assets/procurement", json={"order_type": "purchase", "notes": f"Order {i}"}, headers=admin_headers, ) response = await client.get( "/api/v1/assets/procurement", headers=admin_headers, ) assert response.status_code == 200 assert response.json()["meta"]["total"] >= 2 @pytest.mark.asyncio async def test_get_procurement_order_detail( self, client: AsyncClient, admin_headers: dict, ): """Test getting procurement order detail.""" create_resp = await client.post( "/api/v1/assets/procurement", json={"order_type": "lease", "department": "Finance"}, headers=admin_headers, ) order_id = create_resp.json()["data"]["id"] response = await client.get( f"/api/v1/assets/procurement/{order_id}", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["id"] == order_id assert data["order_type"] == "lease" assert data["department"] == "Finance" @pytest.mark.asyncio async def test_update_procurement_order( self, client: AsyncClient, admin_headers: dict, ): """Test updating a procurement order.""" create_resp = await client.post( "/api/v1/assets/procurement", json={"order_type": "purchase", "notes": "Initial"}, headers=admin_headers, ) order_id = create_resp.json()["data"]["id"] response = await client.patch( f"/api/v1/assets/procurement/{order_id}", json={ "status": "pending", "supplier_name": "HP Direct", "total_amount": "25000.00", }, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["status"] == "pending" assert data["supplier_name"] == "HP Direct" assert float(data["total_amount"]) == 25000.00 @pytest.mark.asyncio async def test_filter_procurement_by_status( self, client: AsyncClient, admin_headers: dict, ): """Test filtering procurement orders by status.""" # Create and update one to 'pending' create_resp = await client.post( "/api/v1/assets/procurement", json={"order_type": "purchase"}, headers=admin_headers, ) order_id = create_resp.json()["data"]["id"] await client.patch( f"/api/v1/assets/procurement/{order_id}", json={"status": "pending"}, headers=admin_headers, ) response = await client.get( "/api/v1/assets/procurement?status=pending", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert all(o["status"] == "pending" for o in data) @pytest.mark.asyncio async def test_procurement_not_found( self, client: AsyncClient, admin_headers: dict, ): """Test getting a non-existent procurement order.""" response = await client.get( f"/api/v1/assets/procurement/{uuid4()}", headers=admin_headers, ) assert response.status_code == 404 @pytest.mark.asyncio async def test_employee_cannot_access_procurement( self, client: AsyncClient, employee_headers: dict, ): """Test that regular employee cannot access procurement orders.""" response = await client.get( "/api/v1/assets/procurement", headers=employee_headers, ) assert response.status_code == 403 class TestEmployeeRequests: """Test suite for employee request CRUD.""" @pytest.mark.asyncio async def test_create_employee_request( self, client: AsyncClient, employee_headers: dict, ): """Test creating an employee request.""" payload = { "request_type": "claim", "category": "laptop", "reason": "New hire needs laptop", } response = await client.post( "/api/v1/assets/employee-requests", json=payload, headers=employee_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["request_type"] == "claim" assert data["status"] == "pending" assert data["request_number"].startswith("REQ-") assert data["category"] == "laptop" @pytest.mark.asyncio async def test_create_request_for_specific_asset( self, client: AsyncClient, employee_headers: dict, db_session: AsyncSession, ): """Test creating a request for a specific asset.""" asset = AssetItem( asset_tag="REQ-ASSET-001", category="monitor", status="idle", ) db_session.add(asset) await db_session.commit() response = await client.post( "/api/v1/assets/employee-requests", json={ "request_type": "self_claim", "asset_id": str(asset.id), "reason": "Need monitor for dual setup", }, headers=employee_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["asset_id"] == str(asset.id) assert data["request_type"] == "self_claim" @pytest.mark.asyncio async def test_list_employee_requests_admin_sees_all( self, client: AsyncClient, admin_headers: dict, employee_headers: dict, ): """Test that admin can see all employee requests.""" # Employee creates a request await client.post( "/api/v1/assets/employee-requests", json={"request_type": "claim", "category": "laptop"}, headers=employee_headers, ) # Admin lists requests response = await client.get( "/api/v1/assets/employee-requests", headers=admin_headers, ) assert response.status_code == 200 assert response.json()["meta"]["total"] >= 1 @pytest.mark.asyncio async def test_list_employee_requests_employee_sees_own( self, client: AsyncClient, admin_headers: dict, employee_headers: dict, employee_user: tuple, ): """Test that employee can only see their own requests.""" employee_user_obj, _ = employee_user # Employee creates a request await client.post( "/api/v1/assets/employee-requests", json={"request_type": "repair", "reason": "Keyboard broken"}, headers=employee_headers, ) # Employee lists requests (should only see their own) response = await client.get( "/api/v1/assets/employee-requests", headers=employee_headers, ) assert response.status_code == 200 data = response.json()["data"] assert all(r["requester_id"] == str(employee_user_obj.id) for r in data) @pytest.mark.asyncio async def test_approve_employee_request( self, client: AsyncClient, admin_headers: dict, employee_headers: dict, ): """Test approving an employee request.""" # Create request create_resp = await client.post( "/api/v1/assets/employee-requests", json={"request_type": "claim", "category": "accessory"}, headers=employee_headers, ) request_id = create_resp.json()["data"]["id"] # Admin approves response = await client.post( f"/api/v1/assets/employee-requests/{request_id}/process", json={"action": "approve", "notes": "Approved - will dispatch accessory"}, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["status"] == "approved" assert data["processed_at"] is not None assert data["notes"] == "Approved - will dispatch accessory" @pytest.mark.asyncio async def test_reject_employee_request( self, client: AsyncClient, admin_headers: dict, employee_headers: dict, ): """Test rejecting an employee request.""" create_resp = await client.post( "/api/v1/assets/employee-requests", json={"request_type": "self_borrow", "category": "monitor"}, headers=employee_headers, ) request_id = create_resp.json()["data"]["id"] response = await client.post( f"/api/v1/assets/employee-requests/{request_id}/process", json={"action": "reject", "notes": "No monitors available"}, headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] assert data["status"] == "rejected" assert data["notes"] == "No monitors available" @pytest.mark.asyncio async def test_process_non_pending_rejected( self, client: AsyncClient, admin_headers: dict, employee_headers: dict, ): """Test that processing a non-pending request is rejected.""" create_resp = await client.post( "/api/v1/assets/employee-requests", json={"request_type": "return", "reason": "Done with it"}, headers=employee_headers, ) request_id = create_resp.json()["data"]["id"] # Approve it first await client.post( f"/api/v1/assets/employee-requests/{request_id}/process", json={"action": "approve"}, headers=admin_headers, ) # Try to approve again response = await client.post( f"/api/v1/assets/employee-requests/{request_id}/process", json={"action": "reject"}, headers=admin_headers, ) assert response.status_code == 400 assert "not pending" in response.json()["errors"][0]["message"].lower() @pytest.mark.asyncio async def test_employee_cannot_process_requests( self, client: AsyncClient, employee_headers: dict, ): """Test that employees cannot process requests.""" create_resp = await client.post( "/api/v1/assets/employee-requests", json={"request_type": "claim", "category": "laptop"}, headers=employee_headers, ) request_id = create_resp.json()["data"]["id"] response = await client.post( f"/api/v1/assets/employee-requests/{request_id}/process", json={"action": "approve"}, headers=employee_headers, ) assert response.status_code == 403 @pytest.mark.asyncio async def test_invalid_request_type_rejected( self, client: AsyncClient, employee_headers: dict, ): """Test that invalid request types are rejected.""" response = await client.post( "/api/v1/assets/employee-requests", json={"request_type": "invalid_type"}, headers=employee_headers, ) assert response.status_code == 422 assert "Request type must be one of" in str(response.json()) class TestAssetSorting: """Test suite for asset list sorting.""" @pytest.mark.asyncio async def test_default_sort_created_at_desc( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Default sort should be created_at descending.""" now = datetime.now(UTC) assets = [ AssetItem( asset_tag="SORT-OLD", category="laptop", status="idle", created_at=now - timedelta(days=2), updated_at=now - timedelta(days=2), ), AssetItem( asset_tag="SORT-NEW", category="laptop", status="idle", created_at=now, updated_at=now, ), ] db_session.add_all(assets) await db_session.commit() response = await client.get( "/api/v1/assets?status=idle", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] tags = [item["asset_tag"] for item in data if item["asset_tag"].startswith("SORT-")] assert tags.index("SORT-NEW") < tags.index("SORT-OLD") @pytest.mark.asyncio async def test_sort_by_asset_tag_asc( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Sort by asset_tag ascending.""" assets = [ AssetItem(asset_tag="ZZZ-SORT", category="laptop", status="idle"), AssetItem(asset_tag="AAA-SORT", category="laptop", status="idle"), ] db_session.add_all(assets) await db_session.commit() response = await client.get( "/api/v1/assets?sort_by=asset_tag&sort_order=asc", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] tags = [item["asset_tag"] for item in data] assert tags.index("AAA-SORT") < tags.index("ZZZ-SORT") @pytest.mark.asyncio async def test_sort_by_purchase_cost_desc( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Sort by purchase_cost descending.""" assets = [ AssetItem(asset_tag="COST-LOW", category="laptop", status="idle", purchase_cost=100), AssetItem(asset_tag="COST-HIGH", category="laptop", status="idle", purchase_cost=9999), ] db_session.add_all(assets) await db_session.commit() response = await client.get( "/api/v1/assets?sort_by=purchase_cost&sort_order=desc", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] tags = [item["asset_tag"] for item in data] assert tags.index("COST-HIGH") < tags.index("COST-LOW") @pytest.mark.asyncio async def test_sort_by_purchase_date_asc( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Sort by purchase_date ascending.""" assets = [ AssetItem( asset_tag="DATE-NEW", category="laptop", status="idle", purchase_date=date(2025, 12, 1), ), AssetItem( asset_tag="DATE-OLD", category="laptop", status="idle", purchase_date=date(2020, 1, 1), ), ] db_session.add_all(assets) await db_session.commit() response = await client.get( "/api/v1/assets?sort_by=purchase_date&sort_order=asc", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] tags = [item["asset_tag"] for item in data] assert tags.index("DATE-OLD") < tags.index("DATE-NEW") @pytest.mark.asyncio async def test_sort_nullable_field_nulls_last( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Null values should always sort last regardless of sort order.""" assets = [ AssetItem(asset_tag="NULLS-HAS", category="laptop", status="idle", purchase_cost=500), AssetItem(asset_tag="NULLS-NONE", category="laptop", status="idle", purchase_cost=None), ] db_session.add_all(assets) await db_session.commit() # Ascending: nulls should be last response = await client.get( "/api/v1/assets?sort_by=purchase_cost&sort_order=asc", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] tags = [item["asset_tag"] for item in data if item["asset_tag"].startswith("NULLS-")] assert tags == ["NULLS-HAS", "NULLS-NONE"] # Descending: nulls should still be last response = await client.get( "/api/v1/assets?sort_by=purchase_cost&sort_order=desc", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] tags = [item["asset_tag"] for item in data if item["asset_tag"].startswith("NULLS-")] assert tags == ["NULLS-HAS", "NULLS-NONE"] @pytest.mark.asyncio async def test_sort_invalid_field_rejected( self, client: AsyncClient, admin_headers: dict, ): """Invalid sort field should return 400.""" response = await client.get( "/api/v1/assets?sort_by=hacked_field", headers=admin_headers, ) assert response.status_code == 400 @pytest.mark.asyncio async def test_sort_invalid_order_rejected( self, client: AsyncClient, admin_headers: dict, ): """Invalid sort order should return 422.""" response = await client.get( "/api/v1/assets?sort_by=created_at&sort_order=random", headers=admin_headers, ) assert response.status_code == 422 @pytest.mark.asyncio async def test_sort_with_filters_combined( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Sort should work together with filters.""" assets = [ AssetItem(asset_tag="CF-A", category="monitor", status="idle", purchase_cost=200), AssetItem(asset_tag="CF-B", category="monitor", status="idle", purchase_cost=800), AssetItem(asset_tag="CF-C", category="laptop", status="idle", purchase_cost=500), ] db_session.add_all(assets) await db_session.commit() response = await client.get( "/api/v1/assets?category=monitor&sort_by=purchase_cost&sort_order=asc", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] tags = [item["asset_tag"] for item in data] assert tags.index("CF-A") < tags.index("CF-B") assert "CF-C" not in tags @pytest.mark.asyncio async def test_sort_by_updated_at( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Sort by updated_at should be accepted and return updated_at in response.""" now = datetime.now(UTC) assets = [ AssetItem( asset_tag="UPD-OLD", category="laptop", status="idle", created_at=now - timedelta(days=5), updated_at=now - timedelta(days=3), ), AssetItem( asset_tag="UPD-NEW", category="laptop", status="idle", created_at=now - timedelta(days=5), updated_at=now, ), ] db_session.add_all(assets) await db_session.commit() response = await client.get( "/api/v1/assets?sort_by=updated_at&sort_order=desc", headers=admin_headers, ) assert response.status_code == 200 data = response.json()["data"] # Response should include updated_at field upd_items = [item for item in data if item["asset_tag"].startswith("UPD-")] assert len(upd_items) == 2 assert upd_items[0]["asset_tag"] == "UPD-NEW" assert upd_items[1]["asset_tag"] == "UPD-OLD" # updated_at must be present in response assert "updated_at" in upd_items[0] assert upd_items[0]["updated_at"] is not None class TestIntakeDefaults: """Verify intake writes correct default values for source_type and device_status.""" @pytest.mark.asyncio async def test_intake_defaults_source_type_purchased( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Intake without explicit source_type should default to 'purchased'.""" payload = { "asset_tag": "DFLT-SRC-001", "category": "laptop", "asset_name": "Default Source Test", } response = await client.post( "/api/v1/assets/intake", json=payload, headers=admin_headers, ) assert response.status_code == 200 asset_data = response.json()["data"]["asset"] assert asset_data["source_type"] == "purchased" # Verify in DB result = await db_session.execute( select(AssetItem).where(AssetItem.asset_tag == "DFLT-SRC-001") ) asset = result.scalar_one() assert asset.source_type == "purchased" @pytest.mark.asyncio async def test_intake_defaults_device_status_normal( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Intake without explicit device_status should default to 'normal'.""" payload = { "asset_tag": "DFLT-DEV-001", "category": "monitor", "asset_name": "Default Device Test", } response = await client.post( "/api/v1/assets/intake", json=payload, headers=admin_headers, ) assert response.status_code == 200 asset_data = response.json()["data"]["asset"] assert asset_data["device_status"] == "normal" # Verify in DB result = await db_session.execute( select(AssetItem).where(AssetItem.asset_tag == "DFLT-DEV-001") ) asset = result.scalar_one() assert asset.device_status == "normal" @pytest.mark.asyncio async def test_intake_explicit_source_type_preserved( self, client: AsyncClient, admin_headers: dict, ): """Intake with explicit source_type should preserve the value.""" payload = { "asset_tag": "DFLT-SRC-002", "category": "laptop", "source_type": "leased", } response = await client.post( "/api/v1/assets/intake", json=payload, headers=admin_headers, ) assert response.status_code == 200 assert response.json()["data"]["asset"]["source_type"] == "leased" class TestAssetTagAutoGeneration: """Tag-generation rule: ``IT-{purchase_entity.code}-{code}-{YYYYMMDDHHMMSS}`` where ``code`` comes from the ``asset_category_configs`` row matching the asset category (seeded in conftest, mirrors migration 205). """ @pytest.mark.asyncio async def test_intake_auto_generates_tag( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Submitting intake with ``purchase_entity_id`` and no ``asset_tag`` should auto-generate the tag in the format ``IT-{entity.code}-{category code}-{14-digit timestamp}``. The category code (``laptop`` → ``NB``) comes from the seeded ``asset_category_configs`` row, not a hardcoded mapping. """ entity = AssetLegalEntity(name="PingCAP CN", code="CN", is_active=True) db_session.add(entity) await db_session.commit() response = await client.post( "/api/v1/assets/intake", json={ "category": "laptop", "purchase_entity_id": str(entity.id), }, headers=admin_headers, ) assert response.status_code == 200, response.text tag = response.json()["data"]["asset"]["asset_tag"] # IT-CN-NB-YYYYMMDDHHMMSS = 4 + 1 + 2 + 1 + 2 + 1 + 14 = 25 chars assert tag.startswith("IT-CN-NB-") assert len(tag) == len("IT-CN-NB-") + 14 assert tag[len("IT-CN-NB-"):].isdigit() # purchase_entity (free-text) should mirror the resolved name assert response.json()["data"]["asset"]["purchase_entity"] == "PingCAP CN" @pytest.mark.asyncio async def test_tag_timestamp_is_beijing_time( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """The 14-digit timestamp must be Asia/Shanghai (UTC+8) wall-clock, not the server's UTC — regardless of container TZ. """ entity = AssetLegalEntity(name="PingCAP BJ", code="BJ", is_active=True) db_session.add(entity) await db_session.commit() response = await client.post( "/api/v1/assets/intake", json={"category": "laptop", "purchase_entity_id": str(entity.id)}, headers=admin_headers, ) assert response.status_code == 200, response.text stamp = response.json()["data"]["asset"]["asset_tag"].rsplit("-", 1)[1] parsed = datetime.strptime(stamp, "%Y%m%d%H%M%S") beijing_now = datetime.now(ZoneInfo("Asia/Shanghai")).replace(tzinfo=None) # Within 2 min of Beijing now (proves +8, not UTC which is 8h off). assert abs((parsed - beijing_now).total_seconds()) < 120, ( f"tag stamp {parsed} not aligned to Beijing {beijing_now}" ) @pytest.mark.asyncio async def test_intake_without_entity_when_tag_missing_400s( self, client: AsyncClient, admin_headers: dict, ): """Auto-generation needs ``purchase_entity_id``. Without it AND without ``asset_tag``, intake must reject with the clear ``PURCHASE_ENTITY_REQUIRED`` code (not a 500 / generic error). """ response = await client.post( "/api/v1/assets/intake", json={"category": "laptop"}, headers=admin_headers, ) assert response.status_code == 400 assert response.json()["errors"][0]["code"] == "PURCHASE_ENTITY_REQUIRED" @pytest.mark.asyncio async def test_intake_with_entity_missing_code_400s( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """An entity row without ``code`` cannot drive tag generation — admins must fill the code in the 公司主体信息 admin tab first. Service surfaces ``LEGAL_ENTITY_MISSING_CODE``. """ entity = AssetLegalEntity(name="Codeless Co", code=None, is_active=True) db_session.add(entity) await db_session.commit() response = await client.post( "/api/v1/assets/intake", json={ "category": "laptop", "purchase_entity_id": str(entity.id), }, headers=admin_headers, ) assert response.status_code == 400 assert response.json()["errors"][0]["code"] == "LEGAL_ENTITY_MISSING_CODE" @pytest.mark.asyncio async def test_intake_with_inactive_entity_400s( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Inactive entities can't be picked at intake. The validation runs in ``intake_asset`` before tag generation so the error code is ``LEGAL_ENTITY_INACTIVE`` (not ``MISSING_CODE``). """ entity = AssetLegalEntity(name="Retired Co", code="OLD", is_active=False) db_session.add(entity) await db_session.commit() response = await client.post( "/api/v1/assets/intake", json={ "category": "laptop", "purchase_entity_id": str(entity.id), }, headers=admin_headers, ) assert response.status_code == 400 assert response.json()["errors"][0]["code"] == "LEGAL_ENTITY_INACTIVE" @pytest.mark.asyncio async def test_intake_explicit_tag_skips_auto_generation( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Admins migrating legacy assets with externally-printed labels can still pass ``asset_tag`` explicitly; the service honours it as-is and skips the auto-generation path. """ entity = AssetLegalEntity(name="PingCAP US", code="US", is_active=True) db_session.add(entity) await db_session.commit() response = await client.post( "/api/v1/assets/intake", json={ "asset_tag": "LEGACY-LABEL-2019-007", "category": "laptop", "purchase_entity_id": str(entity.id), }, headers=admin_headers, ) assert response.status_code == 200 assert response.json()["data"]["asset"]["asset_tag"] == "LEGACY-LABEL-2019-007" @pytest.mark.asyncio async def test_generate_tag_endpoint_resolves_code_from_config( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """``POST /assets/generate-tag`` takes a category *enum* and resolves its tag segment from ``asset_category_configs`` (``laptop`` → ``NB``, ``desktop`` → ``DT``). """ entity = AssetLegalEntity(name="PingCAP SG", code="SG", is_active=True) db_session.add(entity) await db_session.commit() for category, expected_segment in (("laptop", "NB"), ("desktop", "DT")): response = await client.post( "/api/v1/assets/generate-tag", json={ "purchase_entity_id": str(entity.id), "category": category, }, headers=admin_headers, ) assert response.status_code == 200, response.text tag = response.json()["data"]["asset_tag"] assert tag.startswith(f"IT-SG-{expected_segment}-"), tag @pytest.mark.asyncio async def test_tag_code_follows_config_edit( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Editing the category's ``code`` in 分类管理 changes the tag segment for new intakes — proving the mapping is DB-driven, not hardcoded. (Drives the whole point of this feature.) """ entity = AssetLegalEntity(name="PingCAP HK", code="HK", is_active=True) db_session.add(entity) await db_session.commit() # Flip desktop's code DT → WS via the admin update path. cfg = (await db_session.execute( select(AssetCategoryConfig).where( AssetCategoryConfig.asset_category == "desktop" ) )).scalar_one() cfg.code = "WS" await db_session.commit() response = await client.post( "/api/v1/assets/generate-tag", json={"purchase_entity_id": str(entity.id), "category": "desktop"}, headers=admin_headers, ) assert response.status_code == 200, response.text assert response.json()["data"]["asset_tag"].startswith("IT-HK-WS-") @pytest.mark.asyncio async def test_tag_generation_missing_config_400s( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """A category with no active config row cannot mint a tag — fail loud with ``CATEGORY_TAG_CODE_MISSING`` rather than a garbage tag. """ entity = AssetLegalEntity(name="PingCAP TW", code="TW", is_active=True) db_session.add(entity) # Deactivate laptop's config so the lookup misses. cfg = (await db_session.execute( select(AssetCategoryConfig).where( AssetCategoryConfig.asset_category == "laptop" ) )).scalar_one() cfg.is_active = False await db_session.commit() response = await client.post( "/api/v1/assets/generate-tag", json={"purchase_entity_id": str(entity.id), "category": "laptop"}, headers=admin_headers, ) assert response.status_code == 400 assert response.json()["errors"][0]["code"] == "CATEGORY_TAG_CODE_MISSING" class TestCategoryConfigProtection: """Canonical (``asset_category``-linked) categories back the intake form's fixed dropdown + tag-code lookup, so they must always exist and stay active. Admins may edit their ``code`` but not deactivate/delete them. """ async def _desktop_category_id(self, db_session: AsyncSession) -> str: cfg = (await db_session.execute( select(AssetCategoryConfig).where( AssetCategoryConfig.asset_category == "desktop" ) )).scalar_one() return str(cfg.id) @pytest.mark.asyncio async def test_delete_system_category_blocked( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): cat_id = await self._desktop_category_id(db_session) response = await client.delete( f"/api/v1/assets/categories/{cat_id}", headers=admin_headers, ) assert response.status_code == 409 assert response.json()["errors"][0]["code"] == "CATEGORY_SYSTEM_PROTECTED" @pytest.mark.asyncio async def test_deactivate_system_category_blocked( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): cat_id = await self._desktop_category_id(db_session) response = await client.patch( f"/api/v1/assets/categories/{cat_id}", json={"is_active": False}, headers=admin_headers, ) assert response.status_code == 409 assert response.json()["errors"][0]["code"] == "CATEGORY_SYSTEM_PROTECTED" @pytest.mark.asyncio async def test_edit_system_category_code_allowed( self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession, ): """Editing the code IS the point — must succeed (and then drive tags).""" cat_id = await self._desktop_category_id(db_session) response = await client.patch( f"/api/v1/assets/categories/{cat_id}", json={"code": "WS"}, headers=admin_headers, ) assert response.status_code == 200, response.text assert response.json()["data"]["code"] == "WS" @pytest.mark.asyncio async def test_duplicate_category_code_returns_409_not_500( self, client: AsyncClient, admin_headers: dict, ): """Creating a category with an already-used code must be a clean 409 — regression for the positional-arg ``AppException`` crash. """ response = await client.post( "/api/v1/assets/categories", json={"code": "DT", "name": "Dup"}, # DT already seeded (desktop) headers=admin_headers, ) assert response.status_code == 409 assert response.json()["errors"][0]["code"] == "CATEGORY_CODE_EXISTS" @pytest.mark.asyncio async def test_nonsystem_category_deletable( self, client: AsyncClient, admin_headers: dict, ): """An admin-created extra category (no asset_category) is freely deletable — it backs no intake option and labels no asset. """ created = await client.post( "/api/v1/assets/categories", json={"code": "ZZ", "name": "Custom"}, headers=admin_headers, ) assert created.status_code == 200, created.text new_id = created.json()["data"]["id"] assert created.json()["data"]["asset_category"] is None deleted = await client.delete( f"/api/v1/assets/categories/{new_id}", headers=admin_headers, ) assert deleted.status_code == 200, deleted.text