"""Reconciliation center contract tests (PR-E). Asserts: * RBAC: employee 403, admin 200, auditor 200 * seat lens flags under-licensed plans + 6-bucket identity breaks * ownership lens flags NULL / inactive / soft-deleted owner refs * leaver lens lists every active resource still pointing at a 'left' person across 6 resource types """ from __future__ import annotations from datetime import UTC, date, datetime from decimal import Decimal from uuid import uuid4 import pytest from app.domains.apikey.models import ApiKeyAssignment, ApiKeyProvider from app.domains.asset.models import ( AssetItem, AssetLocation, AssetLocationAdmin, AssetSoftwareCatalog, SoftwareLicensePlan, SoftwareSeatSyncStat, ) from app.domains.cmdb.models import ( CmdbCdnDomain, CmdbCloudAccessGrant, CmdbCloudScope, CmdbDnsRecord, CmdbDomain, ) from app.domains.procurement.models import ( ProcurementContract, ProcurementOrder, ProcurementSupplier, ) from app.identity.person_models import IdentityPerson from app.integrations.models import Integration def _mkperson(status: str = "active") -> IdentityPerson: return IdentityPerson( source_system="test", hris_person_id=f"P{uuid4().hex[:6]}", employee_no=f"E{uuid4().hex[:6]}", work_email=f"{uuid4().hex[:8]}@test.local", display_name=f"P {uuid4().hex[:4]}", employment_status=status, termination_date=date(2026, 1, 1) if status == "left" else None, ) # ── RBAC ───────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_overview_admin_200_all_lenses(client, admin_user): _, token = admin_user resp = await client.get( "/api/v1/admin/reconciliation/overview", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 body = resp.json()["data"] # Five lenses since provisioning-gap (2026-08-06); was # seat/ownership/leaver, then + cost. assert set(body.keys()) == { "seat", "ownership", "leaver", "cost", "provisioning_gap", } # The cost lens ran (clean or inline-error capsule, never absent). assert "findings" in body["cost"] or body["cost"].get("status") == "error" @pytest.mark.asyncio async def test_auditor_can_read(client, auditor_user): _, token = auditor_user resp = await client.get( "/api/v1/admin/reconciliation/overview", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 # ── Seat lens ─────────────────────────────────────────────────── @pytest.mark.asyncio async def test_seat_lens_flags_under_licensed_and_identity_broken( client, admin_user, db_session, ): _, token = admin_user software = AssetSoftwareCatalog( id=uuid4(), name=f"SW {uuid4().hex[:6]}", status="active", category="productivity", ) db_session.add(software) await db_session.flush() # Plan A — under_licensed (vendor=15 > max_seats=10), identity ok. plan_a = SoftwareLicensePlan( id=uuid4(), software_id=software.id, name="Premium", price=Decimal("0"), billing_cycle="monthly", currency="USD", max_seats=10, status="active", ) # Plan B — identity_broken (vendor=20 but buckets sum to 18). plan_b = SoftwareLicensePlan( id=uuid4(), software_id=software.id, name="Standard", price=Decimal("0"), billing_cycle="monthly", currency="USD", max_seats=None, status="active", ) db_session.add_all([plan_a, plan_b]) slug_a = uuid4().hex[:8] integ_a = Integration( id=uuid4(), name=f"i-{slug_a}", slug=f"i-{slug_a}", kind="software", provider="x", enabled=True, config_public={}, ) db_session.add(integ_a) await db_session.flush() db_session.add(SoftwareSeatSyncStat( id=uuid4(), integration_id=integ_a.id, software_id=software.id, plan_id=plan_a.id, provider="x", vendor_license_holders=15, holders_matched=15, holders_external=0, holders_no_email=0, holders_leaver_residual=0, holders_vendor_inactive=0, holders_off_paid_site=0, synced_at=datetime.now(UTC), )) db_session.add(SoftwareSeatSyncStat( id=uuid4(), integration_id=integ_a.id, software_id=software.id, plan_id=plan_b.id, provider="x", vendor_license_holders=20, holders_matched=18, holders_external=0, holders_no_email=0, holders_leaver_residual=0, holders_vendor_inactive=0, holders_off_paid_site=0, synced_at=datetime.now(UTC), )) await db_session.flush() resp = await client.get( "/api/v1/admin/reconciliation/seat", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 body = resp.json()["data"] under_ids = [r["plan_id"] for r in body["under_licensed"]] broken_ids = [r["plan_id"] for r in body["identity_broken"]] assert str(plan_a.id) in under_ids assert str(plan_b.id) in broken_ids # ── Ownership lens ────────────────────────────────────────────── @pytest.mark.asyncio async def test_ownership_lens_flags_inactive_and_null_owner( client, admin_user, db_session, ): _, token = admin_user # Inactive (left) person — should be flagged as inactive owner. left_person = _mkperson(status="left") db_session.add(left_person) await db_session.flush() slug = uuid4().hex[:8] integ = Integration( id=uuid4(), name=f"i-{slug}", slug=f"i-{slug}", kind="cloud", provider="aws", enabled=True, config_public={}, ) db_session.add(integ) await db_session.flush() # Scope with NULL owner — flagged 'null'. db_session.add(CmdbCloudScope( id=uuid4(), integration_id=str(integ.id), scope_type="aws_account", external_id=f"acct-{uuid4().hex[:6]}", display_name="orphan scope", owner_person_id=None, status="active", )) # Domain owned by left person — flagged 'inactive'. db_session.add(CmdbDomain( id=uuid4(), integration_id=str(integ.id), domain_name=f"d-{uuid4().hex[:6]}.example.com", owner_person_id=str(left_person.id), )) await db_session.flush() resp = await client.get( "/api/v1/admin/reconciliation/ownership", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 body = resp.json()["data"] states_scope = [r["owner_state"] for r in body["issues"]["cloud_scope"]] states_domain = [r["owner_state"] for r in body["issues"]["domain"]] assert "null" in states_scope assert "inactive" in states_domain assert body["counts"]["cloud_scope"] >= 1 assert body["counts"]["domain"] >= 1 # ── Terminal-state filters (regression: PR #621-fix) ─────────── # # Closed / decommissioned / disposed / offline / expired resources # are intentionally orphaned by their respective lifecycle flows # (decommission_scope clears the owner on close; asset disposal # preserves ``assigned_to`` for audit; domain registration lapses # in place). Reconciliation MUST NOT flag any of these as # ownership drift — they'd otherwise drown out the actually- # actionable signal. Mirrors the same filter the admin overview # health console uses (admin_overview_service.py:172). @pytest.mark.asyncio async def test_ownership_excludes_closed_and_decommissioned_scopes( client, admin_user, db_session, ): _, token = admin_user slug = uuid4().hex[:8] integ = Integration( id=uuid4(), name=f"i-{slug}", slug=f"i-{slug}", kind="cloud", provider="aws", enabled=True, config_public={}, ) db_session.add(integ) await db_session.flush() # status='closed' + decommissioned_at set, NULL owner — must NOT # appear in the ownership lens (decommission flow intentionally # orphaned this scope). closed_scope = CmdbCloudScope( id=uuid4(), integration_id=str(integ.id), scope_type="aws_account", external_id=f"acct-closed-{uuid4().hex[:6]}", display_name="closed scope", owner_person_id=None, status="closed", decommissioned_at=datetime.now(UTC), ) # Sanity control: status='active' NULL owner DOES appear. active_scope = CmdbCloudScope( id=uuid4(), integration_id=str(integ.id), scope_type="aws_account", external_id=f"acct-active-{uuid4().hex[:6]}", display_name="active orphan scope", owner_person_id=None, status="active", ) db_session.add_all([closed_scope, active_scope]) await db_session.flush() resp = await client.get( "/api/v1/admin/reconciliation/ownership", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 body = resp.json()["data"] scope_ids = {r["resource_id"] for r in body["issues"]["cloud_scope"]} assert str(closed_scope.id) not in scope_ids, ( "Closed scope must not appear in ownership lens" ) assert str(active_scope.id) in scope_ids, ( "Active orphan scope still must appear" ) @pytest.mark.asyncio async def test_ownership_excludes_terminal_resources( client, admin_user, db_session, ): """CDN offline, domain expired, asset disposed — all NULL/inactive owner — must not show up as drift.""" _, token = admin_user slug = uuid4().hex[:8] integ = Integration( id=uuid4(), name=f"i-{slug}", slug=f"i-{slug}", kind="cloud", provider="aws", enabled=True, config_public={}, ) db_session.add(integ) await db_session.flush() left_person = _mkperson(status="left") db_session.add(left_person) await db_session.flush() offline_cdn = CmdbCdnDomain( id=uuid4(), integration_id=str(integ.id), domain_name=f"cdn-off-{uuid4().hex[:6]}.example.com", status="offline", owner_person_id=str(left_person.id), ) expired_domain = CmdbDomain( id=uuid4(), integration_id=str(integ.id), domain_name=f"d-exp-{uuid4().hex[:6]}.example.com", status="expired", owner_person_id=str(left_person.id), ) disposed_asset = AssetItem( id=uuid4(), asset_tag=f"A-disp-{uuid4().hex[:6]}", asset_name="laptop", category="hardware", status="disposed", assigned_to=left_person.id, ) db_session.add_all([offline_cdn, expired_domain, disposed_asset]) await db_session.flush() resp = await client.get( "/api/v1/admin/reconciliation/ownership", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 body = resp.json()["data"] cdn_ids = {r["resource_id"] for r in body["issues"]["cdn_domain"]} domain_ids = {r["resource_id"] for r in body["issues"]["domain"]} asset_ids = {r["resource_id"] for r in body["issues"]["asset"]} assert str(offline_cdn.id) not in cdn_ids assert str(expired_domain.id) not in domain_ids assert str(disposed_asset.id) not in asset_ids @pytest.mark.asyncio async def test_leaver_residual_excludes_terminal_resources( client, admin_user, db_session, ): """A leaver whose scope is closed / asset is disposed must NOT surface — that's expected provenance, not residue.""" _, token = admin_user left = _mkperson(status="left") db_session.add(left) await db_session.flush() slug = uuid4().hex[:8] integ = Integration( id=uuid4(), name=f"i-{slug}", slug=f"i-{slug}", kind="cloud", provider="aws", enabled=True, config_public={}, ) db_session.add(integ) await db_session.flush() closed_scope = CmdbCloudScope( id=uuid4(), integration_id=str(integ.id), scope_type="aws_account", external_id=f"acct-{uuid4().hex[:6]}", display_name="left's closed scope", owner_person_id=str(left.id), status="closed", decommissioned_at=datetime.now(UTC), ) disposed_asset = AssetItem( id=uuid4(), asset_tag=f"A-{uuid4().hex[:6]}", asset_name="laptop", category="hardware", status="disposed", assigned_to=left.id, ) db_session.add_all([closed_scope, disposed_asset]) await db_session.flush() resp = await client.get( "/api/v1/admin/reconciliation/leaver", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 body = resp.json()["data"] ids = {(r["resource_type"], r["resource_id"]) for r in body["rows"]} assert ("cloud_scope_owner", str(closed_scope.id)) not in ids assert ("asset", str(disposed_asset.id)) not in ids # ── Leaver lens ───────────────────────────────────────────────── @pytest.mark.asyncio async def test_leaver_lens_lists_active_resources_owned_by_left_people( client, admin_user, db_session, ): _, token = admin_user left = _mkperson(status="left") db_session.add(left) await db_session.flush() slug = uuid4().hex[:8] integ = Integration( id=uuid4(), name=f"i-{slug}", slug=f"i-{slug}", kind="cloud", provider="aws", enabled=True, config_public={}, ) db_session.add(integ) await db_session.flush() # 1 active scope owned by left scope = CmdbCloudScope( id=uuid4(), integration_id=str(integ.id), scope_type="aws_account", external_id=f"acct-{uuid4().hex[:6]}", display_name="left's scope", owner_person_id=str(left.id), status="active", ) db_session.add(scope) await db_session.flush() # 1 active grant db_session.add(CmdbCloudAccessGrant( id=uuid4(), scope_id=str(scope.id), person_id=str(left.id), role_key="reader", status="active", granted_at=datetime.now(UTC), )) # 1 active asset in use by left db_session.add(AssetItem( id=uuid4(), asset_tag=f"A-{uuid4().hex[:6]}", asset_name="laptop", category="hardware", status="claimed", # was "in_use" — never a valid asset status (2026-08-07) assigned_to=left.id, )) # 1 active api key assigned to left provider = ApiKeyProvider(id=uuid4(), name=f"p-{uuid4().hex[:6]}") db_session.add(provider) await db_session.flush() db_session.add(ApiKeyAssignment( id=uuid4(), provider_id=provider.id, assigned_to=left.id, request_number=f"AK-{uuid4().hex[:8]}", project_name="p", reason="r", monthly_budget_usd=Decimal("0"), status="active", )) # 1 DNS record owned by left db_session.add(CmdbDnsRecord( id=uuid4(), integration_id=str(integ.id), zone_name="example.com", record_id=f"dns-{uuid4().hex[:6]}", name=f"sub-{uuid4().hex[:6]}", record_type="A", value="1.2.3.4", owner_person_id=str(left.id), )) await db_session.flush() resp = await client.get( "/api/v1/admin/reconciliation/leaver", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 body = resp.json()["data"] types = {r["resource_type"] for r in body["rows"]} assert "cloud_scope_owner" in types assert "cloud_access_grant" in types assert "asset" in types assert "api_key" in types assert "dns_record" in types assert body["persons_count"] >= 1 @pytest.mark.asyncio async def test_leaver_lens_surfaces_procurement_and_location_admin( client, admin_user, db_session, ): """A leaver who still owns an in-flight procurement contract / PO or an office-admin role must surface — these were invisible to both the offboarding scan and the prior reconciliation lenses (PR5).""" _, token = admin_user left = _mkperson(status="left") db_session.add(left) await db_session.flush() supplier = ProcurementSupplier( id=uuid4(), supplier_code=f"S-{uuid4().hex[:6]}", name=f"Supplier {uuid4().hex[:6]}", ) db_session.add(supplier) await db_session.flush() # Active contract owned by the leaver — MUST surface. db_session.add(ProcurementContract( id=uuid4(), contract_number=f"C-{uuid4().hex[:8]}", title="Active MSA", supplier_id=supplier.id, status="active", business_owner_id=left.id, )) # Terminated contract owned by the leaver — historical, MUST be excluded. terminated_id = uuid4() db_session.add(ProcurementContract( id=terminated_id, contract_number=f"C-{uuid4().hex[:8]}", title="Dead contract", supplier_id=supplier.id, status="terminated", business_owner_id=left.id, )) # Draft PO with the leaver as procurement owner — MUST surface. db_session.add(ProcurementOrder( id=uuid4(), order_number=f"PO-{uuid4().hex[:8]}", order_type="purchase", status="draft", applicant_id=left.id, procurement_owner_id=left.id, )) # Contract where the leaver is ONLY the it_contact (not an owner) — # MUST surface (all four roles are checked, not just the two owners). it_only_id = uuid4() db_session.add(ProcurementContract( id=it_only_id, contract_number=f"C-{uuid4().hex[:8]}", title="IT-contact-only", supplier_id=supplier.id, status="signed", it_contact_id=left.id, )) # PO where the leaver is ONLY the it_contact (applicant is an ACTIVE # person) — MUST surface; proves the PO block checks all four roles, not # just the two owners, and that applicant_id is not what surfaces it. active_applicant = _mkperson(status="active") db_session.add(active_applicant) await db_session.flush() po_it_only_id = uuid4() db_session.add(ProcurementOrder( id=po_it_only_id, order_number=f"PO-{uuid4().hex[:8]}", order_type="purchase", status="submitted", applicant_id=active_applicant.id, it_contact_id=left.id, )) # Asset-location administrator still held by the leaver — MUST # surface: the ``asset_admin`` role derives from this row, so the # location's coverage silently rests on someone who has left. location = AssetLocation( id=uuid4(), code=f"L{uuid4().hex[:8]}", name=f"City-{uuid4().hex[:6]}", ) # A DEACTIVATED location the leaver also administers — MUST be # excluded (matches the terminal-state exclusion every other lens # block applies). dead_location = AssetLocation( id=uuid4(), code=f"D{uuid4().hex[:8]}", name=f"Dead-{uuid4().hex[:6]}", is_active=False, ) db_session.add_all([location, dead_location]) await db_session.flush() db_session.add_all([ AssetLocationAdmin(location_id=location.id, person_id=left.id), AssetLocationAdmin(location_id=dead_location.id, person_id=left.id), ]) await db_session.flush() resp = await client.get( "/api/v1/admin/reconciliation/leaver", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200, resp.text body = resp.json()["data"] types = {r["resource_type"] for r in body["rows"]} assert "procurement_contract" in types assert "procurement_order" in types assert "location_admin" in types # The terminated contract must NOT appear. contract_ids = { r["resource_id"] for r in body["rows"] if r["resource_type"] == "procurement_contract" } assert str(terminated_id) not in contract_ids # The it_contact-only contract MUST appear (four-role coverage). assert str(it_only_id) in contract_ids # The it_contact-only PO MUST appear (PO block also checks all four roles). order_ids = { r["resource_id"] for r in body["rows"] if r["resource_type"] == "procurement_order" } assert str(po_it_only_id) in order_ids # The deactivated location's admin row MUST NOT appear. location_admin_ids = { r["resource_id"] for r in body["rows"] if r["resource_type"] == "location_admin" } assert f"{dead_location.id}:{left.id}" not in location_admin_ids assert f"{location.id}:{left.id}" in location_admin_ids