"""Person ↔ GitHub handle External API — POST /open/people/github/lookup.

The people surface gated by the standalone API-client credential
(``helix_api_``). Pins:

* scope gates the endpoint; a client without ``people.github:read`` → 403
* forward lookup by email is tri-state: found / no_github_id (person
  exists, handle not on file) / unknown_person — the first two must never
  collapse together, they mean opposite things to a reconciler
* reverse lookup by handle (the direction a GitHub org member list gives
  you) resolves case-insensitively
* department selection is a path-prefix subtree match, tenant root
  stripped; a sibling whose name merely starts with the same characters
  is NOT swallowed; unmatched paths are reported explicitly
* population ceiling: a client with ``person_visibility`` can never read
  outside it, and an out-of-ceiling person is indistinguishable from a
  non-existent one (no existence oracle)
* employment scope: currently-employed by default, ``include_left`` adds
  departed staff, ``pre_hire`` is never exposed
* egress parity with the Directory pull surface: purged / no-push(2061) /
  non-corporate-email rows never appear
* input guards: empty body → 422, over-cap selectors → 422
"""
from __future__ import annotations

import uuid

import pytest
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.api_clients.models import ApiClient
from app.api_clients.scopes import (
    SCOPE_CLOUD_SUBSCRIPTIONS_READ,
    SCOPE_PEOPLE_GITHUB_READ,
)
from app.api_clients.tokens import ApiClientTokenService
from app.identity.person_models import IdentityOrgMembership, IdentityPerson

pytestmark = pytest.mark.asyncio

URL = "/api/v1/open/people/github/lookup"


def _headers(token: str) -> dict:
    return {"Authorization": f"Bearer {token}"}


async def _mk_person(
    db: AsyncSession,
    *,
    email: str,
    name: str = "Person",
    status: str = "active",
    github_id: str | None = None,
    employee_no: str | None = None,
    dept_path: str | None = None,
    country: str | None = None,
) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        work_email=email,
        display_name=name,
        employment_status=status,
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:10]}",
        github_id=github_id,
        employee_no=employee_no,
        country=country,
    )
    db.add(p)
    await db.flush()
    if dept_path:
        leaf = dept_path.rsplit("/", 1)[-1].strip()
        source_id = f"node-{uuid.uuid4().hex[:8]}"
        db.add(IdentityOrgMembership(
            id=uuid.uuid4(),
            person_id=str(p.id),
            org_node_source_id=source_id,
            org_node_name=leaf,
            org_path=dept_path,
            is_primary=True,
        ))
        await db.flush()
    return p


async def _mk_client(
    db: AsyncSession,
    *,
    scopes: list[str] | None = None,
    person_visibility: dict | None = None,
) -> str:
    owner = await _mk_person(
        db, email=f"co-{uuid.uuid4().hex[:8]}@pingcap.com",
    )
    client = ApiClient(
        id=uuid.uuid4(),
        name="GitHub Reconciler",
        slug=f"ghrecon-{uuid.uuid4().hex[:6]}",
        owner_person_id=owner.id,
        scopes=scopes if scopes is not None else [SCOPE_PEOPLE_GITHUB_READ],
        person_visibility=person_visibility,
    )
    db.add(client)
    await db.flush()
    await db.refresh(client)
    _row, plaintext = await ApiClientTokenService(db).mint(
        client=client, name="t", expires_in_days=None, created_by=None,
    )
    await db.commit()
    return plaintext


async def test_scope_required(client, db_session):
    token = await _mk_client(db_session, scopes=[SCOPE_CLOUD_SUBSCRIPTIONS_READ])
    r = await client.post(
        URL, headers=_headers(token), json={"emails": ["a@pingcap.com"]},
    )
    assert r.status_code == 403, r.text


async def test_empty_body_rejected(client, db_session):
    """No selector = "the whole roster". Never an accident we honour."""
    token = await _mk_client(db_session)
    r = await client.post(URL, headers=_headers(token), json={})
    assert r.status_code == 422, r.text


async def test_email_lookup_tristate(client, db_session):
    await _mk_person(
        db_session, email="has@pingcap.com", name="Has Handle",
        github_id="octo-han", employee_no="P40001",
        dept_path="PingCAP / Engineering Group / TiDB",
    )
    await _mk_person(
        db_session, email="none@pingcap.com", name="No Handle",
        employee_no="P40002",
    )
    token = await _mk_client(db_session)
    r = await client.post(URL, headers=_headers(token), json={
        # Mixed case in + unknown address, to pin normalization and the
        # third state at the same time.
        "emails": ["HAS@pingcap.com", "none@pingcap.com", "ghost@pingcap.com"],
    })
    assert r.status_code == 200, r.text
    rows = {row["email"]: row for row in r.json()["data"]["emails"]}

    assert rows["has@pingcap.com"]["status"] == "found"
    person = rows["has@pingcap.com"]["person"]
    assert person["github_id"] == "octo-han"
    assert person["employee_no"] == "P40001"
    assert person["display_name"] == "Has Handle"
    # Tenant root stripped from the department path.
    assert person["department_full_path"] == "Engineering Group / TiDB"

    # "We know them, no handle on file" — NOT unknown_person.
    assert rows["none@pingcap.com"]["status"] == "no_github_id"
    assert rows["none@pingcap.com"]["person"]["github_id"] is None

    assert rows["ghost@pingcap.com"]["status"] == "unknown_person"
    assert rows["ghost@pingcap.com"]["person"] is None


async def test_reverse_handle_lookup_is_case_insensitive(client, db_session):
    """The direction a GitHub org member list actually gives you.

    The request deliberately uses a THIRD casing (``oCtOcAt``) — different
    from both the stored value and lower-case — so an implementation that
    echoed the folded key, or the stored spelling, fails here. An
    all-lowercase request would pass either way and pin nothing.
    """
    await _mk_person(
        db_session, email="rev@pingcap.com", github_id="OctoCat",
        employee_no="P40010",
    )
    token = await _mk_client(db_session)
    r = await client.post(URL, headers=_headers(token), json={
        "github_ids": ["oCtOcAt", "someone-else"],
    })
    assert r.status_code == 200, r.text
    rows = {row["github_id"]: row for row in r.json()["data"]["github_ids"]}
    # Echoed VERBATIM so the caller can join it onto their own list.
    assert set(rows) == {"oCtOcAt", "someone-else"}
    assert rows["oCtOcAt"]["status"] == "found"
    assert rows["oCtOcAt"]["person"]["work_email"] == "rev@pingcap.com"
    # …while the person payload carries the stored spelling.
    assert rows["oCtOcAt"]["person"]["github_id"] == "OctoCat"
    assert rows["someone-else"]["status"] == "unknown_handle"
    assert rows["someone-else"]["person"] is None


async def test_department_subtree_prefix_not_sibling(client, db_session):
    """Subtree yes, name-prefix sibling no.

    The sibling here shares the *requested string* as a prefix
    ("Engineering" vs "Engineering Support") — that is the only shape
    that actually exercises the trailing-slash guard. A sibling that
    merely looks similar (e.g. asking for "Engineering Group") would pass
    even with a naive ``startswith``, which is what a mutation run caught
    in the first version of this test.
    """
    await _mk_person(
        db_session, email="deep@pingcap.com", github_id="deep-gh",
        dept_path="PingCAP / Engineering / TiDB / Cloud",
    )
    await _mk_person(
        db_session, email="exact@pingcap.com", github_id="exact-gh",
        dept_path="PingCAP / Engineering",
    )
    await _mk_person(
        db_session, email="sibling@pingcap.com", github_id="sib-gh",
        dept_path="PingCAP / Engineering Support",
    )
    token = await _mk_client(db_session)
    r = await client.post(URL, headers=_headers(token), json={
        # Lower-case + stray spacing: path normalization is part of the
        # contract, not an accident of how the test spells it.
        "departments": [" engineering ", "No Such Dept"],
    })
    assert r.status_code == 200, r.text
    dept = r.json()["data"]["departments"]
    emails = {p["work_email"] for p in dept["people"]}
    assert emails == {"deep@pingcap.com", "exact@pingcap.com"}
    assert "sibling@pingcap.com" not in emails
    assert dept["unmatched"] == ["No Such Dept"]


async def test_department_listing_includes_people_without_handle(
    client, db_session,
):
    """A team audit needs the coverage gap, not just the hits."""
    await _mk_person(
        db_session, email="withgh@pingcap.com", github_id="with-gh",
        dept_path="PingCAP / Sales",
    )
    await _mk_person(
        db_session, email="withoutgh@pingcap.com",
        dept_path="PingCAP / Sales",
    )
    token = await _mk_client(db_session)
    r = await client.post(
        URL, headers=_headers(token), json={"departments": ["Sales"]},
    )
    assert r.status_code == 200, r.text
    people = {p["work_email"]: p for p in r.json()["data"]["departments"]["people"]}
    assert set(people) == {"withgh@pingcap.com", "withoutgh@pingcap.com"}
    assert people["withoutgh@pingcap.com"]["github_id"] is None


async def test_person_visibility_is_a_hard_ceiling(client, db_session):
    """A restricted client cannot read outside its population, and an
    out-of-ceiling person looks exactly like a non-existent one — the
    surface must not become an existence oracle."""
    inside = await _mk_person(
        db_session, email="inside@pingcap.com", github_id="inside-gh",
        dept_path="PingCAP / Engineering Group", country="CN",
    )
    await _mk_person(
        db_session, email="outside@pingcap.com", github_id="outside-gh",
        dept_path="PingCAP / Finance", country="CN",
    )
    token = await _mk_client(
        db_session, person_visibility={"person_ids": [str(inside.id)]},
    )

    # Named directly by email → still invisible.
    r = await client.post(URL, headers=_headers(token), json={
        "emails": ["inside@pingcap.com", "outside@pingcap.com"],
        "github_ids": ["outside-gh"],
        "departments": ["Finance"],
    })
    assert r.status_code == 200, r.text
    body = r.json()
    rows = {row["email"]: row for row in body["data"]["emails"]}
    assert rows["inside@pingcap.com"]["status"] == "found"
    # Identical to the ghost case in test_email_lookup_tristate.
    assert rows["outside@pingcap.com"]["status"] == "unknown_person"
    assert rows["outside@pingcap.com"]["person"] is None
    assert body["data"]["github_ids"][0]["status"] == "unknown_handle"
    assert body["data"]["departments"]["people"] == []
    assert body["data"]["departments"]["unmatched"] == ["Finance"]
    assert body["meta"]["visibility"] == "restricted"


async def test_include_left_cannot_widen_past_the_ceiling(client, db_session):
    """``include_left`` narrows within the ceiling; it can never reach
    outside it. An admin filter of active-only means a departed person
    stays invisible however the caller asks."""
    await _mk_person(
        db_session, email="ceil-gone@pingcap.com", github_id="ceil-gone-gh",
        status="left", country="CN",
    )
    token = await _mk_client(
        db_session,
        person_visibility={"countries": ["CN"], "employment_statuses": ["active"]},
    )
    r = await client.post(URL, headers=_headers(token), json={
        "emails": ["ceil-gone@pingcap.com"], "include_left": True,
    })
    assert r.status_code == 200, r.text
    assert r.json()["data"]["emails"][0]["status"] == "unknown_person"


async def test_unrestricted_client_reports_its_reach(client, db_session):
    await _mk_person(db_session, email="anyone@pingcap.com", github_id="any-gh")
    token = await _mk_client(db_session)
    r = await client.post(
        URL, headers=_headers(token), json={"emails": ["anyone@pingcap.com"]},
    )
    assert r.json()["meta"]["visibility"] == "unrestricted"


async def test_leavers_gated_by_include_left_and_prehire_never(
    client, db_session,
):
    await _mk_person(
        db_session, email="active@pingcap.com", github_id="act-gh",
    )
    await _mk_person(
        db_session, email="notice@pingcap.com", github_id="notice-gh",
        status="leaver_pending",
    )
    await _mk_person(
        db_session, email="gone@pingcap.com", github_id="gone-gh",
        status="left",
    )
    await _mk_person(
        db_session, email="future@pingcap.com", github_id="future-gh",
        status="pre_hire",
    )
    token = await _mk_client(db_session)
    emails = [
        "active@pingcap.com", "notice@pingcap.com",
        "gone@pingcap.com", "future@pingcap.com",
    ]

    r = await client.post(URL, headers=_headers(token), json={"emails": emails})
    rows = {x["email"]: x["status"] for x in r.json()["data"]["emails"]}
    # In notice period = still staff, still holding access.
    assert rows["active@pingcap.com"] == "found"
    assert rows["notice@pingcap.com"] == "found"
    assert rows["gone@pingcap.com"] == "unknown_person"
    # An unannounced future hire is never exposed here, on any flag.
    assert rows["future@pingcap.com"] == "unknown_person"

    r = await client.post(
        URL, headers=_headers(token),
        json={"emails": emails, "include_left": True},
    )
    rows = {x["email"]: x["status"] for x in r.json()["data"]["emails"]}
    assert rows["gone@pingcap.com"] == "found"
    assert rows["future@pingcap.com"] == "unknown_person"
    assert r.json()["meta"]["include_left"] is True


async def test_egress_gates_match_directory_surface(client, db_session):
    """Purged / no-push / non-corporate rows never egress — the same gates
    the Directory pull surface applies. The two planes must agree on who
    may leave the building."""
    await _mk_person(
        db_session, email="normal-gh@pingcap.com", github_id="normal-gh",
    )
    purged = await _mk_person(
        db_session, email="purged-gh@pingcap.com", github_id="purged-gh",
    )
    nopush = await _mk_person(
        db_session, email="nopush-gh@pingcap.com", github_id="nopush-gh",
    )
    await _mk_person(db_session, email="junk-gh@gmail.com", github_id="junk-gh")
    await db_session.execute(
        text("UPDATE identity_persons SET hris_record_state='purged' WHERE id=:i"),
        {"i": str(purged.id)},
    )
    await db_session.execute(
        text(
            "UPDATE identity_persons SET sf_downstream_push_code='2061' "
            "WHERE id=:i"
        ),
        {"i": str(nopush.id)},
    )
    await db_session.commit()

    token = await _mk_client(db_session)
    r = await client.post(URL, headers=_headers(token), json={
        "github_ids": ["normal-gh", "purged-gh", "nopush-gh", "junk-gh"],
    })
    rows = {x["github_id"]: x["status"] for x in r.json()["data"]["github_ids"]}
    assert rows["normal-gh"] == "found"
    assert rows["purged-gh"] == "unknown_handle"
    assert rows["nopush-gh"] == "unknown_handle"
    assert rows["junk-gh"] == "unknown_handle"


async def test_department_listing_order_is_stable(client, db_session):
    """A caller diffing two pulls of the same department must see real
    changes, not DB row-order churn."""
    for i in range(5):
        await _mk_person(
            db_session, email=f"stable{i}@pingcap.com", github_id=f"stable-{i}",
            dept_path="PingCAP / Ops",
        )
    token = await _mk_client(db_session)
    first = await client.post(
        URL, headers=_headers(token), json={"departments": ["Ops"]},
    )
    second = await client.post(
        URL, headers=_headers(token), json={"departments": ["Ops"]},
    )
    order_a = [p["work_email"] for p in first.json()["data"]["departments"]["people"]]
    order_b = [p["work_email"] for p in second.json()["data"]["departments"]["people"]]
    assert order_a == order_b
    assert order_a == sorted(order_a)


async def test_selector_caps_rejected(client, db_session):
    """Every selector's cap, at the boundary — 200/200/50 accepted, one
    more rejected. The github_ids cap had no test at all before."""
    token = await _mk_client(db_session)
    for field, cap in (("emails", 200), ("github_ids", 200), ("departments", 50)):
        value = (
            [f"p{i}@pingcap.com" for i in range(cap)] if field == "emails"
            else [f"{field}-{i}" for i in range(cap)]
        )
        ok = await client.post(URL, headers=_headers(token), json={field: value})
        assert ok.status_code == 200, f"{field} at cap: {ok.text}"
        over = await client.post(
            URL, headers=_headers(token), json={field: [*value, "one-too-many"]},
        )
        assert over.status_code == 422, f"{field} over cap: {over.text}"


async def test_department_result_cap_refuses_rather_than_truncates(
    client, db_session, monkeypatch,
):
    """Silently returning the first N would read as "that's the whole
    team" to a reconciler. The cap is patched down so the test doesn't
    have to create 1001 people to reach the branch."""
    from app.identity import open_people_router

    monkeypatch.setattr(open_people_router, "_MAX_PEOPLE", 2)
    for i in range(3):
        await _mk_person(
            db_session, email=f"cap{i}@pingcap.com", dept_path="PingCAP / Capped",
        )
    token = await _mk_client(db_session)
    r = await client.post(
        URL, headers=_headers(token), json={"departments": ["Capped"]},
    )
    assert r.status_code == 422, r.text
    assert "narrow" in r.text.lower()


async def test_overlapping_department_paths_are_all_reported_matched(
    client, db_session,
):
    """Asking for a parent AND its child must not report the child as
    unmatched merely because the parent matched a person first —
    ``unmatched`` means "no such department / none you can see", and a
    caller auditing per-team would treat it as a bad path."""
    await _mk_person(
        db_session, email="overlap@pingcap.com", github_id="ov-gh",
        dept_path="PingCAP / Engineering / TiDB",
    )
    token = await _mk_client(db_session)
    r = await client.post(URL, headers=_headers(token), json={
        "departments": ["Engineering", "Engineering / TiDB"],
    })
    assert r.status_code == 200, r.text
    dept = r.json()["data"]["departments"]
    assert dept["unmatched"] == []
    # Matched by two selectors, still listed once.
    assert [p["work_email"] for p in dept["people"]] == ["overlap@pingcap.com"]


async def test_department_path_spellings_and_tenant_root(client, db_session):
    """Two spellings of one path are two requested entries (echoed as
    sent), and a path copied from another Helix surface — i.e. carrying
    the tenant root — is understood rather than reported as unknown."""
    await _mk_person(
        db_session, email="spell@pingcap.com", github_id="sp-gh",
        dept_path="PingCAP / Marketing",
    )
    token = await _mk_client(db_session)
    r = await client.post(URL, headers=_headers(token), json={
        "departments": ["Marketing", " marketing ", "PingCAP / Marketing"],
    })
    assert r.status_code == 200, r.text
    dept = r.json()["data"]["departments"]
    # " marketing " is the same request as "Marketing" (input dedupe folds
    # case + trims), so it never reaches `requested`. The tenant-rooted
    # spelling is a distinct string and IS echoed.
    assert dept["requested"] == ["Marketing", "PingCAP / Marketing"]
    # Both name the same real department → neither is unmatched, and the
    # person is listed once despite matching both selectors.
    assert dept["unmatched"] == []
    assert len(dept["people"]) == 1


async def test_selector_present_but_empty_keeps_its_block(client, db_session):
    """A selector whose entries are all whitespace must still come back as
    an (empty) block: a client parsing ``data["emails"]`` should never have
    to special-case a missing key."""
    await _mk_person(
        db_session, email="shape@pingcap.com", dept_path="PingCAP / Shape",
    )
    token = await _mk_client(db_session)
    r = await client.post(URL, headers=_headers(token), json={
        "emails": ["   "], "departments": ["Shape"],
    })
    assert r.status_code == 200, r.text
    data = r.json()["data"]
    assert data["emails"] == []
    assert len(data["departments"]["people"]) == 1


async def test_two_spellings_of_one_path_both_stay_in_unmatched(
    client, db_session,
):
    """``requested`` and ``unmatched`` must never disagree in length.

    ``"Engineering"`` and ``"Engineering/"`` are two distinct requested
    strings that normalize to ONE key. Building ``unmatched`` from the
    normalized map dropped one of them; building it from what the caller
    sent keeps both. (The first version of this suite never exercised
    this — the duplicate it used was folded by input dedupe instead.)
    """
    token = await _mk_client(db_session)
    r = await client.post(URL, headers=_headers(token), json={
        "departments": ["No Such Team", "No Such Team/"],
    })
    assert r.status_code == 200, r.text
    dept = r.json()["data"]["departments"]
    assert dept["requested"] == ["No Such Team", "No Such Team/"]
    assert dept["unmatched"] == ["No Such Team", "No Such Team/"]


async def test_empty_ceiling_returns_the_same_shape(client, db_session):
    """A ceiling that resolves to nobody must answer in the same shape as
    a normal call — every selector present, every entry not-found."""
    token = await _mk_client(
        db_session, person_visibility={"countries": ["ZZ"]},
    )
    r = await client.post(URL, headers=_headers(token), json={
        "emails": ["someone@pingcap.com"],
        "github_ids": ["some-handle"],
        "departments": ["Engineering", "Engineering/"],
    })
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["data"]["emails"][0]["status"] == "unknown_person"
    assert body["data"]["github_ids"][0]["status"] == "unknown_handle"
    dept = body["data"]["departments"]
    # Two distinct requested strings collapsing to one normalized key —
    # both must survive into unmatched here too, so this path and the
    # normal one answer identically.
    assert dept["requested"] == ["Engineering", "Engineering/"]
    assert dept["unmatched"] == ["Engineering", "Engineering/"]
    assert dept["people"] == []
    assert body["meta"]["visibility"] == "restricted"
    assert body["meta"]["person_count"] == 0


async def test_malformed_ceiling_fails_closed(client, db_session):
    """A stored ceiling that isn't a JSON object cannot be evaluated, and
    "cannot evaluate" must mean NOBODY — never the ``None`` meaning of
    unrestricted, which would hand a narrowed client the whole roster."""
    await _mk_person(
        db_session, email="fc@pingcap.com", github_id="fc-gh",
    )
    token = await _mk_client(db_session)
    # Bypass the schema validator the way a hand-edited row or an older
    # stored shape would.
    await db_session.execute(
        text(
            "UPDATE api_clients SET person_visibility = :v "
            "WHERE slug LIKE 'ghrecon-%'"
        ),
        {"v": '"restricted-to-nobody-but-not-an-object"'},
    )
    await db_session.commit()

    r = await client.post(
        URL, headers=_headers(token), json={"emails": ["fc@pingcap.com"]},
    )
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["data"]["emails"][0]["status"] == "unknown_person"
    assert body["meta"]["visibility"] == "restricted"


async def test_restricted_ceiling_without_statuses_keeps_notice_period(
    client, db_session,
):
    """A departments/countries-only ceiling must not silently inherit the
    filter engine's active-only pre-gate: that would drop notice-period
    staff — exactly the people an access audit is chasing — while the
    endpoint documents them as part of the default population, and would
    make ``include_left`` a dead flag for such clients."""
    await _mk_person(
        db_session, email="cn-active@pingcap.com", github_id="cn-a-gh",
        country="CN",
    )
    await _mk_person(
        db_session, email="cn-notice@pingcap.com", github_id="cn-n-gh",
        status="leaver_pending", country="CN",
    )
    await _mk_person(
        db_session, email="cn-gone@pingcap.com", github_id="cn-g-gh",
        status="left", country="CN",
    )
    token = await _mk_client(db_session, person_visibility={"countries": ["CN"]})
    emails = [
        "cn-active@pingcap.com", "cn-notice@pingcap.com", "cn-gone@pingcap.com",
    ]

    r = await client.post(URL, headers=_headers(token), json={"emails": emails})
    rows = {x["email"]: x["status"] for x in r.json()["data"]["emails"]}
    assert rows["cn-active@pingcap.com"] == "found"
    assert rows["cn-notice@pingcap.com"] == "found"
    assert rows["cn-gone@pingcap.com"] == "unknown_person"

    # …and include_left still works for such a client.
    r = await client.post(
        URL, headers=_headers(token),
        json={"emails": emails, "include_left": True},
    )
    rows = {x["email"]: x["status"] for x in r.json()["data"]["emails"]}
    assert rows["cn-gone@pingcap.com"] == "found"


async def test_whitespace_only_handle_is_a_coverage_gap(client, db_session):
    """A stored handle of blanks is "not on file", not a handle: reporting
    ``found`` with an unusable value would send a reconciler looking for a
    GitHub user that cannot exist."""
    await _mk_person(
        db_session, email="blank@pingcap.com", github_id="   ",
    )
    await _mk_person(
        db_session, email="padded@pingcap.com", github_id=" padded-gh ",
    )
    token = await _mk_client(db_session)
    r = await client.post(URL, headers=_headers(token), json={
        "emails": ["blank@pingcap.com", "padded@pingcap.com"],
    })
    rows = {x["email"]: x for x in r.json()["data"]["emails"]}
    assert rows["blank@pingcap.com"]["status"] == "no_github_id"
    assert rows["blank@pingcap.com"]["person"]["github_id"] is None
    # A padded handle is trimmed on the way out — the raw value would not
    # match anything on GitHub.
    assert rows["padded@pingcap.com"]["status"] == "found"
    assert rows["padded@pingcap.com"]["person"]["github_id"] == "padded-gh"


async def test_handle_collision_prefers_the_employed_person(client, db_session):
    """Two rows sharing a handle: the answer must describe the live one,
    and must not depend on which row the DB returned first."""
    await _mk_person(
        db_session, email="dup-gone@pingcap.com", github_id="dup-gh",
        status="left", employee_no="P00001",
    )
    await _mk_person(
        db_session, email="dup-live@pingcap.com", github_id="dup-gh",
        employee_no="P99998",
    )
    token = await _mk_client(db_session)
    r = await client.post(URL, headers=_headers(token), json={
        "github_ids": ["dup-gh"], "include_left": True,
    })
    assert r.status_code == 200, r.text
    row = r.json()["data"]["github_ids"][0]
    assert row["status"] == "found"
    assert row["person"]["work_email"] == "dup-live@pingcap.com"


async def test_no_contact_details_on_the_wire(client, db_session):
    """This scope is a mapping table, not a directory export."""
    p = await _mk_person(
        db_session, email="priv@pingcap.com", github_id="priv-gh",
    )
    p.mobile_phone = "+8613800000000"
    p.personal_email = "private@gmail.com"
    await db_session.commit()

    token = await _mk_client(db_session)
    r = await client.post(
        URL, headers=_headers(token), json={"emails": ["priv@pingcap.com"]},
    )
    assert r.status_code == 200
    assert "+8613800000000" not in r.text
    assert "private@gmail.com" not in r.text
    assert "mobile_phone" not in r.text


def test_new_scope_is_registered_in_the_catalog():
    from app.api_clients.scopes import KNOWN_SCOPES, catalog_payload, validate_scopes

    assert SCOPE_PEOPLE_GITHUB_READ in KNOWN_SCOPES
    assert validate_scopes([SCOPE_PEOPLE_GITHUB_READ]) == [SCOPE_PEOPLE_GITHUB_READ]
    entry = next(
        row for row in catalog_payload() if row["key"] == SCOPE_PEOPLE_GITHUB_READ
    )
    # desc_id is the i18n key suffix the admin console renders
    # (apiClients.scopeDesc.<desc_id>) — renaming it silently blanks the
    # scope description in the grant editor.
    assert entry == {
        "key": "people.github:read",
        "desc_id": "people_github_read",
        "kind": "read",
    }


def test_endpoint_is_in_the_curated_openapi_spec():
    """Integrators import this spec into Postman / codegen — an endpoint
    missing from it is invisible to them however well it works."""
    from app.api_clients.open_meta_router import _curated_spec

    assert "/api/v1/open/people/github/lookup" in _curated_spec()["paths"]


async def test_person_without_department_still_resolves(client, db_session):
    """A person with no primary org membership resolves with a null path,
    rather than dropping out of an email lookup entirely."""
    await _mk_person(
        db_session, email="nodept@pingcap.com", github_id="nodept-gh",
    )
    await db_session.commit()
    token = await _mk_client(db_session)
    r = await client.post(
        URL, headers=_headers(token), json={"emails": ["nodept@pingcap.com"]},
    )
    assert r.status_code == 200, r.text
    row = r.json()["data"]["emails"][0]
    assert row["status"] == "found"
    assert row["person"]["department_full_path"] is None