"""Identity data-quality scanner + panel API. Exercises the five reconciliation checks, the upsert/auto-resolve lifecycle across scans, and the admin panel endpoints (summary / list / decide) with RBAC. """ from __future__ import annotations import uuid import pytest from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.identity.data_quality_models import IdentityDataQualityFinding from app.identity.data_quality_service import IdentityDataQualityService from app.identity.person_models import ( IdentityExternalIdentity, IdentityPerson, ) def _person(**kw) -> IdentityPerson: base = dict( id=uuid.uuid4(), employment_status="active", source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", ) base.update(kw) return IdentityPerson(**base) @pytest.fixture async def dq_fixture(db_session: AsyncSession): """Seed one of each detectable problem + one clean active person.""" # Departed person with a live login binding → orphan_login_identity. left = _person( work_email="gone@pingcap.com", display_name="Gone Person", employment_status="left", ) db_session.add(left) await db_session.flush() db_session.add(IdentityExternalIdentity( person_id=left.id, provider="google", provider_subject="sub-gone", is_login_identity=True, identity_status="active", )) # Active real employee (has a job_title, no declared manager, no # reports) with no manager → genuine missing_manager. The detector # excludes org-apex / seeded SF service rows (job_title NULL) / # HRIS-declared-but-unlinked managers, so the fixture must model an # actionable gap to fire the check. db_session.add(_person( work_email="orphan@pingcap.com", display_name="No Manager", job_title="Engineer", )) # Active person whose manager has left → manager_departed. boss = _person( work_email="exboss@pingcap.com", display_name="Ex Boss", employment_status="left", ) db_session.add(boss) await db_session.flush() db_session.add(_person( work_email="report@pingcap.com", display_name="Has Ex Boss", manager_person_id=boss.id, )) await db_session.commit() return {"left_id": left.id} async def test_scan_detects_and_dedups(db_session, dq_fixture): svc = IdentityDataQualityService(db_session) r1 = await svc.run_scan(trigger_source="test") # orphan_login_identity(1) + missing_manager(several incl. boss/left? no: # missing_manager only counts ACTIVE persons w/o manager) + # manager_departed(1). Assert the security-critical one fired. assert r1["by_type"]["orphan_login_identity"] == 1 assert r1["by_type"]["manager_departed"] == 1 assert r1["by_type"]["missing_manager"] >= 1 assert r1["open_high"] >= 1 # orphan_login_identity is high first_new = r1["new_count"] assert first_new >= 3 # Second scan: nothing new (dedup), counts stable. r2 = await svc.run_scan(trigger_source="test") assert r2["new_count"] == 0 assert r2["open_total"] == r1["open_total"] async def test_scan_auto_resolves_disappeared(db_session, dq_fixture): svc = IdentityDataQualityService(db_session) await svc.run_scan(trigger_source="test") # Fix the orphan: disable the login binding. Next scan should # auto-resolve the orphan_login_identity finding. ext = (await db_session.execute( select(IdentityExternalIdentity).where( IdentityExternalIdentity.provider_subject == "sub-gone", ) )).scalars().one() ext.is_login_identity = False await db_session.commit() r = await svc.run_scan(trigger_source="test") assert r["by_type"]["orphan_login_identity"] == 0 resolved = (await db_session.execute( select(IdentityDataQualityFinding).where( IdentityDataQualityFinding.finding_type == "orphan_login_identity", IdentityDataQualityFinding.status == "resolved", ) )).scalars().all() assert len(resolved) == 1 assert resolved[0].auto_resolved is True async def test_scan_reopens_recurred_resolved_finding(db_session, dq_fixture): """A finding that auto-resolved and later recurs must REOPEN its row — a blind re-insert trips the (finding_type, dedup_key) UNIQUE and aborts the whole scan (prod 2026-06-11, IntegrityError 1062).""" svc = IdentityDataQualityService(db_session) await svc.run_scan(trigger_source="test") # Fix the orphan → next scan auto-resolves it. ext = (await db_session.execute( select(IdentityExternalIdentity).where( IdentityExternalIdentity.provider_subject == "sub-gone", ) )).scalars().one() ext.is_login_identity = False await db_session.commit() await svc.run_scan(trigger_source="test") # Simulate a human decision landing on the resolved row, to prove # reopen clears it. resolved = (await db_session.execute( select(IdentityDataQualityFinding).where( IdentityDataQualityFinding.finding_type == "orphan_login_identity", ) )).scalars().one() resolved_id = resolved.id resolved.decision_note = "stale note" await db_session.commit() # The problem recurs. ext.is_login_identity = True await db_session.commit() r = await svc.run_scan(trigger_source="test") # must not raise 1062 assert r["reopened_count"] == 1 assert r["new_count"] == 0 rows = (await db_session.execute( select(IdentityDataQualityFinding).where( IdentityDataQualityFinding.finding_type == "orphan_login_identity", ) )).scalars().all() assert len(rows) == 1 # same row reopened, no sibling inserted row = rows[0] assert row.id == resolved_id assert row.status == "open" assert row.auto_resolved is False assert row.resolved_at is None assert row.decision_note is None assert row.decided_at is None async def test_summary_and_list_endpoints(client, admin_headers, db_session, dq_fixture): await IdentityDataQualityService(db_session).run_scan(trigger_source="test") summ = await client.get( "/api/v1/identity/data-quality/summary", headers=admin_headers, ) assert summ.status_code == 200, summ.text sdata = summ.json()["data"] assert sdata["open_high"] >= 1 assert len(sdata["trend"]) >= 1 lst = await client.get( "/api/v1/identity/data-quality/findings?severity=high", headers=admin_headers, ) assert lst.status_code == 200 items = lst.json()["data"]["items"] assert all(i["severity"] == "high" for i in items) assert any(i["finding_type"] == "orphan_login_identity" for i in items) async def test_decide_and_rbac(client, admin_headers, employee_headers, db_session, dq_fixture): await IdentityDataQualityService(db_session).run_scan(trigger_source="test") finding = (await db_session.execute( select(IdentityDataQualityFinding).limit(1) )).scalars().one() # Employee cannot read or decide. assert (await client.get( "/api/v1/identity/data-quality/summary", headers=employee_headers, )).status_code == 403 assert (await client.patch( f"/api/v1/identity/data-quality/findings/{finding.id}", headers=employee_headers, json={"status": "acknowledged"}, )).status_code == 403 # Admin acknowledges. resp = await client.patch( f"/api/v1/identity/data-quality/findings/{finding.id}", headers=admin_headers, json={"status": "accepted_risk", "note": "known CEO no-manager"}, ) assert resp.status_code == 200, resp.text assert resp.json()["data"]["status"] == "accepted_risk" # Invalid status rejected. bad = await client.patch( f"/api/v1/identity/data-quality/findings/{finding.id}", headers=admin_headers, json={"status": "bogus"}, ) assert bad.status_code in (400, 422)