"""Leave-health insights — utilisation roll-ups by BU / department / lead. Guards the counter-intuitive analytics surface (:mod:`app.domains.hr.leave_insights_service`) and its endpoints / cron: 1. Per-person utilisation = used / entitled over the *current* annual buckets only; comp banked + legal forfeit-at-risk tracked separately; sick excluded; no-quota people are "no data" (utilisation None). 2. Company baseline and relative health bands (a team well below the company utilisation is ``attention``; at/above is ``healthy``). 3. Lead dimension rolls up the FULL reporting subtree (skip-level). 4. Small-group privacy guard → ``insufficient_data``, kept out of ranking. 5. Endpoints enforce hr-read scope (employee = 403) and the snapshot cron writes one idempotent row set per day. """ from __future__ import annotations from datetime import date from uuid import uuid4 import pytest import pytest_asyncio from sqlalchemy import select from app.domains.hr.leave_insights_models import HrLeaveInsightSnapshot from app.domains.hr.leave_insights_service import ( DIMENSION_BU, DIMENSION_DEPARTMENT, DIMENSION_LEAD, HEALTH_ATTENTION, HEALTH_HEALTHY, HEALTH_INSUFFICIENT, HEALTH_UNCOVERED, LeaveInsightsService, fiscal_year_elapsed_fraction, ) from app.domains.hr.leave_models import HrLeaveAccount from app.identity.person_models import IdentityOrgMembership, IdentityPerson async def _person( db, *, name, bu, dept, manager_id=None, status="active", ): pid = uuid4() db.add(IdentityPerson( id=pid, work_email=f"{name}-{pid}@test.com", display_name=name, employment_status=status, source_system="successfactors", hris_person_id=str(pid)[:8], manager_person_id=manager_id, )) db.add(IdentityOrgMembership( person_id=pid, source_system="successfactors", org_node_source_id=f"node-{dept}-{pid}", org_node_type="department", org_node_name=dept, org_path=f"PingCAP / {bu} / {dept}", is_primary=True, membership_status="active", )) return pid async def _annual(db, pid, *, entitled, used, atype="CHN_AnnualLeave", is_current=True): bal = entitled - used db.add(HrLeaveAccount( external_code=str(uuid4()), person_id=str(pid), account_type=atype, account_type_label=atype, fiscal_year=2026, start_date=date(2026, 4, 1), end_date=date(2027, 3, 31), entitled_days=entitled, used_days=used, balance_days=bal, is_current=is_current, )) async def _comp(db, pid, *, balance): db.add(HrLeaveAccount( external_code=str(uuid4()), person_id=str(pid), account_type="CHN_COMPL", fiscal_year=2026, start_date=date(2026, 4, 1), end_date=date(2026, 4, 30), entitled_days=balance, used_days=0, balance_days=balance, is_current=False, account_closed=False, )) @pytest_asyncio.fixture async def org(db_session): """A small org: Lead L (BU Alpha / team Core) ├ A1 entitled 10 used 5 (50%) ├ A2 entitled 10 used 1 (10%) └ Sub-lead M (team Core) ├ B1 entitled 10 used 0 (0%) + comp banked 3 └ B2 entitled 10 used 0 (0%) Lead H (BU Beta / team Rocket) ├ C1 entitled 10 used 8 ├ C2 entitled 10 used 9 └ C3 entitled 10 used 7 Solo S (BU Gamma / team Tiny) entitled 10 used 0 (small group) NoData N (BU Alpha / team Core) — no annual account """ db = db_session L = await _person(db, name="LeadL", bu="Alpha", dept="Core") M = await _person(db, name="SubM", bu="Alpha", dept="Core", manager_id=L) A1 = await _person(db, name="A1", bu="Alpha", dept="Core", manager_id=L) A2 = await _person(db, name="A2", bu="Alpha", dept="Core", manager_id=L) B1 = await _person(db, name="B1", bu="Alpha", dept="Core", manager_id=M) B2 = await _person(db, name="B2", bu="Alpha", dept="Core", manager_id=M) H = await _person(db, name="LeadH", bu="Beta", dept="Rocket") C1 = await _person(db, name="C1", bu="Beta", dept="Rocket", manager_id=H) C2 = await _person(db, name="C2", bu="Beta", dept="Rocket", manager_id=H) C3 = await _person(db, name="C3", bu="Beta", dept="Rocket", manager_id=H) S = await _person(db, name="Solo", bu="Gamma", dept="Tiny") N = await _person(db, name="NoData", bu="Alpha", dept="Core", manager_id=L) await _annual(db, L, entitled=10, used=2) await _annual(db, M, entitled=10, used=1) await _annual(db, A1, entitled=10, used=5) await _annual(db, A2, entitled=10, used=1) await _annual(db, B1, entitled=10, used=0) await _comp(db, B1, balance=3) await _annual(db, B2, entitled=10, used=0) await _annual(db, H, entitled=10, used=6) await _annual(db, C1, entitled=10, used=8) await _annual(db, C2, entitled=10, used=9) await _annual(db, C3, entitled=10, used=7) await _annual(db, S, entitled=10, used=0) # N: no annual account, but add a sick bucket that must be ignored. await _annual(db, N, entitled=10, used=10, atype="CHN_SickLeave") await db.flush() await db.commit() return { "L": L, "M": M, "A1": A1, "A2": A2, "B1": B1, "B2": B2, "H": H, "C1": C1, "C2": C2, "C3": C3, "S": S, "N": N, } @pytest.mark.asyncio async def test_company_baseline_and_per_person(db_session, org): svc = LeaveInsightsService(db_session) ctx = await svc.build_context(as_of=date(2026, 6, 29)) by = ctx.by_person # N has only a sick bucket → no quota, utilisation None, excluded from rate. assert by[str(org["N"])].has_quota is False assert by[str(org["N"])].utilisation is None # A1 = 5/10 assert by[str(org["A1"])].utilisation == 0.5 # comp banked surfaces on B1 only. assert by[str(org["B1"])].comp_balance == 3.0 # forfeit-at-risk = legal annual balance (10-0=10 for B1). assert by[str(org["B1"])].forfeit_at_risk == 10.0 # Company: sum used / sum entitled over the 11 covered people (N excluded). # used = 2+1+5+1+0+0+6+8+9+7+0 = 39 ; entitled = 11*10 = 110 assert ctx.company_entitled_total == 110.0 assert ctx.company_used_total == 39.0 assert ctx.company_utilisation == pytest.approx(0.3545, abs=1e-3) @pytest.mark.asyncio async def test_bu_health_relative(db_session, org): svc = LeaveInsightsService(db_session) ctx = await svc.build_context(as_of=date(2026, 6, 29)) rows = {g.key: g for g in svc.group(DIMENSION_BU, ctx)} # Alpha: used 2+1+5+1+0+0 = 9 / 60 = 0.15 → well below company 0.35 → attention alpha = rows["Alpha"] assert alpha.covered_count == 6 assert alpha.utilisation == pytest.approx(0.15, abs=1e-3) assert alpha.health == HEALTH_ATTENTION # Beta: H(6)+C1(8)+C2(9)+C3(7) = 30 / 40 = 0.75 → above company → healthy beta = rows["Beta"] assert beta.covered_count == 4 assert beta.utilisation == pytest.approx(0.75, abs=1e-3) assert beta.health == HEALTH_HEALTHY # Gamma: only Solo (1 covered) → below min_group_size → insufficient_data assert rows["Gamma"].health == HEALTH_INSUFFICIENT @pytest.mark.asyncio async def test_lead_full_subtree(db_session, org): svc = LeaveInsightsService(db_session) ctx = await svc.build_context(as_of=date(2026, 6, 29)) rows = {g.key: g for g in svc.group(DIMENSION_LEAD, ctx)} # Lead L's subtree = M, A1, A2, B1, B2, N (skip-level through M). N has no # quota so covered = 5; used = 1+5+1+0+0 = 7 / 50 = 0.14. lead_l = rows[str(org["L"])] assert lead_l.member_count == 6 assert lead_l.covered_count == 5 assert lead_l.utilisation == pytest.approx(0.14, abs=1e-3) assert lead_l.comp_banked_total == 3.0 assert lead_l.lead_person_id == str(org["L"]) # Sub-lead M's subtree = B1, B2 only. sub_m = rows[str(org["M"])] assert sub_m.member_count == 2 assert sub_m.utilisation == 0.0 @pytest.mark.asyncio async def test_overview_endpoint_rbac(client, org, hr_headers, employee_headers): # employee forbidden r = await client.get( "/api/v1/hr/leave/insights/overview", headers=employee_headers, ) assert r.status_code == 403 r = await client.get( "/api/v1/hr/leave/insights/overview?dimension=lead", headers=hr_headers, ) assert r.status_code == 200 data = r.json()["data"] assert data["baseline"]["active_population"] >= 11 assert "attention_groups" in data assert data["baseline"]["coverage_pct"] is not None @pytest.mark.asyncio async def test_groups_and_members_endpoints(db_session, client, org, hr_headers): r = await client.get( "/api/v1/hr/leave/insights/groups?dimension=bu", headers=hr_headers, ) assert r.status_code == 200 groups = r.json()["data"]["groups"] # ranked ascending: Alpha (0.15) before Beta (0.8) eligible = [g for g in groups if g["health"] != HEALTH_INSUFFICIENT] assert eligible[0]["key"] == "Alpha" r = await client.get( "/api/v1/hr/leave/insights/members?dimension=bu&key=Alpha", headers=hr_headers, ) assert r.status_code == 200 members = r.json()["data"]["members"] # Alpha/Core = L, M, A1, A2, B1, B2, N (7); lowest utilisation first. assert len(members) == 7 assert members[0]["utilisation"] == 0.0 # N (no quota) sorts last. assert members[-1]["has_quota"] is False @pytest.mark.asyncio async def test_uncovered_distinct_from_insufficient_and_privacy_guard(db_session): """A team where NOBODY has a leave account (e.g. an overseas team — SF EC Time-Off only covers the CN entity) is `uncovered`, NOT `insufficient_data`, and the drill-down privacy guard refuses to expose its individuals.""" db = db_session lead = await _person(db, name="OverseasLead", bu="Overseas", dept="US Team") await _person(db, name="OA", bu="Overseas", dept="US Team", manager_id=lead) await _person(db, name="OB", bu="Overseas", dept="US Team", manager_id=lead) # No _annual()/_comp() — these people have zero leave accounts. await db.flush() await db.commit() svc = LeaveInsightsService(db_session) ctx = await svc.build_context(as_of=date(2026, 6, 29)) rows = {g.key: g for g in svc.group(DIMENSION_BU, ctx)} assert rows["Overseas"].health == HEALTH_UNCOVERED assert rows["Overseas"].covered_count == 0 # Redacted serialisation: no leave metrics leak for an uncovered group. d = rows["Overseas"].as_dict() assert d["utilisation"] is None assert d["annual_entitled_total"] is None # Privacy guard blocks the member drill-down. members = svc.members_of(DIMENSION_BU, "Overseas", ctx) assert svc.below_privacy_guard(members, ctx) is True @pytest.mark.asyncio async def test_members_endpoint_blocks_small_group(client, org, hr_headers): # Gamma = Solo only (1 covered) → below the min-group guard → 4xx, never # returns the individual's row. r = await client.get( "/api/v1/hr/leave/insights/members?dimension=bu&key=Gamma", headers=hr_headers, ) assert r.status_code >= 400 assert r.status_code != 500 def test_pace_index_none_early_in_fiscal_year(): # Day 1 of the FY: elapsed ≈ 0.003 — the pace index must be None, not a # ~300x artifact from dividing by a near-zero elapsed fraction. frac = fiscal_year_elapsed_fraction(2026, 4, date(2026, 4, 1)) assert frac < 0.02 from app.domains.hr.leave_insights_service import _pace assert _pace(0.5, frac) is None # Mid-year it is a real number. assert _pace(0.25, 0.25) == 1.0 @pytest.mark.asyncio async def test_snapshot_cron_idempotent(db_session, org): from app.scheduled_jobs.hr_leave_insights_handlers import ( run_hr_leave_insight_snapshot, ) out1 = await run_hr_leave_insight_snapshot(db_session, {}) assert out1["rows_written"] > 0 rows1 = (await db_session.execute( select(HrLeaveInsightSnapshot) )).scalars().all() n1 = len(rows1) # Re-run same day → wholesale replace, no duplication. out2 = await run_hr_leave_insight_snapshot(db_session, {}) rows2 = (await db_session.execute( select(HrLeaveInsightSnapshot) )).scalars().all() assert len(rows2) == n1 == out2["rows_written"] # A company baseline row exists. assert any(r.dimension == "company" for r in rows2) @pytest.mark.asyncio async def test_case_whitespace_variant_departments_fold_into_one_group(db_session): """HRIS spells the same team inconsistently ('PreSales' / 'Presales ' / 'presales'). ``org_node_name`` and the snapshot ``group_key`` are ``utf8mb4_unicode_ci`` (case- + trailing-space-insensitive), so two raw spellings are ONE row of the snapshot unique index. Grouping must fold them into a single group, otherwise the cron emits two department rows that 1062-clash and aborts the whole day's snapshot. Regression for that crash.""" db = db_session lead = await _person(db, name="PSLead", bu="Sales", dept="PreSales") a = await _person(db, name="PSA", bu="Sales", dept="Presales ", manager_id=lead) b = await _person(db, name="PSB", bu="Sales", dept="presales", manager_id=lead) await _annual(db, lead, entitled=10, used=3) await _annual(db, a, entitled=10, used=2) await _annual(db, b, entitled=10, used=1) await db.flush() await db.commit() svc = LeaveInsightsService(db_session) ctx = await svc.build_context(as_of=date(2026, 6, 29)) # All three spellings collapse into ONE department group. dept_groups = svc.group(DIMENSION_DEPARTMENT, ctx) presales = [g for g in dept_groups if svc.canon_group_key(g.key) == "presales"] assert len(presales) == 1 grp = presales[0] assert grp.covered_count == 3 assert grp.member_count == 3 # Label is the most common original spelling (here a 3-way tie → alphabetical). assert grp.label in {"PreSales", "Presales", "presales"} # Drill-down resolves every member regardless of the spelling echoed back. for echoed in ("PreSales", "presales", "Presales "): members = svc.members_of(DIMENSION_DEPARTMENT, echoed, ctx) assert len(members) == 3 # The cron persists exactly one department row for the folded group and # never raises the 1062 IntegrityError. from app.scheduled_jobs.hr_leave_insights_handlers import ( run_hr_leave_insight_snapshot, ) out = await run_hr_leave_insight_snapshot(db_session, {}) assert out["rows_written"] > 0 dept_rows = (await db_session.execute( select(HrLeaveInsightSnapshot).where( HrLeaveInsightSnapshot.dimension == "department", ) )).scalars().all() presales_rows = [ r for r in dept_rows if svc.canon_group_key(r.group_key) == "presales" ] assert len(presales_rows) == 1