"""Tests for the IT-admin "审批中心" (unified all-approvals) endpoints. Backend routes covered: - ``GET /api/v1/approval/admin/requests`` — paginated, filterable list across every employee_requests row. - ``GET /api/v1/approval/admin/requests/{id}`` — drawer detail with the live Feishu approval timeline (or a graceful fallback when Feishu is unreachable, surfaced via ``timeline_error``). """ from unittest.mock import AsyncMock, patch import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from app.domains.asset.models import AssetEmployeeRequest _FAKE_FEISHU_CFG = { "app_id": "cli_test", "app_secret": "secret_test", "event_verification_token": "", } @pytest.fixture(autouse=True) def _patch_feishu_config(monkeypatch): async def _fake_cfg(**_kwargs): return dict(_FAKE_FEISHU_CFG) monkeypatch.setattr("app.integrations.compat.get_feishu_config", _fake_cfg) monkeypatch.setattr("app.approval.service.get_feishu_config", _fake_cfg) monkeypatch.setattr( "app.approval.router.get_feishu_config", _fake_cfg, raising=False, ) async def _fake_token(self): return "fake-tenant-token" monkeypatch.setattr( "app.approval.feishu_client.FeishuApprovalClient.get_tenant_access_token", _fake_token, ) async def _seed_request( db: AsyncSession, *, requester_id, request_type: str = "laptop_replacement", status: str = "pending", request_number: str = "REQ-ADMIN-001", reason: str | None = "broken hinge", instance_code: str | None = "feishu_admin_inst_001", approval_name: str | None = "硬件更换审批", latest_instance_detail: dict | None = None, ) -> AssetEmployeeRequest: feishu = {} if instance_code: feishu = { "approval_code": "APPROVAL_CODE_X", "approval_name": approval_name, "instance_code": instance_code, "instance_status": "PENDING", } if latest_instance_detail is not None: feishu["latest_instance_detail"] = latest_instance_detail row = AssetEmployeeRequest( request_number=request_number, request_type=request_type, status=status, requester_id=requester_id, reason=reason, detail={"feishu": feishu} if feishu else {}, ) db.add(row) await db.flush() await db.commit() return row class TestAdminApprovalCenterAPI: @pytest.mark.asyncio async def test_admin_lists_expose_cached_current_approvers( self, client: AsyncClient, db_session: AsyncSession, admin_headers: dict, employee_user, ): employee, _ = employee_user await _seed_request( db_session, requester_id=employee.id, request_number="REQ-APPROVER-001", latest_instance_detail={ "task_list": [ { "node_id": "manager", "node_name": "Manager approval", "user_id": "ou-manager", "user_name": "Alice Manager", "status": "PENDING", }, ], }, ) for path in ("/api/v1/approval/admin/overview", "/api/v1/approval/admin/all-requests"): resp = await client.get(path, headers=admin_headers) assert resp.status_code == 200 items = resp.json()["data"]["items"] item = next(row for row in items if row["subtitle"] == "REQ-APPROVER-001") assert item["current_approvers"] == ["Alice Manager"] @pytest.mark.asyncio async def test_list_requires_admin( self, client: AsyncClient, db_session: AsyncSession, employee_user, employee_headers: dict, ): emp_user, _ = employee_user await _seed_request(db_session, requester_id=emp_user.id) resp = await client.get( "/api/v1/approval/admin/requests", headers=employee_headers, ) assert resp.status_code == 403 @pytest.mark.asyncio async def test_list_returns_all_kinds_for_admin( self, client: AsyncClient, db_session: AsyncSession, admin_user, admin_headers: dict, employee_user, ): emp_user, _ = employee_user admin, _ = admin_user await _seed_request( db_session, requester_id=emp_user.id, request_type="laptop_replacement", request_number="REQ-A-1", ) await _seed_request( db_session, requester_id=admin.id, request_type="api_key_request", request_number="REQ-A-2", ) await _seed_request( db_session, requester_id=emp_user.id, request_type="google_mail_group", request_number="REQ-A-3", ) resp = await client.get( "/api/v1/approval/admin/requests", headers=admin_headers, ) assert resp.status_code == 200 body = resp.json() assert body["success"] is True assert body["meta"]["total"] >= 3 request_numbers = {row["request_number"] for row in body["data"]} assert {"REQ-A-1", "REQ-A-2", "REQ-A-3"}.issubset(request_numbers) first = next(r for r in body["data"] if r["request_number"] == "REQ-A-1") assert first["requester_email"] == emp_user.work_email assert first["approval_instance_code"] == "feishu_admin_inst_001" @pytest.mark.asyncio async def test_list_keyword_search_matches_requester( self, client: AsyncClient, db_session: AsyncSession, admin_headers: dict, employee_user, ): emp_user, _ = employee_user await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-K-1", reason="company laptop dead", ) # Search by requester display_name should hit even though the # request_number / reason don't match the keyword. resp = await client.get( "/api/v1/approval/admin/requests", headers=admin_headers, params={"keyword": emp_user.display_name}, ) assert resp.status_code == 200 rows = resp.json()["data"] assert any(r["request_number"] == "REQ-K-1" for r in rows) @pytest.mark.asyncio async def test_list_filter_by_requester_and_status( self, client: AsyncClient, db_session: AsyncSession, admin_user, admin_headers: dict, employee_user, ): emp_user, _ = employee_user admin, _ = admin_user await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-F-EMP-PENDING", status="pending", ) await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-F-EMP-APPROVED", status="approved", ) await _seed_request( db_session, requester_id=admin.id, request_number="REQ-F-ADMIN-PENDING", status="pending", ) resp = await client.get( "/api/v1/approval/admin/requests", headers=admin_headers, params={ "requester_id": str(emp_user.id), "status": "pending", }, ) assert resp.status_code == 200 rows = resp.json()["data"] request_numbers = {r["request_number"] for r in rows} assert "REQ-F-EMP-PENDING" in request_numbers assert "REQ-F-EMP-APPROVED" not in request_numbers assert "REQ-F-ADMIN-PENDING" not in request_numbers @pytest.mark.asyncio async def test_detail_includes_live_timeline_from_feishu( self, client: AsyncClient, db_session: AsyncSession, admin_headers: dict, employee_user, ): emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-D-1", instance_code="feishu_admin_inst_detail", ) # Feishu returns a fully-baked instance with one approved task # and one comment. The service merges them and returns a # time-sorted timeline. with patch( "app.approval.service.FeishuApprovalClient.get_instance_detail", new=AsyncMock( return_value={ "instance_code": "feishu_admin_inst_detail", "status": "APPROVED", "form": [], "task_list": [ { "user_id": "FEISHU_USR_1", "open_id": "ou_1", "user_name": "Manager Mei", "status": "APPROVED", "start_time": "1714400000000", "end_time": "1714400600000", }, ], "comment_list": [ { "user_id": "FEISHU_USR_2", "open_id": "ou_2", "user_name": "Director Li", "comment": "ok looks fine", "create_time": "1714400900000", }, ], }, ), ): resp = await client.get( f"/api/v1/approval/admin/requests/{row.id}", headers=admin_headers, ) assert resp.status_code == 200 body = resp.json()["data"] assert body["request_number"] == "REQ-D-1" assert body["timeline_error"] is None timeline = body["timeline"] assert len(timeline) == 2 # First entry = approver action (earlier end_time), second = comment assert timeline[0]["kind"] == "task" assert timeline[0]["status"] == "APPROVED" assert timeline[0]["user_name"] == "Manager Mei" assert timeline[1]["kind"] == "comment" assert timeline[1]["comment"] == "ok looks fine" @pytest.mark.asyncio async def test_detail_falls_back_to_cached_when_feishu_fails( self, client: AsyncClient, db_session: AsyncSession, admin_headers: dict, employee_user, ): emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-D-STALE", instance_code="feishu_admin_inst_stale", ) # Pre-populate detail.feishu.latest_instance_detail to simulate # the snapshot the Feishu callback handler caches on every push. row.detail = { "feishu": { "approval_name": "缓存模板", "instance_code": "feishu_admin_inst_stale", "instance_status": "PENDING", "latest_instance_detail": { "instance_code": "feishu_admin_inst_stale", "status": "PENDING", "task_list": [ { "user_id": "FEISHU_USR_3", "user_name": "Cached Approver", "status": "PENDING", "start_time": 1714000000000, }, ], "comment_list": [], }, }, } db_session.add(row) await db_session.commit() with patch( "app.approval.service.FeishuApprovalClient.get_instance_detail", new=AsyncMock(side_effect=RuntimeError("feishu down")), ): resp = await client.get( f"/api/v1/approval/admin/requests/{row.id}", headers=admin_headers, ) assert resp.status_code == 200 body = resp.json()["data"] assert body["timeline_error"] == "feishu down" timeline = body["timeline"] assert len(timeline) == 1 assert timeline[0]["kind"] == "task" assert timeline[0]["user_name"] == "Cached Approver" @pytest.mark.asyncio async def test_detail_skips_feishu_fetch_when_refresh_false( self, client: AsyncClient, db_session: AsyncSession, admin_headers: dict, employee_user, ): emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-D-NOREFRESH", instance_code="feishu_admin_inst_noref", ) fetcher = AsyncMock() with patch( "app.approval.service.FeishuApprovalClient.get_instance_detail", new=fetcher, ): resp = await client.get( f"/api/v1/approval/admin/requests/{row.id}", headers=admin_headers, params={"refresh": "false"}, ) assert resp.status_code == 200 # The HTTP layer must not have hit Feishu at all. fetcher.assert_not_called() @pytest.mark.asyncio async def test_detail_404_for_unknown_id( self, client: AsyncClient, admin_headers: dict, ): import uuid as _uuid resp = await client.get( f"/api/v1/approval/admin/requests/{_uuid.uuid4()}", headers=admin_headers, ) assert resp.status_code == 404 @pytest.mark.asyncio async def test_detail_includes_progress_summary( self, client: AsyncClient, db_session: AsyncSession, admin_headers: dict, employee_user, ): """Multi-node Feishu task_list → ``progress`` field with the right total / completed / current_node_index. Two nodes: 1. Manager — APPROVED (terminal) 2. Director — PENDING (current) Expected: total=2, completed=1, current_node_index=2, current_node_name="Director Review", current_approvers=["Director Li"]. """ emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-D-PROG-1", instance_code="feishu_admin_inst_prog1", ) with patch( "app.approval.service.FeishuApprovalClient.get_instance_detail", new=AsyncMock( return_value={ "instance_code": "feishu_admin_inst_prog1", "status": "PENDING", "form": [], "task_list": [ { "node_id": "node_mgr", "node_name": "Manager Review", "user_id": "USR_M", "user_name": "Manager Mei", "status": "APPROVED", "end_time": "1714400000000", }, { "node_id": "node_dir", "node_name": "Director Review", "user_id": "USR_D", "user_name": "Director Li", "status": "PENDING", "start_time": "1714400500000", }, ], "comment_list": [], }, ), ): resp = await client.get( f"/api/v1/approval/admin/requests/{row.id}", headers=admin_headers, ) assert resp.status_code == 200 progress = resp.json()["data"]["progress"] assert progress is not None assert progress["total_nodes"] == 2 assert progress["completed_nodes"] == 1 assert progress["current_node_index"] == 2 assert progress["current_node_name"] == "Director Review" assert progress["current_approvers"] == ["Director Li"] assert progress["is_finished"] is False # Pinned in both branches so a regression that silently # flips the sync_lag flag is caught. assert progress["sync_lag"] is False assert len(progress["nodes"]) == 2 assert progress["nodes"][0]["status"] == "APPROVED" assert progress["nodes"][1]["status"] == "PENDING" # ``node_id`` surfaced 2026-05-15 so the frontend timeline can # bucket entries by stable id instead of by ``name`` (legacy # templates legitimately have two same-named nodes that # collapsed all entries into the first match under the old # name-based bucketing). Mirrors what the mocked task_list # supplied above; if a future refactor drops the field both # cases fail. assert progress["nodes"][0]["node_id"] == "node_mgr" assert progress["nodes"][1]["node_id"] == "node_dir" @pytest.mark.asyncio async def test_detail_accessible_to_asset_admin( self, client: AsyncClient, db_session: AsyncSession, asset_admin_headers: dict, employee_user, ): """``/admin/requests/{id}`` MUST be reachable by ``asset_admin``, not just ``global_admin``. The frontend ``EmployeeRequestPage`` at ``/admin/assets/{employee,software}- requests`` is route-guarded for ``[global_admin, asset_admin]`` (``web/src/App.tsx:405``) — locking this endpoint to global_admin only (as it was pre-2026-05-15) created a 403 regression when P0547's fix routed asset_admin clicks through the admin detail endpoint instead of the owner-scoped one. """ emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-AA-1", instance_code=None, # no Feishu fetch needed for this gate test ) resp = await client.get( f"/api/v1/approval/admin/requests/{row.id}", headers=asset_admin_headers, ) assert resp.status_code == 200, resp.text assert resp.json()["data"]["request_number"] == "REQ-AA-1" @pytest.mark.asyncio async def test_detail_still_forbidden_to_plain_employee( self, client: AsyncClient, db_session: AsyncSession, employee_user, employee_headers: dict, ): """Loosening the gate from ``require_admin`` (global_admin only) to ``require_asset_admin`` (global_admin OR asset_admin) MUST NOT widen access to plain employees — they still get 403, and the owner-scoped ``/my/requests/{id}`` endpoint is the only surface they have for fetching their own request detail. """ emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-AA-2", instance_code=None, ) resp = await client.get( f"/api/v1/approval/admin/requests/{row.id}", headers=employee_headers, ) assert resp.status_code == 403 @pytest.mark.asyncio async def test_detail_progress_collapses_or_node_approvers( self, client: AsyncClient, db_session: AsyncSession, admin_headers: dict, employee_user, ): """Three tasks on the same node_id (OR-mode approvers) collapse to one progress node with three approvers — common Feishu pattern for "any one of N can approve" steps. """ emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-D-PROG-2", instance_code="feishu_admin_inst_prog2", ) with patch( "app.approval.service.FeishuApprovalClient.get_instance_detail", new=AsyncMock( return_value={ "instance_code": "feishu_admin_inst_prog2", "status": "PENDING", "form": [], "task_list": [ { "node_id": "node_it", "node_name": "IT Review", "user_name": "Alice", "status": "PENDING", "start_time": "1714400000000", }, { "node_id": "node_it", "node_name": "IT Review", "user_name": "Bob", "status": "PENDING", "start_time": "1714400000000", }, { "node_id": "node_it", "node_name": "IT Review", "user_name": "Carol", "status": "PENDING", "start_time": "1714400000000", }, ], "comment_list": [], }, ), ): resp = await client.get( f"/api/v1/approval/admin/requests/{row.id}", headers=admin_headers, ) assert resp.status_code == 200 progress = resp.json()["data"]["progress"] assert progress["total_nodes"] == 1 assert progress["completed_nodes"] == 0 assert progress["current_node_index"] == 1 # Explicit count assertion — a dedup bug that produced # duplicates would fail this even if the names happened to # round-trip. assert len(progress["current_approvers"]) == 3 assert sorted(progress["current_approvers"]) == ["Alice", "Bob", "Carol"] @pytest.mark.asyncio async def test_detail_progress_or_node_first_approver_wins( self, client: AsyncClient, db_session: AsyncSession, admin_headers: dict, employee_user, ): """OR-mode node where Alice APPROVED and Bob's task got marked REVOKED (Feishu's "loser" status when another approver acted first). The collapsed status must be APPROVED — without the skipped_statuses filter the code would either fall through to REJECTED (the original bug) or render the wrong colour. """ emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-D-PROG-OR-WIN", instance_code="feishu_admin_inst_or_win", ) with patch( "app.approval.service.FeishuApprovalClient.get_instance_detail", new=AsyncMock( return_value={ "instance_code": "feishu_admin_inst_or_win", "status": "APPROVED", "form": [], "task_list": [ { "node_id": "node_or", "node_name": "Either-Or Approval", "user_name": "Alice", "status": "APPROVED", "end_time": "1714400000000", }, { "node_id": "node_or", "node_name": "Either-Or Approval", "user_name": "Bob", "status": "REVOKED", "end_time": "1714400000000", }, ], "comment_list": [], }, ), ): resp = await client.get( f"/api/v1/approval/admin/requests/{row.id}", headers=admin_headers, ) assert resp.status_code == 200 progress = resp.json()["data"]["progress"] assert progress["total_nodes"] == 1 assert progress["nodes"][0]["status"] == "APPROVED" assert progress["completed_nodes"] == 1 assert progress["is_finished"] is True assert progress["sync_lag"] is False @pytest.mark.asyncio async def test_detail_progress_dedup_user_id_and_user_name( self, client: AsyncClient, db_session: AsyncSession, admin_headers: dict, employee_user, ): """Feishu sometimes emits two task entries for one human at one node — one with ``user_id`` only and one with ``user_name`` only (rare but observed). The progress builder must NOT surface the same person twice with different labels. """ emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-D-PROG-DEDUP", instance_code="feishu_admin_inst_dedup", ) with patch( "app.approval.service.FeishuApprovalClient.get_instance_detail", new=AsyncMock( return_value={ "instance_code": "feishu_admin_inst_dedup", "status": "PENDING", "form": [], "task_list": [ { "node_id": "node_mgr", "node_name": "Manager", "user_id": "USR_A", "user_name": "Alice", "status": "PENDING", "start_time": "1714400000000", }, { "node_id": "node_mgr", "node_name": "Manager", "user_id": "USR_A", "user_name": "Alice", "status": "PENDING", "start_time": "1714400000000", }, ], "comment_list": [], }, ), ): resp = await client.get( f"/api/v1/approval/admin/requests/{row.id}", headers=admin_headers, ) assert resp.status_code == 200 progress = resp.json()["data"]["progress"] assert progress["current_approvers"] == ["Alice"] @pytest.mark.asyncio async def test_detail_progress_dedup_full_row_then_partial( self, client: AsyncClient, db_session: AsyncSession, admin_headers: dict, employee_user, ): """Multiple task rows for the same person at the same node collapse to one entry. uid-primary dedup: when both rows carry the same ``user_id``, the second is treated as a duplicate regardless of whether it also has a name. """ emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-D-PROG-DEDUP-FULL", instance_code="feishu_admin_inst_dedup_full", ) with patch( "app.approval.service.FeishuApprovalClient.get_instance_detail", new=AsyncMock( return_value={ "instance_code": "feishu_admin_inst_dedup_full", "status": "PENDING", "form": [], "task_list": [ # First row: both uid and uname. Establishes both. { "node_id": "n1", "user_id": "USR_A", "user_name": "Alice", "status": "PENDING", "start_time": "1714400000000", }, # Second row: only uid → must match via uid set. { "node_id": "n1", "user_id": "USR_A", "status": "PENDING", "start_time": "1714400000000", }, # Third row: only uname → must match via uname set. { "node_id": "n1", "user_name": "Alice", "status": "PENDING", "start_time": "1714400000000", }, ], "comment_list": [], }, ), ): resp = await client.get( f"/api/v1/approval/admin/requests/{row.id}", headers=admin_headers, ) assert resp.status_code == 200 progress = resp.json()["data"]["progress"] # All three rows reference the same person — only one entry. assert progress["current_approvers"] == ["Alice"] @pytest.mark.asyncio async def test_detail_progress_dedup_same_name_different_users_not_collapsed( self, client: AsyncClient, db_session: AsyncSession, admin_headers: dict, employee_user, ): """Round-3 regression: two DIFFERENT humans sharing a display name at the same OR-mode node (common in Chinese orgs with frequent "张伟" / "李娜" duplicates) MUST surface as two entries. A previous attempt at "dedup on either uid OR uname" silently dropped the second person — an admin would have thought only one approver was required when in reality two distinct people needed to act. """ emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-D-PROG-SAMENAME", instance_code="feishu_admin_inst_samename", ) with patch( "app.approval.service.FeishuApprovalClient.get_instance_detail", new=AsyncMock( return_value={ "instance_code": "feishu_admin_inst_samename", "status": "PENDING", "form": [], "task_list": [ { "node_id": "n1", "user_id": "USR_A", "user_name": "张伟", "status": "PENDING", "start_time": "1714400000000", }, { "node_id": "n1", "user_id": "USR_B", "user_name": "张伟", "status": "PENDING", "start_time": "1714400000000", }, ], "comment_list": [], }, ), ): resp = await client.get( f"/api/v1/approval/admin/requests/{row.id}", headers=admin_headers, ) assert resp.status_code == 200 progress = resp.json()["data"]["progress"] # Two different humans, both surfaced — accurate (if a # bit ambiguous to the eye, but Feishu's own UI shows the # same). assert len(progress["current_approvers"]) == 2 assert progress["current_approvers"] == ["张伟", "张伟"] @pytest.mark.asyncio async def test_detail_progress_dedup_uid_only_vs_uname_only( self, client: AsyncClient, db_session: AsyncSession, admin_headers: dict, employee_user, ): """**PIN test — documents a fundamentally-unresolvable case.** A row with only ``user_id="USR_A"`` and a separate row with only ``user_name="Alice"`` for the SAME person cannot be collapsed without an external user-id↔display-name map (we don't query Feishu for that — it would 2x the API cost on every drawer click). Current dedup surfaces them as TWO entries. **If a future change introduces a user-id resolver (e.g. a Helix-side cache of Feishu user metadata) and is able to collapse this case, REPLACE the ``len == 2`` assertion with ``len == 1``** — don't dance around the assertion. Until then, this test pins the current behaviour so a regression that silently changes the count is caught. """ emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-D-PROG-DEDUP-SPLIT", instance_code="feishu_admin_inst_dedup_split", ) # Feed both directions to pin both the (uid-then-uname) and # (uname-then-uid) ordering — without symmetric checks the # dedup could regress on one direction only. with patch( "app.approval.service.FeishuApprovalClient.get_instance_detail", new=AsyncMock( return_value={ "instance_code": "feishu_admin_inst_dedup_split", "status": "PENDING", "form": [], "task_list": [ # uid-only first; uname-only second (same person) { "node_id": "n1", "user_id": "USR_A", "status": "PENDING", "start_time": "1714400000000", }, { "node_id": "n1", "user_name": "Alice", "status": "PENDING", "start_time": "1714400000000", }, ], "comment_list": [], }, ), ): resp = await client.get( f"/api/v1/approval/admin/requests/{row.id}", headers=admin_headers, ) assert resp.status_code == 200 progress = resp.json()["data"]["progress"] # Round-3 uid-primary semantics: row 1 has only uid, so # it lands on the uid path with no uname to record. Row 2 # has only uname, lands on the uname path, and finds # nothing in ``seen_unames`` (the uid path only records # uname when one accompanied the uid). Both rows are # surfaced — the case is fundamentally unresolvable # without an external user-id ↔ display-name map. Two # entries with labels in row order: ``["USR_A", "Alice"]``. assert sorted(progress["current_approvers"]) == sorted(["USR_A", "Alice"]) assert len(progress["current_approvers"]) == 2 @pytest.mark.asyncio async def test_detail_progress_none_status_keeps_not_finished( self, client: AsyncClient, db_session: AsyncSession, admin_headers: dict, employee_user, ): """A node whose tasks have ``status: null`` (Feishu CC / observer nodes) must NOT be silently treated as terminal. The flow stays ``is_finished=False`` until the indeterminate node resolves. """ emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-D-PROG-NULL-STAT", instance_code="feishu_admin_inst_nullstat", ) with patch( "app.approval.service.FeishuApprovalClient.get_instance_detail", new=AsyncMock( return_value={ "instance_code": "feishu_admin_inst_nullstat", # Instance says APPROVED but a CC node still has # null status — the drawer must defer to the # explicit instance signal in this case. "status": "APPROVED", "form": [], "task_list": [ { "node_id": "n1", "node_name": "Manager", "user_name": "Mei", "status": "APPROVED", "end_time": "1714400000000", }, { "node_id": "n2", "node_name": "CC", "user_name": "CC Person", "status": None, }, ], "comment_list": [], }, ), ): resp = await client.get( f"/api/v1/approval/admin/requests/{row.id}", headers=admin_headers, ) assert resp.status_code == 200 progress = resp.json()["data"]["progress"] # Instance-says-finished wins — the drawer trusts Feishu's # explicit APPROVED here. assert progress["is_finished"] is True @pytest.mark.asyncio async def test_detail_progress_none_status_no_instance_signal_stays_pending( self, client: AsyncClient, db_session: AsyncSession, admin_headers: dict, employee_user, ): """Same shape as above but instance status is also PENDING — ``nodes_say_finished`` must NOT shortcut to True just because the null-status node was filtered out of the all-check. """ emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-D-PROG-NULL-PEND", instance_code="feishu_admin_inst_null_pend", ) with patch( "app.approval.service.FeishuApprovalClient.get_instance_detail", new=AsyncMock( return_value={ "instance_code": "feishu_admin_inst_null_pend", "status": "PENDING", "form": [], "task_list": [ { "node_id": "n1", "user_name": "Mei", "status": "APPROVED", "end_time": "1714400000000", }, { "node_id": "n2", "user_name": "CC", "status": None, }, ], "comment_list": [], }, ), ): resp = await client.get( f"/api/v1/approval/admin/requests/{row.id}", headers=admin_headers, ) assert resp.status_code == 200 progress = resp.json()["data"]["progress"] assert progress["is_finished"] is False # Sync-lag UX: current_node_index points past the last # completed node so the Steps strip doesn't highlight node # 1 as in-progress while it's already done. The explicit # sync_lag flag is the frontend's signal for the "等待飞书 # 状态同步" hint — pin it here so a regression that drops # the flag is caught. assert progress["current_node_index"] == 2 assert progress["sync_lag"] is True @pytest.mark.asyncio async def test_detail_non_dict_live_response_falls_back_to_cache( self, client: AsyncClient, db_session: AsyncSession, admin_headers: dict, employee_user, ): """A malformed Feishu response (non-dict — e.g. partial parse, unexpected primitive) must NOT zero-out the timeline while still hydrating progress from cache. Round-2 fix: treat as a fetch failure, fall back to cached snapshot for both, surface ``timeline_error``. """ emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-D-NONDICT", instance_code="feishu_admin_inst_nondict", ) # Pre-populate cache so the fallback has something to render. row.detail = { "feishu": { "approval_name": "缓存模板", "instance_code": "feishu_admin_inst_nondict", "instance_status": "PENDING", "latest_instance_detail": { "instance_code": "feishu_admin_inst_nondict", "status": "PENDING", "task_list": [ { "node_id": "n1", "user_name": "Cached Manager", "status": "PENDING", "start_time": 1714000000000, }, ], "comment_list": [], }, }, } db_session.add(row) await db_session.commit() # Feishu returns a string instead of a dict — a real failure # mode seen when Feishu's body parser short-circuits on a # 200 with empty body. with patch( "app.approval.service.FeishuApprovalClient.get_instance_detail", new=AsyncMock(return_value="bad payload"), ): resp = await client.get( f"/api/v1/approval/admin/requests/{row.id}", headers=admin_headers, ) assert resp.status_code == 200 body = resp.json()["data"] # Surfaces an error AND uses cache for both timeline and # progress — no asymmetry. assert body["timeline_error"] is not None assert "non-dict" in body["timeline_error"] assert len(body["timeline"]) == 1 assert body["timeline"][0]["user_name"] == "Cached Manager" assert body["progress"] is not None assert body["progress"]["current_node_index"] == 1 assert body["progress"]["current_approvers"] == ["Cached Manager"] @pytest.mark.asyncio async def test_detail_progress_transferred_treated_as_pending( self, client: AsyncClient, db_session: AsyncSession, admin_headers: dict, employee_user, ): """``TRANSFERRED`` is Feishu's "approver delegated this task to someone else" status — semantically still waiting on a human. ``current_node_index`` should point at the TRANSFERRED node so the drawer doesn't render "已完结". """ emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-D-PROG-XFER", instance_code="feishu_admin_inst_xfer", ) with patch( "app.approval.service.FeishuApprovalClient.get_instance_detail", new=AsyncMock( return_value={ "instance_code": "feishu_admin_inst_xfer", "status": "PENDING", "form": [], "task_list": [ { "node_id": "n1", "node_name": "Manager", "user_name": "Mei", "status": "APPROVED", "end_time": "1714400000000", }, { "node_id": "n2", "node_name": "Director", "user_name": "Li", "status": "TRANSFERRED", "start_time": "1714400500000", }, ], "comment_list": [], }, ), ): resp = await client.get( f"/api/v1/approval/admin/requests/{row.id}", headers=admin_headers, ) assert resp.status_code == 200 progress = resp.json()["data"]["progress"] assert progress["is_finished"] is False assert progress["current_node_index"] == 2 assert progress["nodes"][1]["status"] == "PENDING" @pytest.mark.asyncio async def test_detail_progress_pending_instance_keeps_not_finished( self, client: AsyncClient, db_session: AsyncSession, admin_headers: dict, employee_user, ): """Edge case: every node collapsed to APPROVED but Feishu's ``instance.status`` still says PENDING (propagation lag — Feishu marks instance APPROVED a beat after the final task flips). Drawer must NOT prematurely say "已完结". """ emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-D-PROG-LAG", instance_code="feishu_admin_inst_lag", ) with patch( "app.approval.service.FeishuApprovalClient.get_instance_detail", new=AsyncMock( return_value={ "instance_code": "feishu_admin_inst_lag", "status": "PENDING", "form": [], "task_list": [ { "node_id": "n1", "node_name": "Manager", "user_name": "Mei", "status": "APPROVED", "end_time": "1714400000000", }, ], "comment_list": [], }, ), ): resp = await client.get( f"/api/v1/approval/admin/requests/{row.id}", headers=admin_headers, ) assert resp.status_code == 200 progress = resp.json()["data"]["progress"] assert progress["is_finished"] is False # Sync-lag UX (round-2 fix, round-3 made the signal an # explicit flag): no PENDING node and flow not finished → # ``current_node_index`` points past the last completed # node AND ``sync_lag=True`` so the frontend can render # the "syncing" summary line instead of highlighting node # 1 as in-progress. Pinned explicitly because the prior # round-2 attempt at inferring this from null names was a # known regression source. assert progress["current_node_index"] == 1 assert progress["current_node_name"] is None assert progress["current_approvers"] == [] assert progress["sync_lag"] is True assert progress["nodes"][0]["status"] == "APPROVED" @pytest.mark.asyncio async def test_detail_persists_refreshed_snapshot_back( self, client: AsyncClient, db_session: AsyncSession, admin_headers: dict, employee_user, ): """A successful live refresh persists the snapshot back to ``row.detail.feishu.latest_instance_detail`` so subsequent callers (admin click + /my/requests + callback handlers) see the same data. Regression test for the ``await db.flush()`` + router-level ``commit()`` contract. """ emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-D-PERSIST", instance_code="feishu_admin_inst_persist", ) with patch( "app.approval.service.FeishuApprovalClient.get_instance_detail", new=AsyncMock( return_value={ "instance_code": "feishu_admin_inst_persist", "status": "APPROVED", "form": [], "task_list": [ { "node_id": "n1", "user_name": "Manager", "status": "APPROVED", "end_time": "1714400000000", }, ], "comment_list": [], }, ), ): resp = await client.get( f"/api/v1/approval/admin/requests/{row.id}", headers=admin_headers, ) assert resp.status_code == 200 # Verify the persist landed by issuing a fresh detail call # with ``?refresh=false`` and patching the Feishu client to # blow up if it gets called — if the cache-write didn't # land, the no-refresh path would short-circuit but the # progress + status would mismatch what the first call # reported. We use ``AsyncMock`` and ``assert_not_called`` # rather than ``side_effect=AssertionError`` because the # service's bare ``except Exception`` would swallow the # AssertionError and surface it as ``timeline_error``, # making the test pass for the wrong reason. no_refresh_mock = AsyncMock() with patch( "app.approval.service.FeishuApprovalClient.get_instance_detail", new=no_refresh_mock, ): recheck = await client.get( f"/api/v1/approval/admin/requests/{row.id}?refresh=false", headers=admin_headers, ) no_refresh_mock.assert_not_called() assert recheck.status_code == 200 body = recheck.json()["data"] # No timeline_error on the cache-only path (any error here # would mean Feishu was contacted — see mock assertion). assert body["timeline_error"] is None # The persisted snapshot drives both the cached timeline AND # the instance status — confirms both fields landed. assert body["instance_status_raw"] == "APPROVED" assert len(body["timeline"]) == 1 assert body["timeline"][0]["status"] == "APPROVED" assert body["progress"] is not None assert body["progress"]["nodes"][0]["status"] == "APPROVED" @pytest.mark.asyncio async def test_detail_refresh_false_uses_cached_timeline_and_progress( self, client: AsyncClient, db_session: AsyncSession, admin_headers: dict, employee_user, ): """``?refresh=false`` MUST surface BOTH timeline and progress from the cached snapshot — not an asymmetric "progress hydrated but timeline empty" rendering. """ emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-D-NOREF-PROG", instance_code="feishu_admin_inst_noref_prog", ) row.detail = { "feishu": { "approval_name": "缓存模板", "instance_code": "feishu_admin_inst_noref_prog", "instance_status": "PENDING", "latest_instance_detail": { "instance_code": "feishu_admin_inst_noref_prog", "status": "PENDING", "task_list": [ { "node_id": "n1", "node_name": "Manager", "user_name": "Cached Manager", "status": "PENDING", "start_time": 1714000000000, }, ], "comment_list": [], }, }, } db_session.add(row) await db_session.commit() # ``assert_not_called`` proves the refresh path was skipped. # A ``side_effect=AssertionError`` mock would be caught by # the service's bare ``except Exception`` and surface as # ``timeline_error`` while the cache fallback still # populates timeline+progress with values that satisfy the # other assertions — the test would pass for the wrong # reason. no_refresh_mock = AsyncMock() with patch( "app.approval.service.FeishuApprovalClient.get_instance_detail", new=no_refresh_mock, ): resp = await client.get( f"/api/v1/approval/admin/requests/{row.id}?refresh=false", headers=admin_headers, ) no_refresh_mock.assert_not_called() assert resp.status_code == 200 body = resp.json()["data"] assert body["timeline_error"] is None assert len(body["timeline"]) == 1 assert body["timeline"][0]["user_name"] == "Cached Manager" assert body["progress"] is not None assert body["progress"]["current_node_index"] == 1 @pytest.mark.asyncio async def test_detail_progress_finished_when_all_terminal( self, client: AsyncClient, db_session: AsyncSession, admin_headers: dict, employee_user, ): emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-D-PROG-DONE", instance_code="feishu_admin_inst_done", ) with patch( "app.approval.service.FeishuApprovalClient.get_instance_detail", new=AsyncMock( return_value={ "instance_code": "feishu_admin_inst_done", "status": "APPROVED", "form": [], "task_list": [ { "node_id": "n1", "node_name": "Step 1", "user_name": "A", "status": "APPROVED", "end_time": "1714400000000", }, { "node_id": "n2", "node_name": "Step 2", "user_name": "B", "status": "APPROVED", "end_time": "1714400600000", }, ], "comment_list": [], }, ), ): resp = await client.get( f"/api/v1/approval/admin/requests/{row.id}", headers=admin_headers, ) assert resp.status_code == 200 progress = resp.json()["data"]["progress"] assert progress["is_finished"] is True assert progress["current_node_index"] is None assert progress["current_approvers"] == [] assert progress["completed_nodes"] == 2 @pytest.mark.asyncio async def test_detail_progress_none_when_no_task_list( self, client: AsyncClient, db_session: AsyncSession, admin_headers: dict, employee_user, ): """A request with no Feishu instance yet (template not configured / instance creation pending) yields ``progress=None`` so the frontend hides the Steps indicator. """ emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-D-PROG-NONE", instance_code=None, ) resp = await client.get( f"/api/v1/approval/admin/requests/{row.id}", headers=admin_headers, ) assert resp.status_code == 200 assert resp.json()["data"]["progress"] is None class TestMyApprovalDetailEndpoint: """Owner-scoped twin of the admin detail endpoint. Backs the "我的申请" detail drawer — same enrichment (timeline + progress + form snapshot) but the route refuses to leak other employees' rows. """ @pytest.mark.asyncio async def test_returns_own_request( self, client: AsyncClient, db_session: AsyncSession, employee_user, employee_headers: dict, ): emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-MINE-1", instance_code="feishu_my_1", ) with patch( "app.approval.service.FeishuApprovalClient.get_instance_detail", new=AsyncMock( return_value={ "instance_code": "feishu_my_1", "status": "PENDING", "form": [], "task_list": [ { "node_id": "n1", "node_name": "Manager", "user_name": "Manager Mei", "status": "PENDING", "start_time": "1714400000000", }, ], "comment_list": [], }, ), ): resp = await client.get( f"/api/v1/approval/my/requests/{row.id}", headers=employee_headers, ) assert resp.status_code == 200 body = resp.json()["data"] assert body["request_number"] == "REQ-MINE-1" assert body["progress"]["current_node_index"] == 1 assert body["progress"]["current_node_name"] == "Manager" assert len(body["timeline"]) == 1 @pytest.mark.asyncio async def test_404_for_other_employees_request( self, client: AsyncClient, db_session: AsyncSession, employee_headers: dict, ): """Caller's id != requester_id → 404, not 403. Uses a SEPARATE employee as the row's owner (not the admin fixture) so the test specifically pins that the gate is ``requester_id == current_user.id``, not role-based. If a future regression flipped the check to "only admins see non-self rows", that bug would still let one employee see another employee's request — this test catches it where the admin-as-other-user version did not. 404 (rather than 403) means the endpoint cannot be used as an existence oracle for someone else's request ids. """ import uuid as _uuid from app.auth.models import AuthUserRole from app.identity.person_models import IdentityPerson from tests.conftest import _get_role_id other_id = _uuid.uuid4() other = IdentityPerson( id=other_id, work_email="other-employee@test.com", display_name="Other Employee", employment_status="active", source_system="successfactors", hris_person_id=f"sf-other-{other_id}", ) db_session.add(other) await db_session.flush() role_id = await _get_role_id(db_session, "employee") db_session.add(AuthUserRole(user_id=other_id, role_id=role_id)) await db_session.commit() row = await _seed_request( db_session, requester_id=other_id, request_number="REQ-NOT-MINE", instance_code=None, ) resp = await client.get( f"/api/v1/approval/my/requests/{row.id}", headers=employee_headers, ) assert resp.status_code == 404 @pytest.mark.asyncio async def test_404_for_unknown_id( self, client: AsyncClient, employee_headers: dict, ): import uuid as _uuid resp = await client.get( f"/api/v1/approval/my/requests/{_uuid.uuid4()}", headers=employee_headers, ) assert resp.status_code == 404 @pytest.mark.asyncio async def test_my_request_detail_honours_refresh_false( self, client: AsyncClient, db_session: AsyncSession, employee_user, employee_headers: dict, ): """``?refresh=false`` on the employee endpoint must read the cached snapshot only — no Feishu round-trip. """ emp_user, _ = employee_user row = await _seed_request( db_session, requester_id=emp_user.id, request_number="REQ-MINE-NOREF", instance_code="feishu_my_noref", ) row.detail = { "feishu": { "approval_name": "缓存模板", "instance_code": "feishu_my_noref", "instance_status": "PENDING", "latest_instance_detail": { "instance_code": "feishu_my_noref", "status": "PENDING", "task_list": [ { "node_id": "n1", "node_name": "Manager", "user_name": "Cached Manager", "status": "PENDING", "start_time": 1714000000000, }, ], "comment_list": [], }, }, } db_session.add(row) await db_session.commit() # ``assert_not_called`` proves the refresh path was skipped # (a side_effect=AssertionError mock would be swallowed by # the service's bare ``except Exception`` and the cache # fallback would still populate the response — see the # admin endpoint's matching test for the same rationale). no_refresh_mock = AsyncMock() with patch( "app.approval.service.FeishuApprovalClient.get_instance_detail", new=no_refresh_mock, ): resp = await client.get( f"/api/v1/approval/my/requests/{row.id}?refresh=false", headers=employee_headers, ) no_refresh_mock.assert_not_called() assert resp.status_code == 200 body = resp.json()["data"] assert body["timeline_error"] is None assert body["progress"] is not None assert len(body["timeline"]) == 1