"""Integration tests for ``GET /api/v1/m/aggregate/inbox``. The action inbox is the spine of the mobile workspace — the one list that answers "what do I need to do?". It crosses domains (asset transfer / provisioning / approval / auditor finding / offboarding) and is strictly role-scoped, so the surface needs explicit guards against: - An employee seeing admin/auditor-only items (privacy + authz leak) - Self-scope items leaking from another user's queue - Severity ordering regressions (urgent must sort above info) - Per-type counts drifting from the surfaced items """ from __future__ import annotations import uuid from datetime import UTC, datetime import pytest from sqlalchemy.ext.asyncio import AsyncSession from app.approval.models import ApprovalRequest from app.domains.asset.models import ( AssetInventoryCycle, AssetInventoryItem, AssetItem, ) from app.domains.attendance.constants import ( STATUS_PENDING_INTERN, STATUS_PENDING_LEADER, ) from app.domains.attendance.models import AttendanceCycle, AttendanceReport from app.domains.auditor.models import ( AccessCertification, AccessCertificationItem, AuditorFinding, ) from app.domains.leaver_blocked.models import ( DUAL_SIGN_PENDING_SECONDARY, LeaverDualSignOverride, ) from app.domains.offboarding.models import ( RESO_PENDING, SEVERITY_BLOCKER, XFER_PENDING, AssetTransferRequest, OffboardingBlocker, OffboardingProcess, ) from app.domains.onboarding.models import ( PROC_SCANNING, OnboardingAckItem, OnboardingProcess, ) from app.domains.provisioning.models import STATUS_PENDING, ProvisioningTask from app.domains.support.models import SupportTicket def _auth(token: str) -> dict[str, str]: return {"Authorization": f"Bearer {token}"} async def _seed_person(db: AsyncSession) -> uuid.UUID: """Seed a throwaway active person and return its id (for FK targets like a provisioning task's other-party user_id / assignee_id).""" from app.identity.person_models import IdentityPerson pid = uuid.uuid4() db.add(IdentityPerson( id=pid, work_email=f"other-{pid.hex[:8]}@test.com", display_name="Other", employment_status="active", source_system="successfactors", hris_person_id=f"sf-{pid.hex[:8]}", )) await db.flush() return pid async def _seed_incoming_transfer( db: AsyncSession, recipient_id: uuid.UUID, label: str = "Test laptop", ) -> AssetTransferRequest: asset = AssetItem( asset_tag=f"TEST-{uuid.uuid4().hex[:8]}", asset_name=label, category="laptop", status="available", ) db.add(asset) await db.flush() transfer = AssetTransferRequest( initiator_id=recipient_id, recipient_id=recipient_id, asset_type="laptop", asset_ref_id=asset.id, asset_label=label, status=XFER_PENDING, ) db.add(transfer) await db.commit() return transfer @pytest.mark.asyncio async def test_inbox_empty_for_fresh_employee(client, employee_user) -> None: _, token = employee_user resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text data = resp.json()["data"] assert data["items"] == [] assert data["total"] == 0 assert data["counts"] == {} @pytest.mark.asyncio async def test_inbox_surfaces_incoming_asset_transfer( client, employee_user, db_session: AsyncSession, ) -> None: """An incoming transfer awaiting my accept/reject shows up as a self-scope item with inline confirm/reject actions and a deep link to the transfer detail page.""" user, token = employee_user transfer = await _seed_incoming_transfer(db_session, user.id, "MacBook Pro") resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text data = resp.json()["data"] assert data["total"] == 1 assert data["counts"] == {"asset_transfer_confirm": 1} item = data["items"][0] assert item["id"] == f"asset_transfer_confirm:{transfer.id}" assert item["type"] == "asset_transfer_confirm" assert item["title"] == "MacBook Pro" assert item["deep_link"] == f"/m/asset-transfer/{transfer.id}" action_keys = {a["key"] for a in item["actions"]} assert action_keys == {"confirm", "reject"} @pytest.mark.asyncio async def test_inbox_provisioning_task_for_assignee( client, employee_user, db_session: AsyncSession, ) -> None: """Provisioning tasks surface to the ASSIGNEE (the software owner / admin who fulfils them via complete/fail), not the subject — that's who the detail page lets act, so the inbox must match.""" user, token = employee_user subject_id = await _seed_person(db_session) task = ProvisioningTask( user_id=subject_id, # subject is someone else assignee_id=user.id, # the caller fulfils it action="grant", target_type="software", source="onboarding", execution_mode="manual", status=STATUS_PENDING, ) db_session.add(task) await db_session.commit() resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text data = resp.json()["data"] assert data["counts"].get("provisioning_ack") == 1 item = next(i for i in data["items"] if i["type"] == "provisioning_ack") assert item["deep_link"] == f"/m/provisioning-task/{task.id}" # No catalogued software → falls back to the target_type enum. assert item["title"] == "software" @pytest.mark.asyncio async def test_inbox_provisioning_subject_does_not_see_others_assigned_task( client, employee_user, db_session: AsyncSession, ) -> None: """Being the subject (user_id) of a task someone else is assigned to must NOT surface it — the subject can't act on it.""" user, token = employee_user other_id = await _seed_person(db_session) db_session.add(ProvisioningTask( user_id=user.id, # caller is the subject assignee_id=other_id, # someone else fulfils it action="grant", target_type="software", source="onboarding", execution_mode="manual", status=STATUS_PENDING, )) await db_session.commit() resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text assert resp.json()["data"]["total"] == 0 @pytest.mark.asyncio async def test_inbox_self_scope_does_not_leak_other_users( client, employee_user, db_session: AsyncSession, ) -> None: """A transfer addressed to someone else must never appear in my inbox — self-scope isolation.""" _, token = employee_user other_id = uuid.uuid4() from app.identity.person_models import IdentityPerson db_session.add(IdentityPerson( id=other_id, work_email=f"other-{other_id.hex[:8]}@test.com", display_name="Other", employment_status="active", source_system="successfactors", hris_person_id=f"sf-{other_id.hex[:8]}", )) await db_session.flush() await _seed_incoming_transfer(db_session, other_id) resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text assert resp.json()["data"]["total"] == 0 @pytest.mark.asyncio async def test_inbox_employee_never_sees_admin_or_auditor_items( client, employee_user, db_session: AsyncSession, ) -> None: """Pending approvals, open findings, and leaver blockers exist in the DB but an employee must see NONE of them — authz isolation.""" _, token = employee_user from datetime import date, datetime, timedelta now = datetime.now(UTC) db_session.add(ApprovalRequest( external_id=f"ext-{uuid.uuid4().hex[:8]}", approval_type="software_access", status="pending", )) db_session.add(AuditorFinding( control_key="mfa_coverage", finding_type="mfa_not_enrolled", dedup_key=f"d-{uuid.uuid4().hex[:8]}", subject_type="person", severity="high", title="Someone without MFA", status="open", first_seen_at=now, last_seen_at=now, )) proc = OffboardingProcess( person_id=uuid.uuid4(), leaver_date=date.today() + timedelta(days=3), status="blocked", ) # person_id FK → seed a person row first from app.identity.person_models import IdentityPerson leaver_id = uuid.uuid4() db_session.add(IdentityPerson( id=leaver_id, work_email=f"leaver-{leaver_id.hex[:8]}@test.com", display_name="Leaver", employment_status="active", source_system="successfactors", hris_person_id=f"sf-l-{leaver_id.hex[:8]}", )) await db_session.flush() proc.person_id = leaver_id db_session.add(proc) await db_session.flush() db_session.add(OffboardingBlocker( process_id=proc.id, resource_type="software_seat", resource_id=uuid.uuid4(), label="Jira seat", severity=SEVERITY_BLOCKER, resolution=RESO_PENDING, )) await db_session.commit() resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text data = resp.json()["data"] surfaced_types = {i["type"] for i in data["items"]} assert "approval" not in surfaced_types assert "compliance_finding" not in surfaced_types assert "leaver_blocker" not in surfaced_types assert data["total"] == 0 @pytest.mark.asyncio async def test_inbox_admin_does_not_see_company_wide_approvals( client, admin_user, db_session: AsyncSession, ) -> None: """Semantic-fix regression (2026-07): company-wide pending ``ApprovalRequest`` rows must NOT surface as a global_admin's personal to-dos. Helix keeps no per-approver attribution for Feishu-mirrored approvals (``ApprovalStep.approver_id`` is never populated; the approver acts in the Feishu app), so listing every pending approval as "my action" was noise the admin could neither own nor clear — the exact "why can I see everyone's cloud approvals?" complaint. The company-level view lives on the admin approvals *overview* page, not the personal action inbox.""" _, token = admin_user db_session.add(ApprovalRequest( external_id=f"ext-{uuid.uuid4().hex[:8]}", approval_type="cloud_access", status="pending", )) await db_session.commit() resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text data = resp.json()["data"] surfaced_types = {i["type"] for i in data["items"]} assert "approval" not in surfaced_types assert data["counts"].get("approval") is None assert data["total"] == 0 @pytest.mark.asyncio async def test_inbox_admin_in_employee_perspective_hides_admin_items( client, admin_user, db_session: AsyncSession, ) -> None: """RoleSwitcher perspective scoping (M-Mobile-Upgrade fix): a global_admin viewing the action centre ``as_role=employee`` must see NONE of their admin items — the employee perspective is a strict narrowing, identical to a real employee's inbox. Regression for the bug where the mobile action centre ignored the active role and leaked security findings / approvals into the employee view.""" _, token = admin_user from datetime import datetime now = datetime.now(UTC) db_session.add(ApprovalRequest( external_id=f"ext-{uuid.uuid4().hex[:8]}", approval_type="software_access", status="pending", )) db_session.add(AuditorFinding( control_key="mfa_coverage", finding_type="mfa_not_enrolled", dedup_key=f"d-{uuid.uuid4().hex[:8]}", subject_type="person", severity="high", title="Someone without MFA", status="open", first_seen_at=now, last_seen_at=now, )) await db_session.commit() resp = await client.get( "/api/v1/m/aggregate/inbox", headers=_auth(token), params={"as_role": "employee"}, ) assert resp.status_code == 200, resp.text data = resp.json()["data"] surfaced_types = {i["type"] for i in data["items"]} assert "approval" not in surfaced_types assert "compliance_finding" not in surfaced_types assert data["total"] == 0 @pytest.mark.asyncio async def test_inbox_global_admin_does_not_see_findings_without_auditor( client, admin_user, db_session: AsyncSession, ) -> None: """Security findings are an AUDITOR-perspective concern, not a general admin one: a global_admin who does NOT hold the auditor role must not see open findings flooding their operational action centre. Regression for the "admin home is a wall of 安全发现" report.""" _, token = admin_user from datetime import datetime now = datetime.now(UTC) db_session.add(AuditorFinding( control_key="secret_rotation", finding_type="secret_overdue", dedup_key=f"d-{uuid.uuid4().hex[:8]}", subject_type="resource", severity="critical", title="Secret overdue", status="open", first_seen_at=now, last_seen_at=now, )) await db_session.commit() resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text data = resp.json()["data"] surfaced_types = {i["type"] for i in data["items"]} assert "compliance_finding" not in surfaced_types @pytest.mark.asyncio async def test_inbox_auditor_sees_open_findings( client, auditor_user, db_session: AsyncSession, ) -> None: _, token = auditor_user from datetime import datetime now = datetime.now(UTC) db_session.add(AuditorFinding( control_key="secret_rotation", finding_type="secret_overdue", dedup_key=f"d-{uuid.uuid4().hex[:8]}", subject_type="resource", subject_display="prod-db-password", severity="critical", title="Secret overdue for rotation", status="open", first_seen_at=now, last_seen_at=now, )) await db_session.commit() resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text data = resp.json()["data"] finding = next(i for i in data["items"] if i["type"] == "compliance_finding") assert finding["severity"] == "urgent" # critical → urgent # All findings deep-link to the unified decidable worklist (W4-PR14). assert finding["deep_link"] == "/m/auditor/findings" assert finding["subtitle"] == "prod-db-password" @pytest.mark.asyncio async def test_inbox_sorts_urgent_before_info( client, auditor_user, db_session: AsyncSession, ) -> None: """An auditor with both an open critical finding (urgent) and an incoming asset transfer (info) must see the finding first.""" user, token = auditor_user from datetime import datetime now = datetime.now(UTC) await _seed_incoming_transfer(db_session, user.id, "Old laptop") db_session.add(AuditorFinding( control_key="mfa_coverage", finding_type="mfa_not_enrolled", dedup_key=f"d-{uuid.uuid4().hex[:8]}", subject_type="person", severity="critical", title="Critical finding", status="open", first_seen_at=now, last_seen_at=now, )) await db_session.commit() resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text items = resp.json()["data"]["items"] assert items[0]["severity"] == "urgent" assert items[0]["type"] == "compliance_finding" @pytest.mark.asyncio async def test_inbox_type_filter_and_pagination( client, employee_user, db_session: AsyncSession, ) -> None: user, token = employee_user for i in range(3): await _seed_incoming_transfer(db_session, user.id, f"laptop-{i}") db_session.add(ProvisioningTask( user_id=user.id, action="grant", target_type="software", source="onboarding", execution_mode="manual", status=STATUS_PENDING, )) await db_session.commit() # Filter to only transfers resp = await client.get( "/api/v1/m/aggregate/inbox", headers=_auth(token), params={"type": "asset_transfer_confirm"}, ) data = resp.json()["data"] assert data["total"] == 3 assert all(i["type"] == "asset_transfer_confirm" for i in data["items"]) # Pagination: page_size=2 → 2 then 1 resp1 = await client.get( "/api/v1/m/aggregate/inbox", headers=_auth(token), params={"type": "asset_transfer_confirm", "page": 1, "page_size": 2}, ) resp2 = await client.get( "/api/v1/m/aggregate/inbox", headers=_auth(token), params={"type": "asset_transfer_confirm", "page": 2, "page_size": 2}, ) assert len(resp1.json()["data"]["items"]) == 2 assert len(resp2.json()["data"]["items"]) == 1 # total/counts reflect the full filtered set, not the page assert resp1.json()["data"]["total"] == 3 @pytest.mark.asyncio async def test_inbox_page_size_over_cap_rejected(client, employee_user) -> None: """page_size > 100 must 422 — the FE contract caps at 100.""" _, token = employee_user resp = await client.get( "/api/v1/m/aggregate/inbox", headers=_auth(token), params={"page_size": 500}, ) assert resp.status_code == 422 # ── New provider coverage (2026-07 action-plane rebuild) ──────────── async def _seed_ticket( db: AsyncSession, *, requester_id: uuid.UUID, status: str, assignee_id: uuid.UUID | None = None, title: str = "Laptop won't boot", ) -> SupportTicket: ticket = SupportTicket( ticket_number=f"HELP-{uuid.uuid4().hex[:8]}", requester_id=requester_id, title=title, body="detail", status=status, assignee_id=assignee_id, ) db.add(ticket) await db.commit() return ticket @pytest.mark.asyncio async def test_inbox_access_attestation_surfaces_pending( client, employee_user, db_session: AsyncSession, ) -> None: """An open self-attestation campaign with an undecided item for me surfaces one aggregate action-inbox row deep-linking to the mobile access-review page.""" from datetime import datetime, timedelta user, token = employee_user now = datetime.now(UTC).replace(tzinfo=None) cert = AccessCertification( id=uuid.uuid4(), label="2026 H1 access review", campaign_type="self_attestation", scope_config={}, status="in_progress", started_at=now, due_at=now + timedelta(days=5), ) db_session.add(cert) await db_session.flush() db_session.add(AccessCertificationItem( id=uuid.uuid4(), certification_id=cert.id, person_id=user.id, access_type="software_seat", access_label="Jira", decision="pending", snapshot_at=now, )) await db_session.commit() resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text data = resp.json()["data"] item = next(i for i in data["items"] if i["type"] == "access_attestation") assert item["title"] == "2026 H1 access review" assert item["subtitle"] == "1" # one undecided item assert item["deep_link"] == "/m/access-review" assert item["deadline"] is not None @pytest.mark.asyncio async def test_inbox_access_attestation_hidden_when_all_decided( client, employee_user, db_session: AsyncSession, ) -> None: from datetime import datetime, timedelta user, token = employee_user now = datetime.now(UTC).replace(tzinfo=None) cert = AccessCertification( id=uuid.uuid4(), label="decided", campaign_type="self_attestation", scope_config={}, status="in_progress", started_at=now, due_at=now + timedelta(days=5), ) db_session.add(cert) await db_session.flush() db_session.add(AccessCertificationItem( id=uuid.uuid4(), certification_id=cert.id, person_id=user.id, access_type="software_seat", access_label="Jira", decision="keep", snapshot_at=now, )) await db_session.commit() resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text surfaced = {i["type"] for i in resp.json()["data"]["items"]} assert "access_attestation" not in surfaced @pytest.mark.asyncio async def test_inbox_support_reply_for_requester( client, employee_user, db_session: AsyncSession, ) -> None: """A ticket I raised that is waiting on my reply is MY to-do.""" user, token = employee_user ticket = await _seed_ticket( db_session, requester_id=user.id, status="waiting_requester", ) resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text item = next( i for i in resp.json()["data"]["items"] if i["type"] == "support_reply" ) assert item["deep_link"] == f"/m/support/{ticket.id}" assert item["subtitle"] == ticket.ticket_number @pytest.mark.asyncio async def test_inbox_support_assigned_for_admin( client, admin_user, db_session: AsyncSession, ) -> None: """A ticket assigned to me (IT) that I still owe work on surfaces to the workbench, deep-linking to the admin ticket page.""" user, token = admin_user other = await _seed_person(db_session) ticket = await _seed_ticket( db_session, requester_id=other, status="in_progress", assignee_id=user.id, ) resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text item = next( i for i in resp.json()["data"]["items"] if i["type"] == "support_assigned" ) assert item["deep_link"] == f"/m/admin/support/tickets/{ticket.id}" @pytest.mark.asyncio async def test_inbox_support_assigned_hidden_in_employee_perspective( client, admin_user, db_session: AsyncSession, ) -> None: """IT-workbench assigned tickets are an admin-perspective concern — an admin viewing as employee must not see them.""" user, token = admin_user other = await _seed_person(db_session) await _seed_ticket( db_session, requester_id=other, status="open", assignee_id=user.id, ) resp = await client.get( "/api/v1/m/aggregate/inbox", headers=_auth(token), params={"as_role": "employee"}, ) assert resp.status_code == 200, resp.text surfaced = {i["type"] for i in resp.json()["data"]["items"]} assert "support_assigned" not in surfaced @pytest.mark.asyncio async def test_inbox_support_assigned_isolation( client, admin_user, db_session: AsyncSession, ) -> None: """A ticket assigned to a DIFFERENT admin must not appear in mine.""" _, token = admin_user other_admin = await _seed_person(db_session) requester = await _seed_person(db_session) await _seed_ticket( db_session, requester_id=requester, status="in_progress", assignee_id=other_admin, ) resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text surfaced = {i["type"] for i in resp.json()["data"]["items"]} assert "support_assigned" not in surfaced async def _seed_attendance( db: AsyncSession, *, person_id: uuid.UUID, status: str, leader_id: uuid.UUID | None = None, ) -> AttendanceReport: from datetime import date cycle = AttendanceCycle( id=uuid.uuid4(), cycle_key="2026-06", start_date=date(2026, 5, 21), end_date=date(2026, 6, 20), ) db.add(cycle) await db.flush() report = AttendanceReport( id=uuid.uuid4(), cycle_id=cycle.id, person_id=person_id, leader_person_id=leader_id, status=status, version=1, snapshot_json={"summary": {}, "days": []}, work_days=20, ) db.add(report) await db.commit() return report @pytest.mark.asyncio async def test_inbox_attendance_confirm_for_intern( client, employee_user, db_session: AsyncSession, ) -> None: user, token = employee_user report = await _seed_attendance( db_session, person_id=user.id, status=STATUS_PENDING_INTERN, ) resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text item = next( i for i in resp.json()["data"]["items"] if i["type"] == "attendance_confirm" ) assert item["title"] == "2026-06" assert item["deep_link"] == f"/m/attendance/{report.id}" @pytest.mark.asyncio async def test_inbox_attendance_confirm_for_leader( client, employee_user, db_session: AsyncSession, ) -> None: """A ``pending_leader`` report surfaces to the LEADER, not the intern.""" user, token = employee_user intern = await _seed_person(db_session) report = await _seed_attendance( db_session, person_id=intern, status=STATUS_PENDING_LEADER, leader_id=user.id, ) resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text item = next( i for i in resp.json()["data"]["items"] if i["type"] == "attendance_confirm" ) assert item["deep_link"] == f"/m/attendance/{report.id}" @pytest.mark.asyncio async def test_inbox_attendance_confirm_isolation( client, employee_user, db_session: AsyncSession, ) -> None: """A ``pending_intern`` report belonging to someone else must not show (I'm not the intern, and I'm not their leader).""" _, token = employee_user other = await _seed_person(db_session) await _seed_attendance( db_session, person_id=other, status=STATUS_PENDING_INTERN, ) resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text surfaced = {i["type"] for i in resp.json()["data"]["items"]} assert "attendance_confirm" not in surfaced @pytest.mark.asyncio async def test_inbox_inventory_response( client, employee_user, db_session: AsyncSession, ) -> None: """Pending inventory items routed to me collapse into one aggregate action item deep-linking to the mobile inventory page.""" user, token = employee_user asset = AssetItem( asset_tag=f"INV-{uuid.uuid4().hex[:8]}", asset_name="Dell", category="laptop", status="assigned", assigned_to=user.id, ) db_session.add(asset) await db_session.flush() cycle = AssetInventoryCycle( id=uuid.uuid4(), name="Q3 count", status="active", started_at=datetime.now(UTC), total_items=1, pending_count=1, ) db_session.add(cycle) await db_session.flush() db_session.add(AssetInventoryItem( id=uuid.uuid4(), cycle_id=cycle.id, asset_id=asset.id, holder_kind="assignee", holder_person_id=user.id, asset_tag_snapshot=asset.asset_tag, asset_category_snapshot=asset.category, asset_status_snapshot=asset.status, status="pending", )) await db_session.commit() resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text item = next( i for i in resp.json()["data"]["items"] if i["type"] == "inventory_response" ) assert item["title"] == "1" assert item["deep_link"] == "/m/inventory" @pytest.mark.asyncio async def test_inbox_onboarding_ack_during_open_process( client, employee_user, db_session: AsyncSession, ) -> None: """A new hire with an OPEN onboarding process and an un-acked active checklist item sees one aggregate onboarding-ack action item.""" from datetime import date user, token = employee_user db_session.add(OnboardingProcess( id=uuid.uuid4(), person_id=user.id, hire_date=date.today(), status=PROC_SCANNING, )) db_session.add(OnboardingAckItem( id=uuid.uuid4(), title="Security policy", required=True, is_active=True, )) await db_session.commit() resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text item = next( i for i in resp.json()["data"]["items"] if i["type"] == "onboarding_ack" ) assert item["deep_link"] == "/m/lifecycle" @pytest.mark.asyncio async def test_inbox_onboarding_ack_hidden_without_open_process( client, employee_user, db_session: AsyncSession, ) -> None: """An active checklist item alone must NOT create a perpetual to-do for a tenured employee with no open onboarding process.""" _, token = employee_user db_session.add(OnboardingAckItem( id=uuid.uuid4(), title="Security policy", required=True, is_active=True, )) await db_session.commit() resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text surfaced = {i["type"] for i in resp.json()["data"]["items"]} assert "onboarding_ack" not in surfaced async def _seed_override( db: AsyncSession, *, secondary_signer_id: uuid.UUID, status: str = DUAL_SIGN_PENDING_SECONDARY, ) -> LeaverDualSignOverride: from datetime import date, datetime leaver_id = await _seed_person(db) primary_id = await _seed_person(db) proc = OffboardingProcess( person_id=leaver_id, leaver_date=date.today(), status="blocked", ) db.add(proc) await db.flush() override = LeaverDualSignOverride( id=uuid.uuid4(), process_id=proc.id, person_id=leaver_id, status=status, primary_signer_user_id=primary_id, primary_signed_at=datetime.now(UTC), primary_reason="urgent departure, waive hardware blocker per director", secondary_signer_user_id=secondary_signer_id, overridden_blocker_ids=[], overridden_blockers_snapshot=[], trace_id=uuid.uuid4(), ) db.add(override) await db.commit() return override @pytest.mark.asyncio async def test_inbox_leaver_override_for_named_signer( client, admin_user, db_session: AsyncSession, ) -> None: """An override awaiting MY secondary sign-off surfaces urgently and deep-links to the decision page.""" user, token = admin_user override = await _seed_override(db_session, secondary_signer_id=user.id) resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text item = next( i for i in resp.json()["data"]["items"] if i["type"] == "leaver_override_signoff" ) assert item["severity"] == "urgent" assert item["deep_link"] == f"/m/admin/leaver-override/{override.id}" @pytest.mark.asyncio async def test_inbox_leaver_override_isolation( client, admin_user, db_session: AsyncSession, ) -> None: """An override naming a DIFFERENT secondary signer must not appear.""" _, token = admin_user other_signer = await _seed_person(db_session) await _seed_override(db_session, secondary_signer_id=other_signer) resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text surfaced = {i["type"] for i in resp.json()["data"]["items"]} assert "leaver_override_signoff" not in surfaced @pytest.mark.asyncio async def test_inbox_leaver_override_not_for_plain_employee( client, employee_user, db_session: AsyncSession, ) -> None: """An employee lacks the secondary-sign baseline role, so even a row naming them (defensive) never surfaces via this provider.""" user, token = employee_user await _seed_override(db_session, secondary_signer_id=user.id) resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text surfaced = {i["type"] for i in resp.json()["data"]["items"]} assert "leaver_override_signoff" not in surfaced # ── Cloud approval providers (PR3 — cloud permission approval closure) ── async def _seed_scope_and_grant( db: AsyncSession, *, owner_id: uuid.UUID | None, grant_person_id: uuid.UUID, grant_status: str = "pending", ): """Seed an integration + scope (optionally owned by ``owner_id``) + a grant requested by ``grant_person_id``. Returns (scope, grant).""" from app.domains.cmdb.models import ( SCOPE_STATUS_ACTIVE, CmdbCloudAccessGrant, CmdbCloudScope, ) from app.integrations.models import Integration integ = Integration( id=uuid.uuid4(), kind="cloud", provider="azure", name=f"azure-{uuid.uuid4().hex[:6]}", slug=f"azure-{uuid.uuid4().hex[:6]}", enabled=True, is_primary=False, capabilities=["servers"], ) db.add(integ) await db.flush() scope = CmdbCloudScope( id=uuid.uuid4(), integration_id=str(integ.id), scope_type="azure_subscription", external_id=uuid.uuid4().hex[:16], display_name=f"sub-{uuid.uuid4().hex[:6]}", status=SCOPE_STATUS_ACTIVE, environment="dev", owner_person_id=str(owner_id) if owner_id else None, ) db.add(scope) await db.flush() grant = CmdbCloudAccessGrant( id=uuid.uuid4(), scope_id=str(scope.id), person_id=str(grant_person_id), role_key="reader", role_display_name="Reader", status=grant_status, granted_at=datetime.now(UTC), ) db.add(grant) await db.commit() return scope, grant @pytest.mark.asyncio async def test_inbox_cloud_grant_approval_for_scope_owner( client, employee_user, db_session: AsyncSession, ) -> None: """A pending grant on a scope I OWN (requested by someone else) is my approval to make.""" user, token = employee_user requester = await _seed_person(db_session) _scope, grant = await _seed_scope_and_grant( db_session, owner_id=user.id, grant_person_id=requester, ) resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text item = next( i for i in resp.json()["data"]["items"] if i["type"] == "cloud_grant_approval" ) assert item["deep_link"] == f"/m/cloud/grant-approval/{grant.id}" assert item["subtitle"] == "Reader" @pytest.mark.asyncio async def test_inbox_cloud_grant_approval_excludes_self_requested( client, employee_user, db_session: AsyncSession, ) -> None: """An owner's OWN pending grant on their scope is not surfaced — four-eyes; self-approve would 409.""" user, token = employee_user await _seed_scope_and_grant( db_session, owner_id=user.id, grant_person_id=user.id, ) resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text surfaced = {i["type"] for i in resp.json()["data"]["items"]} assert "cloud_grant_approval" not in surfaced @pytest.mark.asyncio async def test_inbox_cloud_grant_approval_hidden_for_non_owner( client, employee_user, db_session: AsyncSession, ) -> None: """A pending grant on a scope owned by someone else must not surface to a plain employee who is neither owner nor cloud admin.""" _, token = employee_user owner = await _seed_person(db_session) requester = await _seed_person(db_session) await _seed_scope_and_grant( db_session, owner_id=owner, grant_person_id=requester, ) resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text surfaced = {i["type"] for i in resp.json()["data"]["items"]} assert "cloud_grant_approval" not in surfaced @pytest.mark.asyncio async def test_inbox_cloud_grant_approval_admin_sees_all( client, admin_user, db_session: AsyncSession, ) -> None: """A global_admin (can_manage_cloud) sees every pending grant, even on scopes owned by others.""" _, token = admin_user owner = await _seed_person(db_session) requester = await _seed_person(db_session) _scope, grant = await _seed_scope_and_grant( db_session, owner_id=owner, grant_person_id=requester, ) resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text ids = {i["id"] for i in resp.json()["data"]["items"]} assert f"cloud_grant_approval:{grant.id}" in ids async def _seed_scope_request( db: AsyncSession, *, requester_id: uuid.UUID, status: str = "pending", ): from decimal import Decimal from app.domains.cmdb.models import CmdbCloudScopeRequest from app.integrations.models import Integration integ = Integration( id=uuid.uuid4(), kind="cloud", provider="aws", name=f"aws-{uuid.uuid4().hex[:6]}", slug=f"aws-{uuid.uuid4().hex[:6]}", enabled=True, is_primary=False, capabilities=["servers"], ) db.add(integ) await db.flush() req = CmdbCloudScopeRequest( id=str(uuid.uuid4()), requester_person_id=str(requester_id), integration_id=str(integ.id), provider="aws", name="new-team-account", monthly_budget_usd=Decimal("500"), status=status, ) db.add(req) await db.commit() return req @pytest.mark.asyncio async def test_inbox_cloud_scope_approval_for_admin( client, admin_user, db_session: AsyncSession, ) -> None: _, token = admin_user requester = await _seed_person(db_session) req = await _seed_scope_request(db_session, requester_id=requester) resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text item = next( i for i in resp.json()["data"]["items"] if i["type"] == "cloud_scope_approval" ) assert item["deep_link"] == f"/m/cloud/scope-requests/{req.id}" assert item["title"] == "new-team-account" @pytest.mark.asyncio async def test_inbox_cloud_scope_approval_hidden_for_employee( client, employee_user, db_session: AsyncSession, ) -> None: _, token = employee_user requester = await _seed_person(db_session) await _seed_scope_request(db_session, requester_id=requester) resp = await client.get("/api/v1/m/aggregate/inbox", headers=_auth(token)) assert resp.status_code == 200, resp.text surfaced = {i["type"] for i in resp.json()["data"]["items"]} assert "cloud_scope_approval" not in surfaced