"""Owner-label → real-person resolution (email hint + finance-role name match).""" from __future__ import annotations import uuid import pytest from sqlalchemy import text from app.core.coordinators import finance_close_coordinator as coord async def _make_person(db, *, email, display=None, first=None, last=None, finance=False): from app.identity.person_models import IdentityPerson pid = uuid.uuid4() db.add(IdentityPerson( id=pid, hris_person_id=f"sf-{email.split('@')[0]}", work_email=email, display_name=display, first_name=first, last_name=last, )) await db.flush() if finance: from app.auth.models import AuthUserRole rid = (await db.execute(text("SELECT id FROM auth_roles WHERE name='finance'"))).scalar() db.add(AuthUserRole(user_id=pid, role_id=uuid.UUID(str(rid)))) await db.flush() return str(pid) @pytest.mark.asyncio async def test_email_hint_and_finance_name_match(db_session): cici = await _make_person(db_session, email="chenchao@pingcap.com", display="Chen Chao", finance=True) grace = await _make_person(db_session, email="grace.x@pingcap.com", display="Grace", finance=True) # A non-finance "Eric" must NOT be matched (scoped to finance role). await _make_person(db_session, email="eric.outsider@pingcap.com", first="Eric", finance=False) resolved = await coord.resolve_owner_person_ids( db_session, labels={"Cici", "Grace", "Eric", "Huan"}, email_hints={"Cici": "chenchao@pingcap.com", "Huan": "huan.sun@pingcap.com"}, ) assert resolved["Cici"] == cici # email hint assert resolved["Grace"] == grace # finance-role name match assert "Eric" not in resolved # only a non-finance Eric exists → unmatched assert "Huan" not in resolved # hinted email has no person → unmatched @pytest.mark.asyncio async def test_ambiguous_name_match_left_unresolved(db_session): await _make_person(db_session, email="grace.a@pingcap.com", display="Grace", finance=True) await _make_person(db_session, email="grace.b@pingcap.com", display="Grace", finance=True) resolved = await coord.resolve_owner_person_ids( db_session, labels={"Grace"}, email_hints={}) assert "Grace" not in resolved # 2 finance Graces → not unique → unresolved (proof, not pattern) @pytest.mark.asyncio async def test_resolved_owner_shows_real_name_and_backfills_instances(client, finance_headers, db_session): # Seed a finance person named Grace, then instantiate — Grace-owned tasks # resolve to her and the API surfaces her real display name. await _make_person(db_session, email="grace.real@pingcap.com", display="Grace Liang", first="Grace", last="Liang", finance=True) await db_session.commit() await client.post("/api/v1/finance/close/runs/2026-06/instantiate", headers=finance_headers) r = await client.get("/api/v1/finance/close/runs/2026-06/tasks?process=COGS", headers=finance_headers) grace_tasks = [t for t in r.json()["data"]["tasks"] if t["owner_label"] == "Grace"] assert grace_tasks assert all(t["owner_person_id"] for t in grace_tasks) # resolved + backfilled assert all(t["owner_name"] == "Grace Liang" for t in grace_tasks) # real name surfaced