"""Cloud Admin Overview — multi-tab fleet dashboard endpoint tests. Covers: * **AuthZ posture**: ``GET /api/v1/cmdb/overview/admin`` is ``global_admin``-only (``can_manage_cloud``). Employees, auditors, and asset admins all bounce with 403. Cloud_admin (a perpetual grant marker) is intentionally NOT in the allowlist. * **Shape contract**: response carries ``filter``, ``subscriptions``, ``permissions``, ``iam_users``, ``alerts`` top-level keys. Each sub-section's keys match the frontend's expectations. * **Aggregation correctness**: live scopes count, active grant count, recovered grants split by reason, orphan-swept counts, expiring buckets, owner_health, integration filter scoping. * **Filter scoping**: ``integration_id`` narrows every section. Unknown integration_id → 404. Malformed UUID → 422. These tests intentionally avoid mocking the DB — they hit the real sqlite/MySQL test DB through the standard ``client`` fixture so the SQL paths exercise actual joins. """ from __future__ import annotations import uuid from datetime import UTC, datetime, timedelta import pytest from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.domains.cmdb.models import ( GRANT_REVOKE_REASON_OFFBOARDING, GRANT_REVOKE_REASON_ORPHAN_PRINCIPAL, GRANT_STATUS_ACTIVE, GRANT_STATUS_REVOKED, SCOPE_STATUS_ACTIVE, CmdbCloudAccessGrant, CmdbCloudScope, ) from app.integrations.models import Integration def _now_naive() -> datetime: return datetime.now(UTC).replace(tzinfo=None) async def _mk_integ( db: AsyncSession, *, provider: str = "aws", slug: str | None = None, ) -> Integration: row = Integration( id=uuid.uuid4(), kind="cloud", provider=provider, name=f"{provider}-overview-{uuid.uuid4().hex[:4]}", slug=slug or f"{provider}-{uuid.uuid4().hex[:6]}", enabled=True, is_primary=False, capabilities=["servers"], ) db.add(row) await db.commit() return row async def _mk_scope( db: AsyncSession, integration: Integration, *, status: str = SCOPE_STATUS_ACTIVE, environment: str = "dev", owner_person_id: str | None = None, ) -> CmdbCloudScope: s = CmdbCloudScope( id=uuid.uuid4(), integration_id=str(integration.id), scope_type=f"{integration.provider}_account", external_id=uuid.uuid4().hex[:16], display_name=f"scope-{uuid.uuid4().hex[:6]}", status=status, environment=environment, owner_person_id=owner_person_id, first_seen_at=_now_naive(), last_synced_at=_now_naive(), ) db.add(s) await db.commit() return s async def _mk_org_membership( db: AsyncSession, person_id: str, department: str, ) -> None: """Attach ``person_id`` to ``department`` as their primary org node — the exact shape ``resolve_department_person_ids`` matches on (``is_primary`` + ``org_node_name`` + ``deleted_at IS NULL``). """ from app.identity.person_models import IdentityOrgMembership db.add(IdentityOrgMembership( id=uuid.uuid4(), person_id=uuid.UUID(person_id), source_system="successfactors", org_node_source_id=f"node-{uuid.uuid4().hex[:8]}", org_node_type="department", org_node_name=department, org_path=f"PingCAP / {department}", is_primary=True, membership_status="active", )) await db.commit() async def _mk_person(db: AsyncSession, *, email_tag: str): """Minimal active IdentityPerson for owner/holder attribution.""" from app.identity.person_models import IdentityPerson pid = uuid.uuid4() db.add(IdentityPerson( id=pid, work_email=f"{email_tag}-{pid.hex[:6]}@test.com", display_name=f"Person {email_tag}", employment_status="active", source_system="successfactors", hris_person_id=f"sf-{email_tag}-{pid.hex[:6]}", )) await db.commit() return pid async def _mk_grant( db: AsyncSession, scope: CmdbCloudScope, *, person_id: str | None, role_key: str = "Reader", status: str = GRANT_STATUS_ACTIVE, expires_in_days: int | None = 30, revoke_reason: str | None = None, revoked_days_ago: int | None = None, external_principal_id: str | None = None, ) -> CmdbCloudAccessGrant: expires_at = ( _now_naive() + timedelta(days=expires_in_days) if expires_in_days is not None else None ) revoked_at = ( _now_naive() - timedelta(days=revoked_days_ago) if revoked_days_ago is not None else None ) g = CmdbCloudAccessGrant( id=uuid.uuid4(), scope_id=str(scope.id), person_id=person_id, role_key=role_key, status=status, granted_at=_now_naive(), expires_at=expires_at, revoked_at=revoked_at, revoke_reason=revoke_reason, external_principal_id=external_principal_id, source="helix", ) db.add(g) await db.commit() return g # ── AuthZ ─────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_overview_admin_403_for_asset_admin(client, asset_admin_user): _, token = asset_admin_user resp = await client.get( "/api/v1/cmdb/overview/admin", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 403 @pytest.mark.asyncio async def test_overview_admin_200_for_global_admin( client, admin_user, ): _, token = admin_user resp = await client.get( "/api/v1/cmdb/overview/admin", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200, resp.text body = resp.json() assert body["success"] is True data = body["data"] # Top-level shape contract. for k in ("filter", "subscriptions", "permissions", "iam_users", "alerts"): assert k in data, f"missing top-level key: {k}" @pytest.mark.asyncio async def test_overview_admin_200_for_finance( client, db_session, ): """Finance role gets read access (2026-05-16, user direction: "这个页面给财务人员权限 允许他们查看"). Mutations on individual sub-resources remain admin-only per endpoint. """ from app.auth.models import AuthRole, AuthUserRole from app.core.security import create_access_token from app.identity.person_models import IdentityPerson async def _role_id(name: str) -> uuid.UUID: res = await db_session.execute( select(AuthRole.id).where(AuthRole.name == name) ) rid = res.scalar_one_or_none() if rid is not None: return rid if isinstance(rid, uuid.UUID) else uuid.UUID(str(rid)) # Seed-on-demand: some test DBs predate migration 043's # finance/hr seed (the migration was edited after first apply). # Insert an idempotent system row so the test is robust against # that state without modifying conftest globals. new_id = uuid.uuid4() db_session.add(AuthRole( id=new_id, name=name, description=f"{name} role (test seed)", is_system=True, )) await db_session.flush() return new_id user_id = uuid.uuid4() user = IdentityPerson( id=user_id, work_email="finance-overview@test.com", display_name="Finance Overview Viewer", employment_status="active", source_system="successfactors", hris_person_id="sf-fin-overview-1", ) db_session.add(user) await db_session.flush() db_session.add(AuthUserRole( user_id=user_id, role_id=await _role_id("employee"), )) db_session.add(AuthUserRole( user_id=user_id, role_id=await _role_id("finance"), )) await db_session.commit() token = create_access_token(str(user_id), ["employee", "finance"]) resp = await client.get( "/api/v1/cmdb/overview/admin", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200, resp.text body = resp.json() assert body["success"] is True # Shape parity with the admin path — finance gets the same payload. data = body["data"] for k in ("filter", "subscriptions", "permissions", "iam_users", "alerts"): assert k in data, f"missing top-level key: {k}" # ── Shape ─────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_overview_admin_shape_keys(client, admin_user): _, token = admin_user resp = await client.get( "/api/v1/cmdb/overview/admin", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200, resp.text data = resp.json()["data"] subs = data["subscriptions"] for k in ( "totals", "cost_ranking", "owner_health", "by_status", "by_environment", "by_scope_type", "by_provider", "by_integration", "recent_decommissions", "pending_scope_requests", ): assert k in subs, f"subscriptions missing key: {k}" # cost_ranking shape — month-anchored top spenders for the # Subscriptions tab. Empty fleet is allowed (top:[], fleet_total:0). ranking = subs["cost_ranking"] for k in ("month", "is_partial", "fleet_total_usd", "top"): assert k in ranking, f"cost_ranking missing key: {k}" assert isinstance(ranking["top"], list) # filter echoes the month param (None when default). assert "month" in data["filter"] # ``ai_dedicated`` totals key was removed by the 2026-05-11 # AI-spending refactor (cmdb_ai_resources / is_ai_dedicated deletion); # the four-bucket model lives in /ai-spending now. for k in ("all", "active", "closed", "suspended"): assert k in subs["totals"] # Round-20 (2026-05-12) split: ``owner_left_org`` retained as a # backwards-compatible aggregate; new widgets MUST use the two # disjoint buckets ``owner_in_offboarding`` (leaver_pending) and # ``owner_left`` (left or other non-active). for k in ("owned", "unowned", "owner_left_org", "owner_in_offboarding", "owner_left"): assert k in subs["owner_health"] perms = data["permissions"] for k in ( "totals", "by_status", "by_provider_active", "by_environment_active", "expiring", "recovered", ): assert k in perms, f"permissions missing key: {k}" for k in ("active", "pending", "expired", "revoked", "disabled", "all", "distinct_active_holders"): assert k in perms["totals"] for k in ("t1", "t7", "t14", "t30", "active_past_due"): assert k in perms["expiring"] for k in ("total", "last_30d", "last_90d", "by_reason_30d", "orphan_swept_total", "orphan_swept_30d", "recent"): assert k in perms["recovered"] iam = data["iam_users"] for k in ("total", "active", "revoked_lifetime", "owned", "unowned", "mfa_disabled", "by_provider"): assert k in iam alerts = data["alerts"] for k in ("unowned_scopes", "active_past_due_grants", "prod_overdue_grants", "pending_grants"): assert k in alerts # ── Aggregation correctness ───────────────────────────────────────── @pytest.mark.asyncio async def test_overview_admin_counts_active_and_recovered( client, db_session: AsyncSession, admin_user, employee_user, ): """One scope, two grants: one active, one revoked (offboarding) 5 days ago. Verify totals + recovered.last_30d + by_reason_30d. """ _, admin_token = admin_user employee, _ = employee_user integ = await _mk_integ(db_session, provider="aws") scope = await _mk_scope( db_session, integ, owner_person_id=str(employee.id), environment="prod", ) # Active grant — holder is the employee. await _mk_grant( db_session, scope, person_id=str(employee.id), status=GRANT_STATUS_ACTIVE, expires_in_days=20, ) # Revoked grant 5 days ago, offboarding reason. await _mk_grant( db_session, scope, person_id=str(employee.id), role_key="Contributor", status=GRANT_STATUS_REVOKED, expires_in_days=None, revoke_reason=GRANT_REVOKE_REASON_OFFBOARDING, revoked_days_ago=5, ) resp = await client.get( "/api/v1/cmdb/overview/admin", params={"integration_id": str(integ.id)}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert resp.status_code == 200, resp.text perms = resp.json()["data"]["permissions"] assert perms["totals"]["active"] >= 1 assert perms["totals"]["revoked"] >= 1 assert perms["totals"]["distinct_active_holders"] >= 1 assert perms["recovered"]["last_30d"] >= 1 # by_reason_30d carries our offboarding reason. reasons = {r["name"]: r["count"] for r in perms["recovered"]["by_reason_30d"]} assert reasons.get(GRANT_REVOKE_REASON_OFFBOARDING, 0) >= 1 @pytest.mark.asyncio async def test_overview_admin_orphan_swept_counted( client, db_session: AsyncSession, admin_user, ): """Orphan-swept history rows (person_id NULL, revoke_reason='orphan_principal_swept') count toward ``recovered.orphan_swept_*``. Critical signal — the user explicitly asked for "回收了多少个权限" visibility. """ _, admin_token = admin_user integ = await _mk_integ(db_session, provider="aws") scope = await _mk_scope(db_session, integ) await _mk_grant( db_session, scope, person_id=None, role_key="DBaaS-Prod-Admin-Role", status=GRANT_STATUS_REVOKED, expires_in_days=None, revoke_reason=GRANT_REVOKE_REASON_ORPHAN_PRINCIPAL, revoked_days_ago=2, external_principal_id="95670a3c83-c108d80c-cbbe-4c76-a5c7-d95928e4b566", ) resp = await client.get( "/api/v1/cmdb/overview/admin", params={"integration_id": str(integ.id)}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert resp.status_code == 200, resp.text rec = resp.json()["data"]["permissions"]["recovered"] assert rec["orphan_swept_total"] >= 1 assert rec["orphan_swept_30d"] >= 1 # ``recent`` table includes this orphan row with is_orphan_history=True. found = [r for r in rec["recent"] if r["is_orphan_history"]] assert any( r["revoke_reason"] == GRANT_REVOKE_REASON_ORPHAN_PRINCIPAL for r in found ) @pytest.mark.asyncio async def test_overview_admin_filter_isolates_other_tenants( client, db_session: AsyncSession, admin_user, employee_user, ): """``integration_id=A`` must not bleed counts from integration B. Catches the kind of join-without-filter bug that turns a "single-tenant view" into a fleet view. """ _, admin_token = admin_user employee, _ = employee_user integ_a = await _mk_integ(db_session, provider="aws") integ_b = await _mk_integ(db_session, provider="azure") scope_a = await _mk_scope(db_session, integ_a) scope_b = await _mk_scope(db_session, integ_b) await _mk_grant( db_session, scope_a, person_id=str(employee.id), status=GRANT_STATUS_ACTIVE, expires_in_days=20, ) await _mk_grant( db_session, scope_b, person_id=str(employee.id), role_key="Reader-B", status=GRANT_STATUS_ACTIVE, expires_in_days=20, ) resp_a = await client.get( "/api/v1/cmdb/overview/admin", params={"integration_id": str(integ_a.id)}, headers={"Authorization": f"Bearer {admin_token}"}, ) resp_b = await client.get( "/api/v1/cmdb/overview/admin", params={"integration_id": str(integ_b.id)}, headers={"Authorization": f"Bearer {admin_token}"}, ) a_data = resp_a.json()["data"] b_data = resp_b.json()["data"] # Both filters resolve. assert a_data["filter"]["integration_id"] == str(integ_a.id) assert b_data["filter"]["integration_id"] == str(integ_b.id) # Tenant A only sees the AWS provider in by_provider_active. a_providers = { r["name"] for r in a_data["permissions"]["by_provider_active"] } b_providers = { r["name"] for r in b_data["permissions"]["by_provider_active"] } # The union may have only one row each since only that scope's # provider is represented. assert "aws" in a_providers assert "azure" in b_providers assert "azure" not in a_providers assert "aws" not in b_providers # ── Edge cases ────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_overview_admin_404_unknown_integration(client, admin_user): """An unknown integration_id should 404, not silently return zeros.""" _, admin_token = admin_user resp = await client.get( "/api/v1/cmdb/overview/admin", params={"integration_id": str(uuid.uuid4())}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert resp.status_code == 404 @pytest.mark.asyncio async def test_overview_admin_422_malformed_integration_id(client, admin_user): """Malformed UUID bounces at the validation gate, not deep in SQL.""" _, admin_token = admin_user resp = await client.get( "/api/v1/cmdb/overview/admin", params={"integration_id": "not-a-uuid"}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert resp.status_code == 422 @pytest.mark.asyncio async def test_overview_admin_unauthenticated_401(client): """No bearer token → 401, never anonymous data.""" resp = await client.get("/api/v1/cmdb/overview/admin") assert resp.status_code in (401, 403) # ── Regression pins from launch-readiness reviews ─────────────────── @pytest.mark.asyncio async def test_overview_admin_excludes_soft_deleted_scope_grants( client, db_session: AsyncSession, admin_user, employee_user, ): """Grants whose parent scope is soft-deleted MUST NOT count toward permissions totals or alerts. Pins the P0 launch-review finding — earlier revisions joined through ``CmdbCloudScope`` without a ``deleted_at IS NULL`` filter, leaking grants from decommissioned scopes into Overview/Permissions/Alerts and creating mismatched counters across tabs. """ _, admin_token = admin_user employee, _ = employee_user integ = await _mk_integ(db_session, provider="aws") live_scope = await _mk_scope(db_session, integ) deleted_scope = await _mk_scope(db_session, integ) # Active grant on the live scope — should count. await _mk_grant( db_session, live_scope, person_id=str(employee.id), status=GRANT_STATUS_ACTIVE, expires_in_days=20, ) # Active grant on a scope we then soft-delete — must NOT count. await _mk_grant( db_session, deleted_scope, person_id=str(employee.id), role_key="GhostRole", status=GRANT_STATUS_ACTIVE, expires_in_days=20, ) deleted_scope.deleted_at = _now_naive() await db_session.commit() resp = await client.get( "/api/v1/cmdb/overview/admin", params={"integration_id": str(integ.id)}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert resp.status_code == 200, resp.text perms = resp.json()["data"]["permissions"] # Only the live-scope grant counts. assert perms["totals"]["active"] == 1 # by_status should sum to 1 across all states (no leakage). by_status_sum = sum(r["count"] for r in perms["by_status"]) assert by_status_sum == 1 # ── Department filter (2026-06-03) ────────────────────────────────── @pytest.mark.asyncio async def test_analytics_department_filter_owner_scopable_and_ownerless( db_session: AsyncSession, ): """``CmdbService.get_analytics(department=X)`` narrows owner-attributable models (servers) to that department's owners, and ownerless models (object storage — no ``owner_person_id`` column) report zero under the filter instead of leaking the fleet count. """ from app.domains.cmdb.models import CmdbObjectStorage, CmdbServer from app.domains.cmdb.service import CmdbService integ = await _mk_integ(db_session, provider="aws") dept = f"Eng-{uuid.uuid4().hex[:6]}" person_in = await _mk_person(db_session, email_tag="srv-in") person_out = await _mk_person(db_session, email_tag="srv-out") await _mk_org_membership(db_session, str(person_in), dept) await _mk_org_membership( db_session, str(person_out), f"Ops-{uuid.uuid4().hex[:6]}", ) for owner in (str(person_in), str(person_out)): db_session.add(CmdbServer( id=uuid.uuid4(), integration_id=str(integ.id), instance_id=f"i-{uuid.uuid4().hex[:12]}", status="running", owner_person_id=owner, first_seen_at=_now_naive(), last_synced_at=_now_naive(), )) # An ownerless resource (no owner_person_id column on the model). db_session.add(CmdbObjectStorage( id=uuid.uuid4(), integration_id=str(integ.id), bucket_name=f"bucket-{uuid.uuid4().hex[:10]}", status="active", first_seen_at=_now_naive(), last_synced_at=_now_naive(), )) await db_session.commit() # No department → both servers + the bucket are visible for this tenant. full = await CmdbService.get_analytics( db_session, integration_id=str(integ.id), ) assert full["totals"]["servers"] == 2 assert full["totals"]["object_storage"] == 1 # department=X → only the in-dept server; the bucket can't be # attributed to a department → 0. scoped = await CmdbService.get_analytics( db_session, integration_id=str(integ.id), department=dept, ) assert scoped["totals"]["servers"] == 1 assert scoped["totals"]["object_storage"] == 0 @pytest.mark.asyncio async def test_overview_admin_department_filter_narrows_owner_and_holder( client, db_session: AsyncSession, admin_user, ): """``department=X`` narrows subscriptions to scopes owned by people in X and permissions to grants *held by* people in X. Resources owned / held by someone in another department drop out. Within a single ``integration_id`` so the assertion is exact and isolated from other test data. """ _, admin_token = admin_user integ = await _mk_integ(db_session, provider="aws") dept = f"Dept-{uuid.uuid4().hex[:6]}" person_in = await _mk_person(db_session, email_tag="in") person_out = await _mk_person(db_session, email_tag="out") await _mk_org_membership(db_session, str(person_in), dept) # person_out gets a DIFFERENT department so they never match. await _mk_org_membership( db_session, str(person_out), f"Other-{uuid.uuid4().hex[:6]}", ) scope_in = await _mk_scope( db_session, integ, owner_person_id=str(person_in), ) scope_out = await _mk_scope( db_session, integ, owner_person_id=str(person_out), ) # One active grant held by each person. await _mk_grant( db_session, scope_in, person_id=str(person_in), status=GRANT_STATUS_ACTIVE, expires_in_days=20, ) await _mk_grant( db_session, scope_out, person_id=str(person_out), role_key="Reader-Out", status=GRANT_STATUS_ACTIVE, expires_in_days=20, ) resp = await client.get( "/api/v1/cmdb/overview/admin", params={"integration_id": str(integ.id), "department": dept}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert resp.status_code == 200, resp.text data = resp.json()["data"] # filter echoes the department. assert data["filter"]["department"] == dept # Subscriptions: only the in-department scope counts. assert data["subscriptions"]["totals"]["all"] == 1 assert data["subscriptions"]["owner_health"]["owned"] == 1 # Permissions: only the in-department holder's grant counts. assert data["permissions"]["totals"]["active"] == 1 assert data["permissions"]["totals"]["distinct_active_holders"] == 1 @pytest.mark.asyncio async def test_overview_admin_department_filter_unknown_returns_zero( client, db_session: AsyncSession, admin_user, employee_user, ): """A department with no members narrows every owner/holder section to zero — never leaks the fleet total. Pins the ``IN ()`` empty-set contract of ``resolve_department_person_ids``. """ _, admin_token = admin_user employee, _ = employee_user integ = await _mk_integ(db_session, provider="aws") scope = await _mk_scope( db_session, integ, owner_person_id=str(employee.id), ) await _mk_grant( db_session, scope, person_id=str(employee.id), status=GRANT_STATUS_ACTIVE, expires_in_days=20, ) resp = await client.get( "/api/v1/cmdb/overview/admin", params={ "integration_id": str(integ.id), "department": f"NoSuchDept-{uuid.uuid4().hex[:8]}", }, headers={"Authorization": f"Bearer {admin_token}"}, ) assert resp.status_code == 200, resp.text data = resp.json()["data"] assert data["subscriptions"]["totals"]["all"] == 0 assert data["permissions"]["totals"]["active"] == 0 assert data["alerts"]["pending_grants"] == 0 @pytest.mark.asyncio async def test_overview_admin_no_department_is_unchanged( client, db_session: AsyncSession, admin_user, employee_user, ): """Omitting ``department`` leaves the fleet view intact — ``filter.department`` is null and counts are not narrowed. """ _, admin_token = admin_user employee, _ = employee_user integ = await _mk_integ(db_session, provider="aws") scope = await _mk_scope( db_session, integ, owner_person_id=str(employee.id), ) await _mk_grant( db_session, scope, person_id=str(employee.id), status=GRANT_STATUS_ACTIVE, expires_in_days=20, ) resp = await client.get( "/api/v1/cmdb/overview/admin", params={"integration_id": str(integ.id)}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert resp.status_code == 200, resp.text data = resp.json()["data"] assert data["filter"]["department"] is None assert data["subscriptions"]["totals"]["all"] == 1 assert data["permissions"]["totals"]["active"] == 1 @pytest.mark.asyncio async def test_list_scopes_department_filter_narrows_by_owner( client, db_session: AsyncSession, admin_user, ): """``GET /cmdb/scopes?department=X`` narrows the scope list (the Cloud Overview "云子账号/订阅查询" lookup) to scopes whose OWNER sits in X — the same owner-attribution the rest of the page already applied. A scope owned by someone in another department drops out; omitting the filter returns both. Regression for the ScopeLookupCard ignoring the global department filter. """ _, admin_token = admin_user integ = await _mk_integ(db_session, provider="aws") dept = f"Dept-{uuid.uuid4().hex[:6]}" person_in = await _mk_person(db_session, email_tag="scin") person_out = await _mk_person(db_session, email_tag="scout") await _mk_org_membership(db_session, str(person_in), dept) await _mk_org_membership( db_session, str(person_out), f"Other-{uuid.uuid4().hex[:6]}", ) scope_in = await _mk_scope( db_session, integ, owner_person_id=str(person_in), ) await _mk_scope(db_session, integ, owner_person_id=str(person_out)) headers = {"Authorization": f"Bearer {admin_token}"} # department=X → only the in-department scope. resp = await client.get( "/api/v1/cmdb/scopes", params={"integration_id": str(integ.id), "department": dept}, headers=headers, ) assert resp.status_code == 200, resp.text body = resp.json() assert body["meta"]["total"] == 1 assert [r["id"] for r in body["data"]] == [str(scope_in.id)] # Unknown department (no members) → zero rows, never leaks the fleet. resp = await client.get( "/api/v1/cmdb/scopes", params={ "integration_id": str(integ.id), "department": f"NoSuchDept-{uuid.uuid4().hex[:8]}", }, headers=headers, ) assert resp.status_code == 200, resp.text assert resp.json()["meta"]["total"] == 0 # No department → both scopes visible. resp = await client.get( "/api/v1/cmdb/scopes", params={"integration_id": str(integ.id)}, headers=headers, ) assert resp.status_code == 200, resp.text assert resp.json()["meta"]["total"] == 2 @pytest.mark.asyncio async def test_overview_admin_recovered_total_requires_revoked_at( client, db_session: AsyncSession, admin_user, employee_user, ): """``recovered.total`` MUST require ``revoked_at IS NOT NULL`` so it is consistent with ``last_30d`` / ``last_90d``. Earlier revisions counted ``status='expired'`` rows whose ``revoked_at`` was never stamped (no sweeper run yet) toward the lifetime total but not the windows — total > sum-of-windows for the same time period was confusing. P0 finding from the launch-readiness review pass. """ _, admin_token = admin_user employee, _ = employee_user integ = await _mk_integ(db_session, provider="aws") scope = await _mk_scope(db_session, integ) # An expired grant whose ``revoked_at`` is NULL (sweeper hasn't run). # Must NOT contribute to recovered.total. g = await _mk_grant( db_session, scope, person_id=str(employee.id), role_key="UnreapedExpired", status=GRANT_STATUS_REVOKED, expires_in_days=None, revoke_reason=None, revoked_days_ago=None, ) # Belt-and-braces: explicitly set revoked_at to NULL. g.revoked_at = None await db_session.commit() resp = await client.get( "/api/v1/cmdb/overview/admin", params={"integration_id": str(integ.id)}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert resp.status_code == 200, resp.text rec = resp.json()["data"]["permissions"]["recovered"] # The unreaped row must not contribute to total. assert rec["total"] == 0 assert rec["last_30d"] == 0 assert rec["last_90d"] == 0