"""Tests for auth APIs: Google SSO verify -> JWT -> /auth/me"""

import uuid

import pytest
import respx
from httpx import Response

from app.core.config import settings
from app.identity.person_models import IdentityPerson


@pytest.mark.asyncio
async def test_google_verify_and_me(client, db_session):
    """Google id_token verify -> JWT -> /auth/me returns correct user."""
    # First create a user in DB (simulating HRIS sync)
    import uuid

    from sqlalchemy import text

    from app.auth.models import AuthUserRole
    from app.identity.person_models import IdentityPerson

    user_id = uuid.uuid4()
    user = IdentityPerson(
        id=user_id,
        work_email="testuser@company.com",
        display_name="Test User",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="sf-test-001",
    )
    db_session.add(user)
    await db_session.flush()

    role_result = await db_session.execute(
        text("SELECT id FROM auth_roles WHERE name = 'employee'")
    )
    role_id = role_result.scalar_one()
    if isinstance(role_id, str):
        role_id = uuid.UUID(role_id)
    db_session.add(AuthUserRole(user_id=user_id, role_id=role_id))
    await db_session.flush()

    # Mock Google token verification
    with respx.mock:
        respx.get("https://oauth2.googleapis.com/tokeninfo").mock(
            return_value=Response(
                200,
                json={
                    "sub": "google-sub-123",
                    "email": "testuser@company.com",
                    "email_verified": "true",
                    "name": "Test User",
                    "picture": "https://example.com/photo.jpg",
                    "aud": settings.GOOGLE_CLIENT_ID,
                    "iss": "accounts.google.com",
                },
            )
        )

        # Call google verify
        resp = await client.post(
            "/api/v1/auth/google/verify",
            json={"id_token": "fake-google-token"},
        )

    assert resp.status_code == 200
    data = resp.json()
    assert data["success"] is True
    assert "access_token" in data["data"]
    assert data["data"]["user"]["email"] == "testuser@company.com"

    # Use the access token to call /auth/me
    access_token = data["data"]["access_token"]
    me_resp = await client.get(
        "/api/v1/auth/me",
        headers={"Authorization": f"Bearer {access_token}"},
    )
    assert me_resp.status_code == 200
    me_data = me_resp.json()
    assert me_data["success"] is True
    assert me_data["data"]["email"] == "testuser@company.com"
    assert "employee" in me_data["data"]["roles"]


@pytest.mark.asyncio
async def test_google_verify_user_not_found(client):
    """Google verify fails if user not in DB (not synced from HRIS)."""
    with respx.mock:
        respx.get("https://oauth2.googleapis.com/tokeninfo").mock(
            return_value=Response(
                200,
                json={
                    "sub": "google-sub-unknown",
                    "email": "unknown@company.com",
                    "email_verified": "true",
                    "aud": settings.GOOGLE_CLIENT_ID,
                    "iss": "accounts.google.com",
                },
            )
        )

        resp = await client.post(
            "/api/v1/auth/google/verify",
            json={"id_token": "fake-token"},
        )

    assert resp.status_code == 401
    assert resp.json()["success"] is False


@pytest.mark.asyncio
async def test_me_without_token(client):
    """GET /auth/me without token returns 401."""
    resp = await client.get("/api/v1/auth/me")
    assert resp.status_code == 401


@pytest.mark.asyncio
async def test_me_with_invalid_token(client):
    """GET /auth/me with invalid token returns 401."""
    resp = await client.get(
        "/api/v1/auth/me",
        headers={"Authorization": "Bearer invalid-token"},
    )
    assert resp.status_code == 401


@pytest.mark.asyncio
async def test_logout(client, admin_user):
    """POST /auth/logout clears session."""
    user, token = admin_user
    resp = await client.post(
        "/api/v1/auth/logout",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200
    assert resp.json()["success"] is True



@pytest.mark.asyncio
async def test_refresh_storm_does_not_lock_login(client, db_session, monkeypatch):
    """Regression for 2026-05-19 prod incident.

    Before fix: ``/auth/refresh`` and ``/auth/google/verify`` shared one
    per-IP rate-limit bucket (20 / 15min). A user whose refresh token
    expired and reloaded the dashboard a few times could fan out enough
    parallel 401-triggered refresh calls to exhaust the bucket, then
    received 429 on every subsequent Google sign-in attempt for the rest
    of the window. After fix: ``/auth/refresh`` has its own wider bucket.
    """
    monkeypatch.setattr(settings, "LOGIN_RATE_LIMIT_MAX", 5)
    monkeypatch.setattr(settings, "REFRESH_RATE_LIMIT_MAX", 50)

    user_id = uuid.uuid4()
    user = IdentityPerson(
        id=user_id,
        work_email="refreshstorm@company.com",
        display_name="Refresh Storm",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="sf-refresh-storm",
    )
    db_session.add(user)
    await db_session.flush()

    from sqlalchemy import text

    from app.auth.models import AuthUserRole

    role_result = await db_session.execute(
        text("SELECT id FROM auth_roles WHERE name = 'employee'")
    )
    role_id = role_result.scalar_one()
    if isinstance(role_id, str):
        role_id = uuid.UUID(role_id)
    db_session.add(AuthUserRole(user_id=user_id, role_id=role_id))
    await db_session.flush()
    await db_session.commit()

    # Burn 30 refresh attempts (well past the old 5-attempt shared cap).
    # Each request 401s because there's no refresh cookie — that's fine,
    # the rate limiter ticks on every request regardless of auth outcome.
    for _ in range(30):
        r = await client.post("/api/v1/auth/refresh")
        assert r.status_code in (401, 200), (
            f"refresh returned {r.status_code} during burst — should 401, not 429"
        )

    # Google verify on the same IP must still go through.
    with respx.mock:
        respx.get("https://oauth2.googleapis.com/tokeninfo").mock(
            return_value=Response(
                200,
                json={
                    "sub": "google-sub-refresh-storm",
                    "email": "refreshstorm@company.com",
                    "email_verified": "true",
                    "name": "Refresh Storm",
                    "aud": settings.GOOGLE_CLIENT_ID,
                    "iss": "accounts.google.com",
                },
            )
        )
        login = await client.post(
            "/api/v1/auth/google/verify",
            json={"id_token": "fake-token"},
        )

    assert login.status_code == 200, (
        f"google/verify returned {login.status_code} after a refresh burst — "
        "refresh and login buckets are not separated"
    )


async def _seed_user_with_refresh_token(db_session, email: str, hris_id: str):
    """Helper: insert an active person + employee-role row, mint a
    refresh token, plant the JTI in Redis. Returns (user, refresh_token).
    """
    from sqlalchemy import text

    from app.auth.models import AuthUserRole
    from app.auth.service import _store_refresh_jti
    from app.core.security import create_refresh_token

    user_id = uuid.uuid4()
    user = IdentityPerson(
        id=user_id,
        work_email=email,
        display_name=hris_id,
        employment_status="active",
        source_system="successfactors",
        hris_person_id=hris_id,
    )
    db_session.add(user)
    await db_session.flush()
    role_id = (
        await db_session.execute(text("SELECT id FROM auth_roles WHERE name = 'employee'"))
    ).scalar_one()
    if isinstance(role_id, str):
        role_id = uuid.UUID(role_id)
    db_session.add(AuthUserRole(user_id=user_id, role_id=role_id))
    await db_session.flush()
    await db_session.commit()

    refresh_token = create_refresh_token(str(user_id))
    await _store_refresh_jti(refresh_token, str(user_id))
    return user, refresh_token


@pytest.mark.asyncio
async def test_consume_refresh_jti_atomically_stamps_grace():
    """Unit-level guarantee for the 2026-06-02 hardening: a single atomic
    ``_consume_refresh_jti`` deletes the JTI AND stamps the grace marker,
    so the loser of a concurrent rotation is guaranteed to find the
    marker the instant it sees the JTI consumed — no poll, no window.

    First call: wins (returns owner), JTI gone, grace marker == owner.
    Second call: loses (returns None), grace marker still present.
    """
    from app.auth.service import (
        _REDIS_JTI_PREFIX,
        _REDIS_REFRESH_GRACE_PREFIX,
        _consume_refresh_jti,
        _store_refresh_jti,
    )
    from app.core.redis_client import get_redis
    from app.core.security import create_refresh_token, decode_token

    uid = str(uuid.uuid4())
    rt = create_refresh_token(uid)
    jti = decode_token(rt)["jti"]
    await _store_refresh_jti(rt, uid)

    redis = get_redis()

    first = await _consume_refresh_jti(jti)
    assert first == uid, "winning consume must return the JTI owner id"
    assert await redis.get(f"{_REDIS_JTI_PREFIX}{jti}") is None, (
        "JTI must be deleted (one-time use) by the atomic consume"
    )
    marker = await redis.get(f"{_REDIS_REFRESH_GRACE_PREFIX}{jti}")
    if isinstance(marker, bytes):
        marker = marker.decode("utf-8")
    assert marker == uid, (
        "grace marker must be stamped atomically with the consume — this "
        "is what makes the loser's read deterministic"
    )

    second = await _consume_refresh_jti(jti)
    assert second is None, "second consume of the same JTI must lose"
    marker2 = await redis.get(f"{_REDIS_REFRESH_GRACE_PREFIX}{jti}")
    if isinstance(marker2, bytes):
        marker2 = marker2.decode("utf-8")
    assert marker2 == uid, "grace marker must survive the losing consume"

    # Cleanup so the short-TTL key doesn't linger across the suite.
    await redis.delete(f"{_REDIS_REFRESH_GRACE_PREFIX}{jti}")


@pytest.mark.asyncio
async def test_refresh_token_rotation_grace_absorbs_browser_race(client, db_session):
    """Regression for 2026-06-02 prod report "刷新页面就退出登录".

    Root cause: React StrictMode (dev) double-invokes effect setups, so
    ``AuthRoute.useEffect(initialize)`` fired ``POST /auth/refresh``
    twice in parallel with the same cookie. Server consumed the JTI on
    the first call, the second hit Round-49's reuse-detect path, flipped
    ``force_logout_at``, and every access token issued by the first call
    (iat < force_logout_at) was rejected — user was kicked out on every
    single page reload.

    Fix: a 15-second rotation-grace marker in Redis. The first consume
    rotates J1→J2 and writes a marker (user-id ONLY — never the signed
    tokens) keyed by J1. A concurrent re-present of J1 inside the window
    is recognised as a legitimate race and minted its OWN fresh pair
    instead of tripping reuse-detect. Real attacker reuse landing
    seconds-to-hours after still flips force_logout_at.

    2026-06-02 follow-up (review-sweep): the marker stores no tokens
    (credential-at-rest invariant) and the loser re-mints rather than
    replaying the winner's pair (so two tabs never share one JTI).
    """
    user, refresh_token = await _seed_user_with_refresh_token(
        db_session, "racegrace@company.com", "sf-race-grace"
    )

    client.cookies.set("refresh_token", refresh_token)
    try:
        first = await client.post("/api/v1/auth/refresh")
        # The first response sets a NEW refresh_token cookie; restore the
        # original so the second call simulates a parallel browser tab
        # that hasn't seen the rotation yet.
        client.cookies.set("refresh_token", refresh_token)
        second = await client.post("/api/v1/auth/refresh")
    finally:
        client.cookies.delete("refresh_token")

    assert first.status_code == 200, f"first refresh failed: {first.status_code} {first.text}"
    assert second.status_code == 200, (
        f"second refresh of the same JTI returned {second.status_code} — "
        "rotation-grace should absorb the race, not trip Round-49's "
        "force_logout_at flip."
    )
    # Both calls yield usable access tokens (the loser re-mints its own).
    assert first.json()["data"]["access_token"].startswith("eyJ")
    assert second.json()["data"]["access_token"].startswith("eyJ")

    # Credential-at-rest guard (review-sweep #6): the grace marker must
    # NEVER contain a signed JWT — only the opaque user id.
    from app.auth.service import _REDIS_REFRESH_GRACE_PREFIX
    from app.core.redis_client import get_redis
    from app.core.security import decode_token

    j1 = decode_token(refresh_token)["jti"]
    raw = await get_redis().get(f"{_REDIS_REFRESH_GRACE_PREFIX}{j1}")
    if raw is not None:
        raw_s = raw.decode("utf-8") if isinstance(raw, bytes) else raw
        assert "eyJ" not in raw_s, (
            "refresh-grace marker leaked a signed JWT into Redis — "
            "credential-at-rest invariant broken."
        )

    await db_session.refresh(user, ["force_logout_at"])
    assert user.force_logout_at is None, (
        "force_logout_at was flipped on a legitimate browser race — the "
        "grace marker failed to suppress reuse-detect."
    )


@pytest.mark.asyncio
async def test_refresh_token_loser_remints_from_atomic_grace(db_session, monkeypatch):
    """Regression (2026-06-02 hardening): the consume now stamps the
    grace marker ATOMICALLY (``_CONSUME_AND_GRACE_SCRIPT``), so the
    instant a concurrent loser sees the JTI as already-consumed the
    marker is GUARANTEED present — no poll, no TOCTOU window that could
    otherwise let a benign race flip ``force_logout_at`` and re-create
    the "刷新就退出登录" kick-out.

    Simulate the loser deterministically: consume returns ``None`` (race
    lost) and a SINGLE grace read returns the owner. Assert refresh_tokens
    re-mints a fresh pair (its own session — not the winner's replayed
    tokens) on the first read and never flips force_logout_at.
    """
    from app.auth import service as auth_service
    from app.auth.service import AuthService
    from app.core.security import decode_token

    user, refresh_token = await _seed_user_with_refresh_token(
        db_session, "loserpoll@company.com", "sf-loser-poll"
    )
    payload = decode_token(refresh_token)

    async def _lost(_jti):
        # New contract: ``None`` == did not win the one-time consume.
        return None

    monkeypatch.setattr(auth_service, "_consume_refresh_jti", _lost)

    # The atomic consume guarantees the marker is present on the FIRST
    # read — assert exactly one fetch, no poll loop.
    calls = {"n": 0}

    async def _grace(_jti):
        calls["n"] += 1
        return payload["sub"]

    monkeypatch.setattr(auth_service, "_fetch_refresh_grace", _grace)

    u, access, refresh = await AuthService(db_session).refresh_tokens(dict(payload))
    assert access.startswith("eyJ") and refresh.startswith("eyJ")
    assert calls["n"] == 1, (
        "loser should read the atomically-stamped grace marker exactly "
        "once — the poll loop is gone."
    )

    await db_session.rollback()
    await db_session.refresh(user, ["force_logout_at"])
    assert user.force_logout_at is None, (
        "force_logout_at flipped on a legitimate concurrent refresh — the "
        "atomic grace stamp failed to suppress reuse-detect."
    )


@pytest.mark.asyncio
async def test_refresh_token_reuse_after_grace_still_force_logs_out(client, db_session, monkeypatch):
    """Companion to the grace-race test: real token theft (reuse outside
    the grace window) MUST still trip Round-49's defense. The grace is a
    15-second concession to browser race patterns, not a hole.
    """
    from app.auth import service as auth_service

    # Simulate the grace marker having EXPIRED (the reuse landed past the
    # 15s TTL) without sleeping: force the grace read to miss. The atomic
    # consume still stamps the marker, but a real replay arriving after
    # the TTL would find it gone — exactly what this no-op reproduces.
    async def _expired_grace(*_a, **_k):
        return None

    monkeypatch.setattr(auth_service, "_fetch_refresh_grace", _expired_grace)

    user, refresh_token = await _seed_user_with_refresh_token(
        db_session, "postgrace@company.com", "sf-post-grace"
    )

    client.cookies.set("refresh_token", refresh_token)
    try:
        first = await client.post("/api/v1/auth/refresh")
        client.cookies.set("refresh_token", refresh_token)
        second = await client.post("/api/v1/auth/refresh")
    finally:
        client.cookies.delete("refresh_token")

    assert first.status_code == 200
    # With grace disabled, the second consume of the SAME JTI is
    # indistinguishable from real theft → 401 + force_logout_at flip.
    assert second.status_code == 401, (
        f"post-grace reuse returned {second.status_code} — Round-49 "
        "reuse-detect must still fire when the grace window has elapsed."
    )

    # The force_logout_at flip commits in a separate session
    # (independent of the caller's rolled-back context, by design).
    # MySQL REPEATABLE READ would hide that commit from the test
    # session's open transaction, so end the current transaction
    # before refreshing — otherwise we'd assert against a stale
    # snapshot and pass against the bug.
    await db_session.rollback()
    await db_session.refresh(user, ["force_logout_at"])
    assert user.force_logout_at is not None, (
        "force_logout_at must be flipped on a real reuse-detect — the "
        "RFC 6819 / OAuth 2.1 defense is not a debug-only feature."
    )


@pytest.mark.asyncio
async def test_refresh_reuse_replay_of_same_jti_does_not_relock_after_login(
    client, db_session, monkeypatch
):
    """Regression for the 2026-08-04 account-lockout DoS.

    A ``python-requests`` script was replaying ONE dead refresh token for
    a real user every ~12s. The reuse-detect branch flipped
    ``force_logout_at = now()`` on EVERY hit. Because a successful
    interactive login clears the flag, the victim could re-login — but the
    next replay a few seconds later re-bumped the cutoff past their fresh
    token's ``iat`` and kicked them out again, forever. One leaked refresh
    token = perpetual lockout.

    Fix: the force-logout flip is claimed once per distinct reused JTI
    (Redis SET NX). The FIRST reuse still flips (RFC 6819 intact); a
    REPLAY of the same already-flagged JTI is a no-op, so a re-login is no
    longer un-done by the attacker's loop.
    """
    from app.auth import service as auth_service

    # Past-grace: every present of the dead JTI is treated as reuse.
    async def _expired_grace(*_a, **_k):
        return None

    monkeypatch.setattr(auth_service, "_fetch_refresh_grace", _expired_grace)

    user, refresh_token = await _seed_user_with_refresh_token(
        db_session, "relock@company.com", "sf-relock"
    )

    client.cookies.set("refresh_token", refresh_token)
    try:
        # Prime: legitimate consume rotates the JTI away (200).
        first = await client.post("/api/v1/auth/refresh")
        assert first.status_code == 200

        # Reuse #1 of the now-dead JTI → 401 + force_logout flipped once.
        client.cookies.set("refresh_token", refresh_token)
        second = await client.post("/api/v1/auth/refresh")
        assert second.status_code == 401
    finally:
        client.cookies.delete("refresh_token")

    await db_session.rollback()
    await db_session.refresh(user, ["force_logout_at"])
    assert user.force_logout_at is not None, (
        "first reuse-detect must still flip force_logout_at (RFC 6819)."
    )

    # Simulate the victim re-logging-in: a successful interactive login
    # clears the sticky flag (_clear_stale_force_logout).
    user.force_logout_at = None
    await db_session.commit()

    # Reuse #2 of the SAME dead JTI (the script's next loop iteration).
    # This MUST NOT re-flip — otherwise the victim is locked out again.
    client.cookies.set("refresh_token", refresh_token)
    try:
        third = await client.post("/api/v1/auth/refresh")
        assert third.status_code == 401, "replay is still rejected"
    finally:
        client.cookies.delete("refresh_token")

    await db_session.rollback()
    await db_session.refresh(user, ["force_logout_at"])
    assert user.force_logout_at is None, (
        "replaying an already-flagged dead JTI re-locked the account after "
        "re-login — the force_logout amplification DoS has regressed."
    )


async def _seed_password_user(
    db_session, email: str, hris_id: str, password: str,
    *, employment_status: str = "active", force_logout_at=None,
    termination_date=None,
):
    """Helper: insert a person with a bcrypt password (for login_via_password)
    and an employee role, optionally pre-bricked with ``force_logout_at``.
    Returns the IdentityPerson.
    """
    from sqlalchemy import text

    from app.auth.models import AuthUserRole
    from app.core.security import hash_password

    user_id = uuid.uuid4()
    user = IdentityPerson(
        id=user_id,
        work_email=email,
        display_name=hris_id,
        employment_status=employment_status,
        source_system="successfactors",
        hris_person_id=hris_id,
        hashed_password=hash_password(password),
        force_logout_at=force_logout_at,
        termination_date=termination_date,
    )
    db_session.add(user)
    await db_session.flush()
    role_id = (
        await db_session.execute(text("SELECT id FROM auth_roles WHERE name = 'employee'"))
    ).scalar_one()
    if isinstance(role_id, str):
        role_id = uuid.UUID(role_id)
    db_session.add(AuthUserRole(user_id=user_id, role_id=role_id))
    await db_session.flush()
    await db_session.commit()
    return user


@pytest.mark.asyncio
async def test_successful_login_clears_stale_force_logout(db_session):
    """A successful interactive login must release a sticky
    ``force_logout_at`` left by a reuse-detect false positive.

    Repro of the recurring "刷新页面就退出登录" report: a Redis JTI
    loss (restart-without-persistence / FLUSHDB / eviction / local
    ``docker compose down -v``) makes a legitimate ``/auth/refresh``
    indistinguishable from a replay → reuse-detect flips
    ``force_logout_at``. Before this fix NO login path cleared it (only
    offboarding-reopen), so the unconditional guard in ``refresh_tokens``
    kicked the user on every refresh forever — re-login could not save
    them. The clear-on-login escape hatch breaks that loop while leaving
    RFC 6819 reuse-detect intact (an attacker cannot interactively log
    in to undo it).
    """
    from datetime import UTC, datetime, timedelta

    from app.auth.service import AuthService

    stale = datetime.now(UTC) - timedelta(minutes=5)
    user = await _seed_password_user(
        db_session, "bricked@company.com", "sf-bricked", "pw-correct-1",
        force_logout_at=stale,
    )

    svc = AuthService(db_session)
    out_user, access, refresh = await svc.login_via_password(
        "bricked@company.com", "pw-correct-1"
    )
    await db_session.commit()

    assert access and refresh, "login must still issue a fresh token pair"
    await db_session.refresh(user, ["force_logout_at"])
    assert user.force_logout_at is None, (
        "a successful interactive login must clear the stale force_logout_at "
        "so the next /auth/refresh stops 401-ing the user into a re-login loop"
    )


@pytest.mark.asyncio
async def test_departed_user_cannot_self_clear_force_logout(db_session):
    """Offboarding safety: a departed user is rejected at the
    is_effectively_departed gate BEFORE the clear runs, so an
    offboarding-set force_logout_at can never be self-cleared by login.
    """
    from datetime import UTC, datetime

    from app.auth.service import AuthService

    stale = datetime.now(UTC).replace(microsecond=0)
    user = await _seed_password_user(
        db_session, "leaver@company.com", "sf-leaver", "pw-correct-2",
        employment_status="left", force_logout_at=stale,
    )

    svc = AuthService(db_session)
    with pytest.raises(ValueError):
        await svc.login_via_password("leaver@company.com", "pw-correct-2")
    await db_session.rollback()

    await db_session.refresh(user, ["force_logout_at"])
    assert user.force_logout_at is not None, (
        "a departed user must NOT be able to clear an offboarding shutdown "
        "by attempting to log in"
    )


@pytest.mark.asyncio
async def test_google_verify_subject_match_survives_email_rotation(
    client, db_session,
):
    """Existing Google ext_identity row (provider_subject = sub) → login
    succeeds even if the id_token's ``email`` no longer matches the
    Helix ``work_email`` (e.g. Workspace primary-alias rename).

    Mirror of the Feishu subject-first regression: ensures the same
    durable anchor exists on the Google path.
    """
    import uuid as _uuid

    from sqlalchemy import text

    from app.auth.models import AuthUserRole
    from app.identity.person_models import IdentityExternalIdentity, IdentityPerson

    person_id = _uuid.uuid4()
    db_session.add(
        IdentityPerson(
            id=person_id,
            work_email="gold@company.com",
            display_name="Subject Match User",
            employment_status="active",
            source_system="successfactors",
            hris_person_id="sf-subj-google-1",
        )
    )
    await db_session.flush()
    role_id = (
        await db_session.execute(
            text("SELECT id FROM auth_roles WHERE name = 'employee'")
        )
    ).scalar_one()
    if isinstance(role_id, str):
        role_id = _uuid.UUID(role_id)
    db_session.add(AuthUserRole(user_id=person_id, role_id=role_id))
    db_session.add(
        IdentityExternalIdentity(
            person_id=person_id,
            provider="google",
            provider_subject="google-sub-stable-7",
            provider_email="gold@company.com",
            is_login_identity=True,
        )
    )
    await db_session.commit()

    with respx.mock:
        respx.get("https://oauth2.googleapis.com/tokeninfo").mock(
            return_value=Response(
                200,
                json={
                    "sub": "google-sub-stable-7",
                    # Email rotated to an alias that has no Helix row.
                    "email": "renamed@company.com",
                    "email_verified": "true",
                    "name": "Subject Match User",
                    "picture": "https://example.com/p.jpg",
                    "aud": settings.GOOGLE_CLIENT_ID,
                    "iss": "accounts.google.com",
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/google/verify",
            json={"id_token": "fake-token"},
        )

    assert resp.status_code == 200, resp.text
    # The resolved user is the existing Helix row, NOT a stranger.
    assert resp.json()["data"]["user"]["email"] == "gold@company.com"

    # provider_email refreshed to the latest verified value for forensics.
    rows = (
        await db_session.execute(
            text(
                "SELECT provider_subject, provider_email "
                "FROM identity_external_identities "
                "WHERE person_id = :pid AND provider = 'google'"
            ),
            {"pid": str(person_id)},
        )
    ).all()
    assert len(rows) == 1
    assert rows[0].provider_subject == "google-sub-stable-7"
    assert rows[0].provider_email == "renamed@company.com"


@pytest.mark.asyncio
async def test_google_verify_subject_match_blocked_when_user_departed(
    client, db_session,
):
    """Departed user with prior Google binding → 401, NOT a bypass."""
    import uuid as _uuid

    from sqlalchemy import text

    from app.auth.models import AuthUserRole
    from app.identity.person_models import IdentityExternalIdentity, IdentityPerson

    person_id = _uuid.uuid4()
    db_session.add(
        IdentityPerson(
            id=person_id,
            work_email="left-google@company.com",
            display_name="Departed Google User",
            employment_status="left",
            source_system="successfactors",
            hris_person_id="sf-dep-google-1",
        )
    )
    await db_session.flush()
    role_id = (
        await db_session.execute(
            text("SELECT id FROM auth_roles WHERE name = 'employee'")
        )
    ).scalar_one()
    if isinstance(role_id, str):
        role_id = _uuid.UUID(role_id)
    db_session.add(AuthUserRole(user_id=person_id, role_id=role_id))
    db_session.add(
        IdentityExternalIdentity(
            person_id=person_id,
            provider="google",
            provider_subject="google-sub-departed-1",
            provider_email="left-google@company.com",
            is_login_identity=True,
        )
    )
    await db_session.commit()

    with respx.mock:
        respx.get("https://oauth2.googleapis.com/tokeninfo").mock(
            return_value=Response(
                200,
                json={
                    "sub": "google-sub-departed-1",
                    "email": "left-google@company.com",
                    "email_verified": "true",
                    "aud": settings.GOOGLE_CLIENT_ID,
                    "iss": "accounts.google.com",
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/google/verify",
            json={"id_token": "fake-token"},
        )
    assert resp.status_code == 401


@pytest.mark.asyncio
async def test_google_verify_revives_soft_deleted_ext_identity(client, db_session):
    """Soft-deleted google ext_identity must be revived in the upsert,
    NOT collided with via INSERT. Mirrors the Feishu regression for
    the same uq_identity_external_person_provider 1062 hazard.
    """
    import uuid as _uuid
    from datetime import UTC
    from datetime import datetime as _dt

    from sqlalchemy import text

    from app.auth.models import AuthUserRole
    from app.identity.person_models import IdentityExternalIdentity, IdentityPerson

    person_id = _uuid.uuid4()
    db_session.add(
        IdentityPerson(
            id=person_id,
            work_email="revive-google@company.com",
            display_name="Revive Google User",
            employment_status="active",
            source_system="successfactors",
            hris_person_id="sf-revive-google-1",
        )
    )
    await db_session.flush()
    role_id = (
        await db_session.execute(
            text("SELECT id FROM auth_roles WHERE name = 'employee'")
        )
    ).scalar_one()
    if isinstance(role_id, str):
        role_id = _uuid.UUID(role_id)
    db_session.add(AuthUserRole(user_id=person_id, role_id=role_id))
    db_session.add(
        IdentityExternalIdentity(
            person_id=person_id,
            provider="google",
            provider_subject="google-sub-tombstoned",
            provider_email="revive-google@company.com",
            is_login_identity=True,
            deleted_at=_dt.now(UTC),
        )
    )
    await db_session.commit()

    with respx.mock:
        respx.get("https://oauth2.googleapis.com/tokeninfo").mock(
            return_value=Response(
                200,
                json={
                    "sub": "google-sub-alive",
                    "email": "revive-google@company.com",
                    "email_verified": "true",
                    "aud": settings.GOOGLE_CLIENT_ID,
                    "iss": "accounts.google.com",
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/google/verify",
            json={"id_token": "fake-token"},
        )
    assert resp.status_code == 200, resp.text
    rows = (
        await db_session.execute(
            text(
                "SELECT provider_subject, deleted_at "
                "FROM identity_external_identities "
                "WHERE person_id = :pid AND provider = 'google'"
            ),
            {"pid": str(person_id)},
        )
    ).all()
    # Tombstone revived; no duplicate INSERT.
    assert len(rows) == 1
    assert rows[0].provider_subject == "google-sub-alive"
    assert rows[0].deleted_at is None


@pytest.mark.asyncio
async def test_me_accepts_uuid_subject_token(client, employee_user):
    """GET /auth/me accepts JWT sub as a stringified UUID under MySQL/TiDB."""
    _, token = employee_user

    resp = await client.get(
        "/api/v1/auth/me",
        headers={"Authorization": f"Bearer {token}"},
    )

    assert resp.status_code == 200
    data = resp.json()
    assert data["success"] is True
    assert data["data"]["email"] == "employee@test.com"


@pytest.mark.asyncio
async def test_org_tree_returns_nested_departments_with_people(client, db_session, admin_headers):
    from app.identity.person_models import IdentityOrgMembership, IdentityOrgNode, IdentityPerson

    root = IdentityOrgNode(
        source_system="successfactors",
        org_node_source_id="DP-ROOT",
        org_node_type="department",
        org_node_code="DP-ROOT",
        org_node_name="Engineering Group",
        org_level=1,
    )
    child = IdentityOrgNode(
        source_system="successfactors",
        org_node_source_id="DP-PLATFORM",
        org_node_type="department",
        org_node_code="DP-PLATFORM",
        org_node_name="Platform",
        parent_org_node_source_id="DP-ROOT",
        org_level=2,
    )
    grandchild = IdentityOrgNode(
        source_system="successfactors",
        org_node_source_id="DP-DB",
        org_node_type="department",
        org_node_code="DP-DB",
        org_node_name="Database Infra",
        parent_org_node_source_id="DP-PLATFORM",
        org_level=3,
    )
    db_session.add_all([root, child, grandchild])

    user = IdentityPerson(
        id=uuid.uuid4(),
        work_email="db@example.com",
        display_name="DB Owner",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="sf-db-owner",
    )
    db_session.add(user)
    await db_session.flush()

    db_session.add(
        IdentityOrgMembership(
            person_id=user.id,
            source_system="successfactors",
            org_node_source_id="DP-DB",
            org_node_type="department",
            org_node_code="DP-DB",
            org_node_name="Database Infra",
            parent_org_node_source_id="DP-PLATFORM",
            org_path="PingCAP / Engineering Group / Platform / Database Infra",
            is_primary=True,
        )
    )
    await db_session.commit()

    resp = await client.get("/api/v1/auth/org-tree", headers=admin_headers)

    assert resp.status_code == 200
    data = resp.json()["data"]
    assert data[0]["node_type"] == "department"
    assert data[0]["name"] == "Engineering Group"
    assert data[0]["member_count"] == 1
    assert data[0]["children"][0]["name"] == "Platform"
    assert data[0]["children"][0]["children"][0]["name"] == "Database Infra"
    assert data[0]["children"][0]["children"][0]["children"][0]["node_type"] == "person"
    assert data[0]["children"][0]["children"][0]["children"][0]["name"] == "DB Owner"


@pytest.mark.asyncio
async def test_org_tree_excludes_departed_employees(client, db_session, admin_headers):
    """Regression (2026-06-17): the tree view must not list people whose
    ``employment_status='left'`` — it's a live directory, not a forensic
    record. Mirrors ``get_position_org_chart``'s existing filter."""
    from app.identity.person_models import IdentityOrgMembership, IdentityOrgNode, IdentityPerson

    dept = IdentityOrgNode(
        source_system="successfactors",
        org_node_source_id="DP-SALES",
        org_node_type="department",
        org_node_code="DP-SALES",
        org_node_name="Sales",
        org_level=1,
    )
    db_session.add(dept)

    active = IdentityPerson(
        id=uuid.uuid4(), work_email="active.sales@example.com",
        display_name="Active Seller", employment_status="active",
        source_system="successfactors", hris_person_id="sf-active-sales",
    )
    departed = IdentityPerson(
        id=uuid.uuid4(), work_email="gone.sales@example.com",
        display_name="Departed Seller", employment_status="left",
        source_system="successfactors", hris_person_id="sf-gone-sales",
    )
    db_session.add_all([active, departed])
    await db_session.flush()
    for person in (active, departed):
        db_session.add(IdentityOrgMembership(
            person_id=person.id, source_system="successfactors",
            org_node_source_id="DP-SALES", org_node_type="department",
            org_node_code="DP-SALES", org_node_name="Sales",
            org_path="PingCAP / Sales", is_primary=True,
        ))
    await db_session.commit()

    resp = await client.get("/api/v1/auth/org-tree", headers=admin_headers)
    assert resp.status_code == 200

    names: list[str] = []

    def _collect(nodes):
        for n in nodes:
            if n["node_type"] == "person":
                names.append(n["name"])
            _collect(n.get("children", []))

    _collect(resp.json()["data"])
    assert "Active Seller" in names
    assert "Departed Seller" not in names


@pytest.mark.asyncio
async def test_position_org_chart_returns_nested_departments_with_incumbents(
    client,
    db_session,
    admin_headers,
):
    """Position org chart builds tree from local org_nodes + persons."""
    from app.identity.person_models import IdentityOrgMembership, IdentityOrgNode, IdentityPerson

    # Create department hierarchy: Company > Finance
    root_dept = IdentityOrgNode(
        source_system="successfactors",
        org_node_source_id="DP_ROOT",
        org_node_type="department",
        org_node_name="Company HQ",
        parent_org_node_source_id=None,
        node_status="active",
    )
    child_dept = IdentityOrgNode(
        source_system="successfactors",
        org_node_source_id="DP_FIN",
        org_node_type="department",
        org_node_name="Finance",
        parent_org_node_source_id="DP_ROOT",
        node_status="active",
    )
    db_session.add_all([root_dept, child_dept])
    await db_session.flush()

    # Create people
    ceo = IdentityPerson(
        id=uuid.uuid4(),
        work_email="ceo@example.com",
        display_name="CEO Person",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="U-CEO",
    )
    fin_lead = IdentityPerson(
        id=uuid.uuid4(),
        work_email="finance@example.com",
        display_name="Finance Lead",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="U-FIN",
    )
    db_session.add_all([ceo, fin_lead])
    await db_session.flush()

    # Assign to departments via org memberships
    db_session.add_all([
        IdentityOrgMembership(
            person_id=ceo.id,
            source_system="successfactors",
            org_node_source_id="DP_ROOT",
            org_node_type="department",
            org_node_name="Company HQ",
            is_primary=True,
        ),
        IdentityOrgMembership(
            person_id=fin_lead.id,
            source_system="successfactors",
            org_node_source_id="DP_FIN",
            org_node_type="department",
            org_node_name="Finance",
            is_primary=True,
        ),
    ])
    await db_session.commit()

    resp = await client.get("/api/v1/auth/position-org-chart", headers=admin_headers)
    assert resp.status_code == 200
    data = resp.json()["data"]

    # Find Company HQ root (there may also be an "unassigned" bucket for fixture users)
    named_roots = [n for n in data if n["code"] == "DP_ROOT"]
    assert len(named_roots) == 1
    root = named_roots[0]
    assert root["code"] == "DP_ROOT"
    assert root["name"] == "Company HQ"
    assert root["current_fte"] == 1
    assert root["direct_reports_count"] == 1
    assert root["incumbents"][0]["display_name"] == "CEO Person"
    assert root["incumbents"][0]["helix_person_id"] == str(ceo.id)

    # Child should be Finance
    assert root["children"][0]["code"] == "DP_FIN"
    assert root["children"][0]["name"] == "Finance"
    assert root["children"][0]["incumbents"][0]["display_name"] == "Finance Lead"


@pytest.mark.asyncio
async def test_position_org_chart_empty_when_no_departments(
    client,
    db_session,
    admin_headers,
):
    """Returns empty list when no org nodes exist."""
    resp = await client.get("/api/v1/auth/position-org-chart", headers=admin_headers)
    assert resp.status_code == 200
    assert resp.json()["data"] == []