"""/api/v1/me/requests contract tests — the unified '我的申请' tracker. Asserts: * Anonymous → 401 * No submitted requests → 200 with empty rows (not 403) * Aggregates across source tables (employee_requests + vpn + announcement + domain_registration), normalising each source's status vocabulary * Only the CALLER's own requests appear (another person's are excluded) * Pending items expose can_cancel + cancel_path; terminal ones do not * ``status=open`` and ``category`` filters narrow correctly while counts_* always reflect the unfiltered set """ from __future__ import annotations import uuid from datetime import UTC, datetime import pytest from app.domains.announcement_request.models import AnnouncementRequest from app.domains.asset.models import EmployeeRequest from app.domains.domain_registration.models import DomainRegistration from app.domains.vpn_application.models import VpnApplication from app.identity.person_models import IdentityPerson def _emp_request(requester_id, request_type: str, status: str) -> EmployeeRequest: return EmployeeRequest( id=uuid.uuid4(), request_number=f"REQ-{uuid.uuid4().hex[:10]}", request_type=request_type, status=status, requester_id=requester_id, created_at=datetime.now(UTC), ) # ── RBAC ───────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_no_requests_returns_empty_200(client, employee_user): _, token = employee_user resp = await client.get( "/api/v1/me/requests", headers={"Authorization": f"Bearer {token}"} ) assert resp.status_code == 200 body = resp.json() assert body["success"] is True assert body["data"]["total"] == 0 assert body["data"]["items"] == [] # ── Cross-source aggregation ───────────────────────────────────── @pytest.mark.asyncio async def test_aggregates_across_sources_self_scoped( client, employee_user, db_session, ): me, token = employee_user # Another person with a request that must NOT leak into my tracker. other = IdentityPerson( id=uuid.uuid4(), work_email=f"{uuid.uuid4().hex[:8]}@test.local", display_name="Other", employment_status="active", source_system="test", hris_person_id=f"sf-{uuid.uuid4().hex[:6]}", ) db_session.add(other) await db_session.flush() db_session.add(_emp_request(other.id, "laptop_replacement", "pending")) # My requests across four source tables, four lifecycle states. db_session.add(_emp_request(me.id, "laptop_replacement", "pending")) db_session.add(VpnApplication( person_id=str(me.id), vpn_node="tokyo-gw", usage_description="dev access", resource_checklist="git, ci", status="pending", )) db_session.add(AnnouncementRequest( person_id=str(me.id), description="All-hands notice", # approved is the TERMINAL success state for announcement (no later # active/provisioned state exists), so the tracker normalises it to # completed — NOT a perpetual "处理中". Regression guard for the five # sibling source domains #1108 left half-fixed # (sso/vpn/ai_tool/announcement/domain). status="approved", # → normalised completed (terminal source) )) db_session.add(DomainRegistration( person_id=str(me.id), domain_name="example.dev", purpose="marketing", status="granted", # → normalised completed )) await db_session.commit() resp = await client.get( "/api/v1/me/requests", headers={"Authorization": f"Bearer {token}"} ) assert resp.status_code == 200 data = resp.json()["data"] # Four mine, other person's one excluded. assert data["total"] == 4 sources = {it["source"] for it in data["items"]} assert sources == {"employee_request", "vpn", "announcement", "domain"} by_source = {it["source"]: it for it in data["items"]} # Pending employee request → cancellable with a concrete cancel_path. emp = by_source["employee_request"] assert emp["status"] == "pending" assert emp["is_open"] is True assert emp["can_cancel"] is True assert emp["cancel_path"].startswith("/api/v1/assets/employee-requests/") assert emp["deep_link"].startswith("/my/requests?focus=") # Status normalisation across vocabularies. approved is terminal for # the announcement source → completed (was wrongly in_progress before). assert by_source["announcement"]["status"] == "completed" assert by_source["announcement"]["raw_status"] == "approved" assert by_source["announcement"]["is_open"] is False assert by_source["domain"]["status"] == "completed" assert by_source["domain"]["raw_status"] == "granted" assert by_source["domain"]["is_open"] is False assert by_source["domain"]["can_cancel"] is False # Counts reflect every submitted request. assert data["counts_by_status"]["pending"] == 2 # emp + vpn assert data["counts_by_status"]["completed"] == 2 # announcement + domain assert data["counts_by_status"].get("in_progress", 0) == 0 assert data["counts_by_category"]["network"] == 2 # vpn + domain @pytest.mark.asyncio async def test_pending_employee_request_exposes_cached_current_approvers( client, employee_user, db_session, ): """Laptop and software requests show the active Feishu approvers inline. The tracker must use the callback-cached snapshot, not make a live Feishu request while listing a page of records. """ me, token = employee_user request = _emp_request(me.id, "software_seat", "pending") request.detail = { "feishu": { "instance_code": "FEISHU-REQ-001", "latest_instance_detail": { "task_list": [ { "node_id": "manager", "node_name": "Manager approval", "user_id": "ou-manager", "user_name": "Alice Manager", "status": "PENDING", }, { "node_id": "it", "node_name": "IT approval", "user_id": "ou-it", "user_name": "Bob IT", "status": "APPROVED", }, ], }, }, } db_session.add(request) await db_session.commit() resp = await client.get( "/api/v1/me/requests", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 item = resp.json()["data"]["items"][0] assert item["feishu_instance_code"] == "FEISHU-REQ-001" assert item["current_approvers"] == ["Alice Manager"] # ── Filters ────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_status_open_filter(client, employee_user, db_session): me, token = employee_user db_session.add(_emp_request(me.id, "software_seat", "pending")) db_session.add(_emp_request(me.id, "software_seat", "completed")) db_session.add(_emp_request(me.id, "software_seat", "rejected")) await db_session.commit() resp = await client.get( "/api/v1/me/requests?status=open", headers={"Authorization": f"Bearer {token}"}, ) data = resp.json()["data"] assert data["total"] == 1 assert all(it["is_open"] for it in data["items"]) # Counts stay full despite the filter. assert data["counts_by_status"]["completed"] == 1 assert data["counts_by_status"]["rejected"] == 1 @pytest.mark.asyncio async def test_category_filter(client, employee_user, db_session): me, token = employee_user db_session.add(_emp_request(me.id, "laptop_replacement", "pending")) # hardware db_session.add(VpnApplication( person_id=str(me.id), vpn_node="sg-gw", usage_description="x", resource_checklist="y", status="pending", )) # network await db_session.commit() resp = await client.get( "/api/v1/me/requests?category=network", headers={"Authorization": f"Bearer {token}"}, ) data = resp.json()["data"] assert data["total"] == 1 assert data["items"][0]["category"] == "network" @pytest.mark.asyncio async def test_q_search_filter(client, employee_user, db_session): """``q`` matches title / subtitle / request_type case-insensitively. Powers the global-search 我的申请 group (PR 2026-06-10) — a DNS name, domain name, VPN node or request number typed in the header search must resolve to the submitted request. """ me, token = employee_user db_session.add(_emp_request(me.id, "laptop_replacement", "pending")) db_session.add(VpnApplication( person_id=str(me.id), vpn_node="Tokyo-GW", usage_description="dev", resource_checklist="git", status="pending", )) db_session.add(DomainRegistration( person_id=str(me.id), domain_name="searchable-example.dev", purpose="mkt", status="pending", )) await db_session.commit() headers = {"Authorization": f"Bearer {token}"} # Title match (domain name), case-insensitive. resp = await client.get("/api/v1/me/requests?q=SEARCHABLE-EX", headers=headers) data = resp.json()["data"] assert data["total"] == 1 assert data["items"][0]["source"] == "domain" # request_type match. resp = await client.get("/api/v1/me/requests?q=laptop_repl", headers=headers) data = resp.json()["data"] assert data["total"] == 1 assert data["items"][0]["source"] == "employee_request" # VPN node title match. resp = await client.get("/api/v1/me/requests?q=tokyo", headers=headers) data = resp.json()["data"] assert data["total"] == 1 assert data["items"][0]["source"] == "vpn" # No match → empty rows, but counts stay full (chips stay stable). resp = await client.get("/api/v1/me/requests?q=zzz-nomatch", headers=headers) data = resp.json()["data"] assert data["total"] == 0 assert data["counts_by_status"]["pending"] == 3 # ── employee_request: approved == done (regression 2026-06-14, widened # to the whole source 2026-08-26) ──────────────────────────────────── # ``employee_requests`` has no post-approval state — every writer sets # pending / approved / rejected / cancelled, and the ``google_*`` / # ``onboarding_claim`` families are stamped ``completed`` directly # without ever passing through ``approved``. So the default # approved→in_progress map pinned finished requests as a perpetual # "处理中": first the granted software seats (fixed 2026-06-14 with a # per-type allowlist), then — still — every hardware request, which the # allowlist deliberately excluded on the theory that ``approved`` meant # "awaiting physical fulfilment". No such fulfilment event is ever # recorded, so those rows could never leave 进行中 (52 prod rows across # laptop_replacement / computer_borrow / asset_return / # offboarding_buyback when employees reported it). @pytest.mark.asyncio async def test_approved_employee_requests_normalise_to_completed( client, employee_user, db_session, ): me, token = employee_user db_session.add(_emp_request(me.id, "software_seat", "approved")) db_session.add(_emp_request(me.id, "api_key_request", "approved")) # Hardware: approval IS the last event the row records — it must read # as done, not as a forever-"处理中". db_session.add(_emp_request(me.id, "laptop_replacement", "approved")) db_session.add(_emp_request(me.id, "computer_borrow", "approved")) # A genuinely-open row keeps the tracker honest about the counts. db_session.add(_emp_request(me.id, "asset_return", "pending")) await db_session.commit() headers = {"Authorization": f"Bearer {token}"} data = (await client.get("/api/v1/me/requests", headers=headers)).json()["data"] by_type = {it["request_type"]: it for it in data["items"]} seat = by_type["software_seat"] assert seat["status"] == "completed" assert seat["raw_status"] == "approved" # display fidelity preserved assert seat["is_open"] is False key = by_type["api_key_request"] assert key["status"] == "completed" assert key["is_open"] is False for hardware_type in ("laptop_replacement", "computer_borrow"): row = by_type[hardware_type] assert row["status"] == "completed", hardware_type assert row["raw_status"] == "approved", hardware_type assert row["is_open"] is False, hardware_type # Counts reflect the remap: four completed, one still pending, zero of # these five counted as a bare "approved". assert data["counts_by_status"].get("completed") == 4 assert data["counts_by_status"].get("pending") == 1 assert data["counts_by_status"].get("in_progress") is None # ── Status resolver unit coverage ──────────────────────────────────── # Pure-function guards for the status normalisation rules (no DB/client), # so a regression in the lifecycle mapping fails fast and unambiguously. def test_resolve_normalized_status_approved_is_terminal_per_source(): from app.domains.me_universe.my_requests import ( SRC_AI_TOOL, SRC_ANNOUNCEMENT, SRC_CLOUD_SCOPE, SRC_DOMAIN, SRC_EMPLOYEE, SRC_SSO, SRC_VPN, resolve_normalized_status, ) # Five sibling domains: approved IS the terminal success state. for src in (SRC_SSO, SRC_VPN, SRC_AI_TOOL, SRC_ANNOUNCEMENT, SRC_DOMAIN): assert resolve_normalized_status(src, "x", "approved") == "completed" # employee_request: terminal for EVERY type — the table records no # event after the approval decision. for req_type in ( "software_seat", "api_key_request", "laptop_replacement", "computer_borrow", "asset_return", "offboarding_buyback", ): assert ( resolve_normalized_status(SRC_EMPLOYEE, req_type, "approved") == "completed" ), req_type # …but only for ``approved``: a still-pending row stays pending. assert resolve_normalized_status(SRC_EMPLOYEE, "laptop_replacement", "pending") == "pending" # cloud_scope: approved is an intermediate state (provisioning follows). assert resolve_normalized_status(SRC_CLOUD_SCOPE, "cloud_scope_request", "approved") == "in_progress" def test_dns_lifecycle_statuses_map_to_unified_states(): from app.domains.me_universe.my_requests import SRC_DNS, resolve_normalized_status # DNSPod write in flight vs written successfully — previously unmapped, # so an executed DNS change never rolled into the completed filter and # an in-flight one was wrongly treated as closed. assert resolve_normalized_status(SRC_DNS, "dns_create", "executing") == "in_progress" assert resolve_normalized_status(SRC_DNS, "dns_create", "executed") == "completed" # approved is intermediate for DNS (execution follows approval). assert resolve_normalized_status(SRC_DNS, "dns_create", "approved") == "in_progress" def test_cloud_grant_lifecycle_status_distinguishes_reject_from_revoke(): """A DENIED cloud access request (reject_grant flips PENDING→revoked + stamps detail.reject_reason) must read as 'rejected', not 'revoked' — otherwise 我的申请 implies access was granted then removed. Pure helper, shared by the list provider and the detail builder.""" from app.domains.me_universe.my_requests import ( ST_COMPLETED, ST_PENDING, ST_REJECTED, ST_REVOKED, cloud_grant_lifecycle_status, ) # denied request → rejected (reject_reason key present, only reject_grant writes it) assert cloud_grant_lifecycle_status("revoked", {"reject_reason": ""}) == ST_REJECTED # granted-then-revoked (no reject_reason) → revoked assert cloud_grant_lifecycle_status("revoked", {}) == ST_REVOKED assert cloud_grant_lifecycle_status("revoked", None) == ST_REVOKED # disabled/expired (absent from the shared map) → terminal revoked assert cloud_grant_lifecycle_status("disabled", None) == ST_REVOKED assert cloud_grant_lifecycle_status("expired", None) == ST_REVOKED # in-flight / live unchanged assert cloud_grant_lifecycle_status("pending", None) == ST_PENDING assert cloud_grant_lifecycle_status("active", None) == ST_COMPLETED