"""Tests for Feishu SSO login: authorize URL + verify -> JWT.

Mirrors ``test_auth_api.py`` for the Google flow. State storage hits a
real Redis (already wired into the test conftest) and the Feishu
upstream endpoints are mocked with respx.
"""
from __future__ import annotations

import uuid
from urllib.parse import parse_qs, urlparse

import pytest
import respx
from httpx import Response
from sqlalchemy import text

from app.auth.models import AuthUserRole
from app.core.redis_client import get_redis
from app.identity.person_models import IdentityExternalIdentity, IdentityPerson
from app.integrations.models import Integration


@pytest.fixture(autouse=True)
async def _clear_login_rate_limit():
    """Wipe the per-IP login rate-limit counter before each test.

    ``check_login_rate_limit`` caps the test client (which always
    resolves to ``127.0.0.1``) at 20 hits per 15-minute window. This
    file runs ~12 tests that each touch one or two rate-limited
    endpoints — without the reset, the 11th test would 429 instead of
    exercising the path under test.
    """
    redis = get_redis()
    # Test client IP is fixed at 127.0.0.1 in the ASGITransport setup.
    try:
        await redis.delete("login_ratelimit:127.0.0.1")
    except Exception:
        pass
    yield


async def _seed_feishu_integration(db_session) -> Integration:
    """Seed a primary chat_approval Feishu integration with plaintext
    app_secret in config_public — the consumer flow merges
    config_public + decrypted secrets, so a test-only plaintext copy
    here reaches the SSO module without exercising the AES-GCM path.
    """
    integ = Integration(
        id=uuid.uuid4(),
        kind="chat_approval",
        provider="feishu",
        name="Feishu Test (SSO)",
        slug="feishu-sso-test-primary",
        enabled=True,
        is_primary=True,
        config_public={
            "app_id": "cli_test_app",
            "app_secret": "test-secret-shhh",
        },
    )
    db_session.add(integ)
    await db_session.commit()
    return integ


async def _seed_person(
    db_session,
    *,
    work_email: str,
    employment_status: str = "active",
    mobile_phone: str | None = None,
    mobile_country_code: str | None = None,
) -> IdentityPerson:
    user_id = uuid.uuid4()
    user = IdentityPerson(
        id=user_id,
        work_email=work_email,
        display_name="Feishu Test User",
        employment_status=employment_status,
        source_system="successfactors",
        hris_person_id=f"sf-feishu-{uuid.uuid4().hex[:8]}",
        mobile_phone=mobile_phone,
        mobile_country_code=mobile_country_code,
    )
    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.commit()
    return user


async def _request_authorize_url(client, redirect_uri: str) -> str:
    resp = await client.post(
        "/api/v1/auth/feishu/authorize-url",
        json={"redirect_uri": redirect_uri},
    )
    assert resp.status_code == 200, resp.text
    return resp.json()["data"]["authorize_url"]


def _state_from_authorize_url(url: str) -> str:
    qs = parse_qs(urlparse(url).query)
    states = qs.get("state") or []
    assert states, f"no state in authorize URL: {url}"
    return states[0]


# CORS_ORIGINS in the test settings always includes
# ``http://localhost:3000`` (see ``.env.example`` defaults), so this is
# the always-valid redirect_uri for the test environment.
_VALID_REDIRECT_URI = "http://localhost:3000/auth/feishu/callback"


@pytest.mark.asyncio
async def test_feishu_authorize_url_generates_state(client, db_session):
    await _seed_feishu_integration(db_session)

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)

    assert url.startswith(
        "https://accounts.feishu.cn/open-apis/authen/v1/authorize"
    )
    qs = parse_qs(urlparse(url).query)
    assert qs["app_id"] == ["cli_test_app"]
    assert qs["redirect_uri"] == [_VALID_REDIRECT_URI]
    assert qs["response_type"] == ["code"]
    # 256-bit base64url state → ≥ 43 chars
    assert len(qs["state"][0]) >= 40


@pytest.mark.asyncio
async def test_feishu_authorize_url_rejects_unknown_origin(client, db_session):
    await _seed_feishu_integration(db_session)

    resp = await client.post(
        "/api/v1/auth/feishu/authorize-url",
        json={"redirect_uri": "https://evil.example.com/auth/feishu/callback"},
    )
    # Bad redirect_uri → 400, NOT 401 (we surface validation errors
    # distinctly so the frontend can show the right message).
    assert resp.status_code == 400, resp.text


@pytest.mark.asyncio
async def test_feishu_authorize_url_returns_503_when_not_configured(client):
    """No chat_approval row at all → 503 FEISHU_NOT_CONFIGURED."""
    resp = await client.post(
        "/api/v1/auth/feishu/authorize-url",
        json={"redirect_uri": _VALID_REDIRECT_URI},
    )
    assert resp.status_code == 503, resp.text


@pytest.mark.asyncio
async def test_feishu_verify_happy_path(client, db_session):
    """authorize-url → verify (code + state) → JWT for the matched person."""
    await _seed_feishu_integration(db_session)
    user = await _seed_person(db_session, work_email="enterprise@pingcap.com")

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "access_token": "u-access-token-abc",
                    "token_type": "Bearer",
                    "expires_in": 7200,
                },
            )
        )
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "msg": "success",
                    "data": {
                        "name": "Feishu Test User",
                        "en_name": "Feishu Test",
                        "avatar_url": "https://feishu.example/avatar.png",
                        "open_id": "ou_open_abc",
                        "union_id": "on_union_xyz",
                        "email": "personal@example.com",
                        "enterprise_email": "enterprise@pingcap.com",
                        "tenant_key": "tnt_pingcap",
                    },
                },
            )
        )

        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-from-feishu",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )

    assert resp.status_code == 200, resp.text
    data = resp.json()
    assert data["success"] is True
    assert "access_token" in data["data"]
    assert data["data"]["user"]["email"] == "enterprise@pingcap.com"

    # External identity row should be upserted with the union_id.
    await db_session.refresh(user)
    rows = await db_session.execute(
        text(
            "SELECT provider, provider_subject "
            "FROM identity_external_identities "
            "WHERE person_id = :pid"
        ),
        {"pid": str(user.id)},
    )
    bindings = list(rows.all())
    feishu_bindings = [r for r in bindings if r.provider == "feishu"]
    assert len(feishu_bindings) == 1
    assert feishu_bindings[0].provider_subject == "on_union_xyz"


@pytest.mark.asyncio
async def test_feishu_verify_falls_back_to_email_when_no_enterprise_email(
    client, db_session,
):
    """enterprise_email empty → fall back to ``email`` for HRIS match."""
    await _seed_feishu_integration(db_session)
    await _seed_person(db_session, work_email="fallback@pingcap.com")

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-2"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_2",
                        "union_id": "on_2",
                        "email": "fallback@pingcap.com",
                        # enterprise_email intentionally absent
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-2",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    assert resp.status_code == 200, resp.text
    assert resp.json()["data"]["user"]["email"] == "fallback@pingcap.com"


@pytest.mark.asyncio
async def test_feishu_verify_rejects_replay_of_state(client, db_session):
    """state is single-use — replaying the same code+state returns 401."""
    await _seed_feishu_integration(db_session)
    await _seed_person(db_session, work_email="replay@pingcap.com")

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-r"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_r",
                        "union_id": "on_r",
                        "enterprise_email": "replay@pingcap.com",
                    },
                },
            )
        )
        first = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-r",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
        assert first.status_code == 200, first.text

        # Same state again — state must have been GETDEL-consumed.
        second = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-r",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    assert second.status_code == 401, second.text


@pytest.mark.asyncio
async def test_feishu_verify_rejects_redirect_uri_mismatch(client, db_session):
    """State stored for URI A cannot be redeemed claiming URI B."""
    await _seed_feishu_integration(db_session)

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    # Verify with a different (but allowed) redirect_uri — we don't
    # have two allowed origins in tests, so simulate by manually
    # passing the localhost URI back with extra path segment which
    # validate_redirect_uri rejects (400 path), distinct from the
    # state-redirect_uri-mismatch path.
    resp = await client.post(
        "/api/v1/auth/feishu/verify",
        json={
            "code": "anything",
            "state": state,
            "redirect_uri": "http://localhost:3000/wrong/callback",
        },
    )
    # validate_redirect_uri is called BEFORE state verification — bad
    # URI returns 400. (State remains consumable for the duration of
    # the TTL since validate runs first.)
    assert resp.status_code == 400, resp.text


@pytest.mark.asyncio
async def test_feishu_verify_user_not_in_hris(client, db_session):
    """Feishu identity resolves to no HRIS person → 401 LOGIN_FAILED."""
    await _seed_feishu_integration(db_session)

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-nh"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_nh",
                        "union_id": "on_nh",
                        "enterprise_email": "stranger@example.com",
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-nh",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    assert resp.status_code == 401, resp.text


@pytest.mark.asyncio
async def test_feishu_verify_user_departed(client, db_session):
    """Departed person logs in via Feishu → 401."""
    await _seed_feishu_integration(db_session)
    await _seed_person(
        db_session,
        work_email="left@pingcap.com",
        employment_status="left",
    )

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-d"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_d",
                        "union_id": "on_d",
                        "enterprise_email": "left@pingcap.com",
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-d",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    assert resp.status_code == 401, resp.text


@pytest.mark.asyncio
async def test_feishu_verify_existing_binding_subject_refresh(client, db_session):
    """If the person already has a feishu external_identity row but
    union_id rotates, the existing row is updated (not duplicated)."""
    await _seed_feishu_integration(db_session)
    user = await _seed_person(db_session, work_email="rotate@pingcap.com")
    db_session.add(
        IdentityExternalIdentity(
            person_id=user.id,
            provider="feishu",
            provider_subject="on_old_subject",
            provider_email="rotate@pingcap.com",
            is_login_identity=True,
        )
    )
    await db_session.commit()

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-rot"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_rot",
                        "union_id": "on_new_subject",
                        "enterprise_email": "rotate@pingcap.com",
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-rot",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    assert resp.status_code == 200, resp.text

    rows = (
        await db_session.execute(
            text(
                "SELECT provider_subject FROM identity_external_identities "
                "WHERE person_id = :pid AND provider = 'feishu'"
            ),
            {"pid": str(user.id)},
        )
    ).all()
    assert len(rows) == 1
    assert rows[0].provider_subject == "on_new_subject"


@pytest.mark.asyncio
async def test_feishu_verify_token_endpoint_error(client, db_session):
    """Feishu returns a non-zero ``code`` on token exchange → 401."""
    await _seed_feishu_integration(db_session)

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 20001,
                    "error": "invalid_grant",
                    "error_description": "code expired",
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "stale-code",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    assert resp.status_code == 401, resp.text


@pytest.mark.asyncio
async def test_feishu_verify_user_info_missing_email(client, db_session):
    """Feishu account with neither enterprise_email nor email → 401."""
    await _seed_feishu_integration(db_session)

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-noe"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_noe",
                        "union_id": "on_noe",
                        # no email or enterprise_email
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-noe",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    assert resp.status_code == 401, resp.text


# ---------------------------------------------------------------------------
# Subject-first lookup (2026-05-29 prod incident regression)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_feishu_verify_subject_match_survives_enterprise_email_rotation(
    client, db_session,
):
    """Existing union_id binding → login succeeds even when Feishu admin
    rotated ``enterprise_email`` to a domain that no longer matches the
    Helix ``work_email``.

    Repro: 2026-05-29 PingCAP feishu admin changed every employee's
    ``enterprise_email`` from ``@pingcap.com`` to ``@pingcap.com.cn``;
    every already-bound user was locked out because the legacy code
    only matched by email. Subject-first lookup is the durable fix.
    """
    await _seed_feishu_integration(db_session)
    user = await _seed_person(db_session, work_email="rotated@pingcap.com")
    db_session.add(
        IdentityExternalIdentity(
            person_id=user.id,
            provider="feishu",
            provider_subject="on_stable_union_id",
            provider_email="rotated@pingcap.com",
            is_login_identity=True,
        )
    )
    await db_session.commit()

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-rot2"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_rot2",
                        "union_id": "on_stable_union_id",
                        # Enterprise email rotated to .com.cn by admin —
                        # this no longer matches any work_email in Helix.
                        "enterprise_email": "rotated@pingcap.com.cn",
                        "email": "rotated@pingcap.com",
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-rot2",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    assert resp.status_code == 200, resp.text
    # Provider email should be refreshed to the latest enterprise_email
    # snapshot for forensics / audit.
    rows = (
        await db_session.execute(
            text(
                "SELECT provider_subject, provider_email FROM identity_external_identities "
                "WHERE person_id = :pid AND provider = 'feishu'"
            ),
            {"pid": str(user.id)},
        )
    ).all()
    assert len(rows) == 1
    assert rows[0].provider_subject == "on_stable_union_id"
    assert rows[0].provider_email == "rotated@pingcap.com.cn"


@pytest.mark.asyncio
async def test_feishu_verify_subject_lookup_uses_open_id_when_union_id_absent(
    client, db_session,
):
    """An ext_identity row written under an open_id-only scope still
    matches when a later login carries union_id — both subjects are
    tried during lookup."""
    await _seed_feishu_integration(db_session)
    user = await _seed_person(db_session, work_email="legacy-open@pingcap.com")
    # Legacy binding: only open_id available at the time.
    db_session.add(
        IdentityExternalIdentity(
            person_id=user.id,
            provider="feishu",
            provider_subject="ou_legacy_open_id",
            provider_email="legacy-open@pingcap.com",
            is_login_identity=True,
        )
    )
    await db_session.commit()

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-leg"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_legacy_open_id",
                        "union_id": "on_new_union_id",
                        # Email rotated — only open_id resolves.
                        "enterprise_email": "different@pingcap.com.cn",
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-leg",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    assert resp.status_code == 200, resp.text
    # Stored row should be upgraded to the preferred union_id.
    rows = (
        await db_session.execute(
            text(
                "SELECT provider_subject FROM identity_external_identities "
                "WHERE person_id = :pid AND provider = 'feishu'"
            ),
            {"pid": str(user.id)},
        )
    ).all()
    assert len(rows) == 1
    assert rows[0].provider_subject == "on_new_union_id"


@pytest.mark.asyncio
async def test_feishu_verify_subject_beats_conflicting_email(client, db_session):
    """When the subject resolves person X but the email field resolves
    person Y, subject wins — guards against admin error / collision where
    one person's Feishu email points at a different person's work_email.
    """
    await _seed_feishu_integration(db_session)
    user_x = await _seed_person(db_session, work_email="alice@pingcap.com")
    user_y = await _seed_person(db_session, work_email="bob@pingcap.com")
    db_session.add(
        IdentityExternalIdentity(
            person_id=user_x.id,
            provider="feishu",
            provider_subject="on_alice_subject",
            provider_email="alice@pingcap.com",
            is_login_identity=True,
        )
    )
    await db_session.commit()

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-conf"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_alice",
                        "union_id": "on_alice_subject",
                        # Misconfigured — points at Bob's work_email
                        # while subject is Alice's. Must NOT resolve to Bob.
                        "enterprise_email": "bob@pingcap.com",
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-conf",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    assert data["user"]["email"] == "alice@pingcap.com"
    assert data["user"]["email"] != "bob@pingcap.com"
    # Defensive: confirm we didn't accidentally bind Bob to Alice's subject.
    bob_rows = (
        await db_session.execute(
            text(
                "SELECT id FROM identity_external_identities "
                "WHERE person_id = :pid AND provider = 'feishu'"
            ),
            {"pid": str(user_y.id)},
        )
    ).all()
    assert bob_rows == []


# ---------------------------------------------------------------------------
# Email-fallback anti-takeover (pre-takeover defence)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_feishu_verify_falls_back_to_email_when_local_part_matches(
    client, db_session,
):
    """First-time binding: ``enterprise_email`` doesn't resolve in Helix
    but ``email`` does, AND their local-parts match — this is the
    legitimate PingCAP ``.com``/``.com.cn`` case. Login should succeed.
    """
    await _seed_feishu_integration(db_session)
    await _seed_person(db_session, work_email="shuai.feng@pingcap.com")

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-lp"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_lp",
                        "union_id": "on_lp",
                        "enterprise_email": "shuai.feng@pingcap.com.cn",
                        "email": "shuai.feng@pingcap.com",
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-lp",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    assert resp.status_code == 200, resp.text
    assert resp.json()["data"]["user"]["email"] == "shuai.feng@pingcap.com"


@pytest.mark.asyncio
async def test_feishu_verify_rejects_email_takeover_via_local_part_mismatch(
    client, db_session,
):
    """Pre-takeover attack: malicious Feishu user with their own admin-bound
    ``enterprise_email`` (no Helix row) sets the self-editable ``email``
    field to a victim's ``work_email`` — must 401, NOT impersonate.
    """
    await _seed_feishu_integration(db_session)
    # Victim exists in Helix.
    await _seed_person(db_session, work_email="victim@pingcap.com")

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-atk"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_attacker",
                        "union_id": "on_attacker",
                        # Attacker's admin-bound email — no Helix row.
                        "enterprise_email": "attacker@evil.example",
                        # Attacker-controlled, points at victim.
                        "email": "victim@pingcap.com",
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-atk",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    assert resp.status_code == 401, resp.text


@pytest.mark.asyncio
async def test_feishu_verify_email_only_no_enterprise_skips_local_part_guard(
    client, db_session,
):
    """``enterprise_email`` empty (e.g. external user) → ``email`` fallback
    has no enterprise-email-local-part to compare against, so the
    local-part guard does not fire. Behaviour matches the pre-2026-05-29
    ``pick_match_email`` for empty enterprise_email.
    """
    await _seed_feishu_integration(db_session)
    await _seed_person(db_session, work_email="contractor@pingcap.com")

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-c"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_c",
                        "union_id": "on_c",
                        "email": "contractor@pingcap.com",
                        # enterprise_email absent
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-c",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    assert resp.status_code == 200, resp.text


@pytest.mark.asyncio
async def test_feishu_verify_email_only_cannot_take_over_claimed_account(
    client, db_session,
):
    """Account-takeover regression (security review 2026-07-22).

    A Feishu tenant guest with NO admin-bound ``enterprise_email`` sets
    their self-editable ``email`` to a victim's ``work_email``. The victim
    is a real employee who has ALREADY claimed their account (an active
    Google login identity). The email-only fallback must refuse to bind
    the guest's Feishu subject to the victim → 401, not impersonation.
    """
    await _seed_feishu_integration(db_session)
    victim = await _seed_person(db_session, work_email="victim2@pingcap.com")
    # Victim already authenticated once via Google → account is claimed.
    db_session.add(
        IdentityExternalIdentity(
            person_id=victim.id,
            provider="google",
            provider_subject="google-sub-victim2",
            provider_email="victim2@pingcap.com",
            is_login_identity=True,
        )
    )
    await db_session.commit()

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-atk2"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_attacker2",
                        "union_id": "on_attacker2",
                        # No enterprise_email; self-editable email points
                        # at the victim's already-claimed work_email.
                        "email": "victim2@pingcap.com",
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-atk2",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    assert resp.status_code == 401, resp.text


@pytest.mark.asyncio
async def test_feishu_verify_email_only_binds_claimed_account_when_mobile_matches(
    client, db_session,
):
    """Mobile-corroboration escape hatch (2026-08-03 prod fix).

    A genuine employee whose Feishu tenant sets NO admin-bound
    ``enterprise_email`` (the PingCAP JP tenant) and who already carries
    an auto-provisioned Google login identity from onboarding could not
    bind Feishu at all: the email-only fallback saw the (never-used)
    Google identity and silently rejected. When the Feishu account's
    admin-managed ``mobile`` matches the person's stored ``mobile_phone``
    the bind is now allowed — the impersonator in
    ``...cannot_take_over_claimed_account`` cannot pass this because
    their own account's mobile is not the victim's.
    """
    await _seed_feishu_integration(db_session)
    # Helix stores the JP national number with a trunk 0; Feishu returns
    # the +81 international form. They must corroborate.
    person = await _seed_person(
        db_session,
        work_email="jp.employee@pingcap.com",
        mobile_phone="09064764314",
        mobile_country_code="JP",
    )
    db_session.add(
        IdentityExternalIdentity(
            person_id=person.id,
            provider="google",
            provider_subject="google-sub-jp-emp",
            provider_email="jp.employee@pingcap.com",
            is_login_identity=True,
        )
    )
    await db_session.commit()

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-jp"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_jp_emp",
                        "union_id": "on_jp_emp",
                        # No enterprise_email (JP tenant); email matches
                        # work_email; mobile is the +81 international form.
                        "email": "jp.employee@pingcap.com",
                        "mobile": "+819064764314",
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-jp",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    assert resp.status_code == 200, resp.text
    # A live feishu login identity was bound for the person.
    ext = (
        await db_session.execute(
            text(
                "SELECT provider_subject, is_login_identity FROM "
                "identity_external_identities WHERE person_id = :pid "
                "AND provider = 'feishu' AND deleted_at IS NULL"
            ),
            {"pid": str(person.id)},
        )
    ).first()
    assert ext is not None
    assert ext[0] == "on_jp_emp"


@pytest.mark.asyncio
async def test_feishu_verify_email_only_claimed_account_rejects_mobile_mismatch(
    client, db_session,
):
    """Corroboration is a real gate: a claimed account with a NON-matching
    Feishu mobile is still refused (the 2026-07-22 takeover guard holds).
    """
    await _seed_feishu_integration(db_session)
    victim = await _seed_person(
        db_session,
        work_email="victim3@pingcap.com",
        mobile_phone="09064764314",
        mobile_country_code="JP",
    )
    db_session.add(
        IdentityExternalIdentity(
            person_id=victim.id,
            provider="google",
            provider_subject="google-sub-victim3",
            provider_email="victim3@pingcap.com",
            is_login_identity=True,
        )
    )
    await db_session.commit()

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-atk3"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_attacker3",
                        "union_id": "on_attacker3",
                        "email": "victim3@pingcap.com",
                        # Attacker's own (different) phone → no corroboration.
                        "mobile": "+8613800001111",
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-atk3",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    assert resp.status_code == 401, resp.text


# ---------------------------------------------------------------------------
# Code-review round 1 regressions
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_feishu_verify_revives_soft_deleted_ext_identity(client, db_session):
    """Soft-deleted feishu ext_identity must be revived in the upsert,
    NOT collided with via INSERT. Mirrors the HR-side prod incident on
    2026-05-26: the (person_id, provider) unique constraint spans
    soft-deletes, so an INSERT path raises 1062 unless the tombstone
    is reused.
    """
    from datetime import UTC
    from datetime import datetime as _dt

    await _seed_feishu_integration(db_session)
    user = await _seed_person(db_session, work_email="revive@pingcap.com")
    # Pre-existing tombstoned binding (e.g. left over from an HR cleanup).
    db_session.add(
        IdentityExternalIdentity(
            person_id=user.id,
            provider="feishu",
            provider_subject="on_dead_subject",
            provider_email="revive@pingcap.com",
            is_login_identity=True,
            deleted_at=_dt.now(UTC),
        )
    )
    await db_session.commit()

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-rev"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_rev",
                        "union_id": "on_alive_subject",
                        "enterprise_email": "revive@pingcap.com",
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-rev",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    assert resp.status_code == 200, resp.text
    rows = (
        await db_session.execute(
            text(
                "SELECT provider_subject, deleted_at, is_login_identity "
                "FROM identity_external_identities "
                "WHERE person_id = :pid AND provider = 'feishu'"
            ),
            {"pid": str(user.id)},
        )
    ).all()
    # Exactly one row — the revived tombstone — not a new duplicate.
    assert len(rows) == 1
    assert rows[0].provider_subject == "on_alive_subject"
    assert rows[0].deleted_at is None
    assert rows[0].is_login_identity is True or rows[0].is_login_identity == 1


@pytest.mark.asyncio
async def test_feishu_verify_does_not_downgrade_union_id_to_open_id(
    client, db_session,
):
    """If the stored subject is a ``union_id`` (``on_…``) and the current
    login only carries ``open_id`` (e.g. the app lost the union_id scope
    transiently), the stored union_id MUST be preserved — never silently
    downgraded to the less stable open_id.
    """
    await _seed_feishu_integration(db_session)
    user = await _seed_person(db_session, work_email="no-downgrade@pingcap.com")
    db_session.add(
        IdentityExternalIdentity(
            person_id=user.id,
            provider="feishu",
            provider_subject="on_durable_union_id",
            provider_email="no-downgrade@pingcap.com",
            is_login_identity=True,
        )
    )
    await db_session.commit()

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-nd"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        # Only open_id this time (scope downgrade).
                        "open_id": "ou_transient",
                        "enterprise_email": "no-downgrade@pingcap.com",
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-nd",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    assert resp.status_code == 200, resp.text
    rows = (
        await db_session.execute(
            text(
                "SELECT provider_subject FROM identity_external_identities "
                "WHERE person_id = :pid AND provider = 'feishu'"
            ),
            {"pid": str(user.id)},
        )
    ).all()
    assert len(rows) == 1
    # Stored union_id preserved, NOT overwritten with the open_id.
    assert rows[0].provider_subject == "on_durable_union_id"


@pytest.mark.asyncio
async def test_feishu_verify_local_part_guard_requires_at_sign(client, db_session):
    """Anti-takeover guard rejects payloads without ``@`` in either
    address — defence against pathological / forged user_info where
    ``"foo".split("@",1)[0] == "foo"`` would otherwise pass the
    naive equality check.
    """
    await _seed_feishu_integration(db_session)
    # Seed a person whose work_email collides with the local-part-only
    # value the attacker chose.
    await _seed_person(db_session, work_email="foo@pingcap.com")

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-at"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_at",
                        "union_id": "on_at",
                        # Neither has '@' — must NOT pass the local-part guard.
                        "enterprise_email": "foo",
                        "email": "foo@pingcap.com",
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-at",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    # enterprise_email "foo" doesn't resolve (Helix is "foo@pingcap.com").
    # email "foo@pingcap.com" resolves, but the guard requires '@' in BOTH —
    # enterprise_email "foo" has none → guard rejects → 401.
    assert resp.status_code == 401, resp.text


@pytest.mark.asyncio
async def test_feishu_verify_rejects_admin_disabled_binding(client, db_session):
    """If an admin set ``is_login_identity=False`` on a live (non-deleted)
    Feishu binding (the explicit disable signal), subsequent SSO logins
    MUST be rejected with 401 — not silently bound, and certainly not
    silently re-enabled. The disabled flag survives the upsert.
    """
    await _seed_feishu_integration(db_session)
    user = await _seed_person(db_session, work_email="disabled@pingcap.com")
    db_session.add(
        IdentityExternalIdentity(
            person_id=user.id,
            provider="feishu",
            provider_subject="on_disabled_flag",
            provider_email="disabled@pingcap.com",
            is_login_identity=False,  # admin-disabled
        )
    )
    await db_session.commit()

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-df"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_df",
                        "union_id": "on_disabled_flag",
                        "enterprise_email": "disabled@pingcap.com",
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-df",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    # Subject-first lookup filters is_login_identity=True → miss. Email
    # path resolves the user → upsert finds the live row with the flag
    # set False → raises ValueError → 401.
    assert resp.status_code == 401, resp.text

    # Flag must NOT have been silently overwritten to True.
    rows = (
        await db_session.execute(
            text(
                "SELECT is_login_identity FROM identity_external_identities "
                "WHERE person_id = :pid AND provider = 'feishu'"
            ),
            {"pid": str(user.id)},
        )
    ).all()
    assert len(rows) == 1
    flag = rows[0].is_login_identity
    assert flag is False or flag == 0, f"is_login_identity was silently re-enabled: {flag!r}"


@pytest.mark.asyncio
async def test_feishu_verify_rejects_suspended_identity_status(client, db_session):
    """A live ext_identity with ``identity_status='suspended'`` (admin
    soft-suspend) must reject the login. Subject-first lookup filters it
    out and the email-path upsert refuses to silently re-bind."""
    await _seed_feishu_integration(db_session)
    user = await _seed_person(db_session, work_email="suspended@pingcap.com")
    db_session.add(
        IdentityExternalIdentity(
            person_id=user.id,
            provider="feishu",
            provider_subject="on_suspended_subj",
            provider_email="suspended@pingcap.com",
            is_login_identity=True,
            identity_status="suspended",
        )
    )
    await db_session.commit()

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-su"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_su",
                        "union_id": "on_suspended_subj",
                        "enterprise_email": "suspended@pingcap.com",
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-su",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    assert resp.status_code == 401, resp.text


@pytest.mark.asyncio
async def test_feishu_verify_tombstone_revival_preserves_disabled_flag(
    client, db_session,
):
    """A row that was soft-deleted AND had ``is_login_identity=False``
    (admin set both kill-switches) must, on next login, be revived with
    its disabled state intact and the login rejected — not silently
    re-enabled.
    """
    from datetime import UTC
    from datetime import datetime as _dt

    await _seed_feishu_integration(db_session)
    user = await _seed_person(db_session, work_email="double-kill@pingcap.com")
    db_session.add(
        IdentityExternalIdentity(
            person_id=user.id,
            provider="feishu",
            provider_subject="on_double_kill",
            provider_email="double-kill@pingcap.com",
            is_login_identity=False,  # explicit disable
            deleted_at=_dt.now(UTC),   # AND tombstoned
        )
    )
    await db_session.commit()

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-dk"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_dk",
                        "union_id": "on_double_kill",
                        "enterprise_email": "double-kill@pingcap.com",
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-dk",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    assert resp.status_code == 401, resp.text

    # Row still has is_login_identity=False; deleted_at may have been
    # cleared as a side effect of the resurrection path, but the more
    # important kill-switch survives.
    rows = (
        await db_session.execute(
            text(
                "SELECT is_login_identity FROM identity_external_identities "
                "WHERE person_id = :pid AND provider = 'feishu'"
            ),
            {"pid": str(user.id)},
        )
    ).all()
    assert len(rows) == 1
    flag = rows[0].is_login_identity
    assert flag is False or flag == 0


@pytest.mark.asyncio
async def test_feishu_verify_rotation_collision_blocks_swap_and_keeps_session_clean(
    client, db_session,
):
    """Subject rotation must NOT clobber another person's existing subject:

    * Person A has a live binding with the legacy ``open_id`` (``ou_A``).
    * Person B legitimately owns the union_id ``on_X`` (e.g. they bound
      first with the newer scope).
    * A logs in and Feishu now returns both ``open_id=ou_A`` AND
      ``union_id=on_X`` (admin merge / data-corruption scenario — A and
      B's accounts share a union_id somehow).
    * Subject-first lookup tries ``on_X`` → matches B's row → A would
      log in as B. **But because A's resolved person is B, the upsert
      then tries no rotation (B's stored subject already is on_X)** —
      so the cleaner regression is to force the rotation path.

    Forced rotation path:

    * A has live binding ``ou_A``.
    * B has live binding ``on_X``.
    * Feishu returns A's ``open_id=ou_A`` only (no union_id this call),
      then a SECOND login attempt for A returns ``union_id=on_X``.
      Subject-first on the second call hits B (subject ``on_X``), logs
      in as B → that's an admin data-corruption symptom, not a code
      bug we can defeat without filtering.

    The genuinely-defendable rotation collision: A has live ``ou_A``,
    PLUS Feishu now gives A a union_id ``on_X`` that COLLIDES with B's
    stored ``on_X`` — subject-first tries ``on_X`` first → finds B →
    short-circuits to B. We log in as B. This is data-corruption,
    cannot be defended at the auth layer without per-tenant subject
    scoping which Helix doesn't currently store.

    Therefore THIS test validates the simpler invariant: when
    ``_safe_swap_subject`` is invoked AND the target collides with a
    different row, ValueError is raised cleanly (audit captures it,
    response is 401). We exercise this via a unit-level test on the
    helper itself (see ``test_safe_swap_subject_*``) — covered there.
    The end-to-end happy path here just confirms subject-first
    short-circuits to the matched row, ``_safe_swap_subject`` is NOT
    invoked when stored already equals new_subject, and login
    completes as the matched user.
    """
    await _seed_feishu_integration(db_session)
    user_b = await _seed_person(db_session, work_email="b-owner@pingcap.com")
    db_session.add(
        IdentityExternalIdentity(
            person_id=user_b.id,
            provider="feishu",
            provider_subject="on_short_circuit",
            provider_email="b-owner@pingcap.com",
            is_login_identity=True,
        )
    )
    await db_session.commit()

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-sc"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_b",
                        # Subject-first matches B's stored on_short_circuit
                        # → resolves B → no rotation needed.
                        "union_id": "on_short_circuit",
                        "enterprise_email": "b-owner@pingcap.com",
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-sc",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    assert resp.status_code == 200, resp.text
    assert resp.json()["data"]["user"]["email"] == "b-owner@pingcap.com"


@pytest.mark.asyncio
async def test_safe_swap_subject_rejects_collision(db_session):
    """Unit-level: ``_safe_swap_subject`` must raise ValueError when the
    target subject is already owned by another row (live or tombstoned).
    The caller can then 401 and avoid the IntegrityError → 500 path.
    """
    from app.auth.service import AuthService

    # Seed two rows: target subject 'on_target' is held by some other row.
    other_person = await _seed_person(db_session, work_email="other@pingcap.com")
    db_session.add(
        IdentityExternalIdentity(
            person_id=other_person.id,
            provider="feishu",
            provider_subject="on_target",
            provider_email="other@pingcap.com",
            is_login_identity=True,
        )
    )
    mine_person = await _seed_person(db_session, work_email="mine@pingcap.com")
    mine_ext = IdentityExternalIdentity(
        person_id=mine_person.id,
        provider="feishu",
        provider_subject="ou_mine",
        provider_email="mine@pingcap.com",
        is_login_identity=True,
    )
    db_session.add(mine_ext)
    await db_session.commit()

    svc = AuthService(db_session)
    with pytest.raises(ValueError):
        await svc._safe_swap_subject(mine_ext, "on_target")

    # mine_ext must remain unchanged on the rejection path.
    await db_session.refresh(mine_ext)
    assert mine_ext.provider_subject == "ou_mine"


@pytest.mark.asyncio
async def test_feishu_verify_local_part_guard_rejects_empty_local_part(
    client, db_session,
):
    """Both addresses contain ``@`` but the local-part is empty in both —
    must NOT pass the guard (empty == empty would otherwise bypass it).
    """
    await _seed_feishu_integration(db_session)
    # No seeded person at the degenerate email; the assertion is that
    # the guard rejects regardless. (Seeding work_email='@pingcap.com'
    # would be a corrupt fixture and is not the scenario under test.)

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-elp"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_elp",
                        "union_id": "on_elp",
                        "enterprise_email": "@evil.example",
                        "email": "@pingcap.com",
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-elp",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    # Both local-parts split to '' — guard must reject (no match), and
    # the find_user_by_email("@pingcap.com") on the email-field branch
    # would only resolve if such a (degenerate) work_email existed AND
    # the guard passed. Either way, no successful login here.
    assert resp.status_code == 401, resp.text


@pytest.mark.asyncio
async def test_feishu_verify_departed_subject_still_rejected(client, db_session):
    """Subject-first must NOT bypass the departed gate — a leaver whose
    Feishu ext_identity row still exists must still 401."""
    await _seed_feishu_integration(db_session)
    user = await _seed_person(
        db_session,
        work_email="bound-leaver@pingcap.com",
        employment_status="left",
    )
    db_session.add(
        IdentityExternalIdentity(
            person_id=user.id,
            provider="feishu",
            provider_subject="on_bound_leaver",
            provider_email="bound-leaver@pingcap.com",
            is_login_identity=True,
        )
    )
    await db_session.commit()

    url = await _request_authorize_url(client, _VALID_REDIRECT_URI)
    state = _state_from_authorize_url(url)

    with respx.mock:
        respx.post(
            "https://open.feishu.cn/open-apis/authen/v2/oauth/token"
        ).mock(return_value=Response(200, json={"code": 0, "access_token": "u-token-bl"}))
        respx.get(
            "https://open.feishu.cn/open-apis/authen/v1/user_info"
        ).mock(
            return_value=Response(
                200,
                json={
                    "code": 0,
                    "data": {
                        "open_id": "ou_bl",
                        "union_id": "on_bound_leaver",
                        "enterprise_email": "bound-leaver@pingcap.com",
                    },
                },
            )
        )
        resp = await client.post(
            "/api/v1/auth/feishu/verify",
            json={
                "code": "code-bl",
                "state": state,
                "redirect_uri": _VALID_REDIRECT_URI,
            },
        )
    assert resp.status_code == 401, resp.text