"""External-origin Feishu approval callback path (PR #254 / migration 149).

Until 149, ``ApprovalService.handle_feishu_callback`` silently dropped
any webhook whose ``instance_code`` wasn't in ``approval_requests`` —
the handler assumed every flow was Helix-initiated. The SOC2 邮件组
flow breaks that assumption: applicants submit directly in Feishu
and IT used to manually paste them into ``soc2@pingcap.com``.

The new path:

1. Feishu pushes ``approval_instance`` for an instance Helix never
   saw → handler synthesizes the parent ``approval_requests`` row
   ONLY when the matching ``integration_approval_flows`` row has
   ``accept_external_origin = TRUE``.
2. Dispatcher routes by ``flow_key`` exactly as for Helix-initiated
   flows.
3. ``Soc2EmailGroupService.handle_feishu_callback`` resolves the
   applicant's email from Feishu and calls
   ``GoogleWorkspaceDriver._group_add``.

These tests pin all three branches:

* opted-in flow → synthesize + dispatch + group add
* flow exists but NOT opted in → silent drop, no row
* no matching flow at all → silent drop, no row
"""
from __future__ import annotations

import uuid
from unittest.mock import AsyncMock, patch

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

from app.approval.models import ApprovalRequest
from app.approval.service import ApprovalService
from app.integrations.driver_base import ProvisioningResult
from app.integrations.models import Integration, IntegrationApprovalFlow

_SOC2_CODE = "9A42C433-6A4B-411C-9B2E-9182BEEBC6BD"
_FEISHU_CFG = {"app_id": "cli_test", "app_secret": "secret_test"}


async def _seed_feishu(db_session, *, accept_external: bool, flow_key="soc2_email_group_grant"):
    integ = Integration(
        id=uuid.uuid4(),
        kind="chat_approval",
        provider="feishu",
        name="Feishu Test",
        slug="feishu-test-primary-ext",
        enabled=True,
        is_primary=True,
        config_public={"app_id": "cli_test"},
    )
    db_session.add(integ)
    await db_session.flush()
    flow = IntegrationApprovalFlow(
        id=uuid.uuid4(),
        integration_id=integ.id,
        flow_key=flow_key,
        label="SOC2 邮件组授权",
        external_code=_SOC2_CODE,
        active=True,
        accept_external_origin=accept_external,
    )
    db_session.add(flow)
    await db_session.commit()
    return integ, flow


async def _seed_google_workspace(db_session):
    """Minimal Google Workspace integration so ``_get_google_driver``
    resolves. The actual driver methods are patched per-test."""
    sa_json = (
        '{"type": "service_account", "project_id": "p", "private_key_id": "k", '
        '"private_key": "-----BEGIN PRIVATE KEY-----\\nMIIBOg==\\n-----END PRIVATE KEY-----\\n", '
        '"client_email": "sa@p.iam.gserviceaccount.com", "client_id": "1", '
        '"token_uri": "https://oauth2.googleapis.com/token"}'
    )
    integ = Integration(
        id=uuid.uuid4(),
        kind="software_provisioning",
        provider="google_workspace",
        name="Google Workspace Test",
        slug="gw-test-primary",
        enabled=True,
        is_primary=True,
        config_public={
            "service_account_json": sa_json,
            "impersonate_subject_email": "admin@pingcap.com",
        },
    )
    db_session.add(integ)
    await db_session.commit()
    return integ


@pytest.fixture(autouse=True)
async def _patch_async_session_factory(db_engine):
    """Repoint ``app.core.database.async_session_factory`` at the
    test engine so the SOC2 service's fresh-session audit writes
    (round-22 review fix) land in the test DB instead of the
    module-level production engine. Mirrors the same shim used in
    ``test_scheduled_jobs/test_failure_ack_and_dedup.py``."""
    from sqlalchemy.ext.asyncio import async_sessionmaker

    from app.core import database as _db

    original = _db.async_session_factory
    _db.async_session_factory = async_sessionmaker(
        db_engine, class_=AsyncSession, expire_on_commit=False,
    )
    try:
        yield
    finally:
        _db.async_session_factory = original


@pytest.fixture(autouse=True)
def _patch_feishu_cfg(monkeypatch):
    async def _fake_cfg(**_):
        return dict(_FEISHU_CFG)

    monkeypatch.setattr("app.integrations.compat.get_feishu_config", _fake_cfg)
    monkeypatch.setattr("app.approval.service.get_feishu_config", _fake_cfg)
    monkeypatch.setattr(
        "app.domains.soc2_email_group.service.get_feishu_config", _fake_cfg,
    )

    async def _fake_token(self):
        return "fake-tenant-token"
    monkeypatch.setattr(
        "app.approval.feishu_client.FeishuApprovalClient.get_tenant_access_token",
        _fake_token,
    )

    # The synthesizer + lifespan sweep both call ``ensure_subscribed``.
    # In production it POSTs to ``/approval/v4/approvals/{code}/subscribe``;
    # under tests we short-circuit to avoid real Feishu network traffic
    # (the production code catches the exception either way, but a
    # fake-token POST against open.feishu.cn still adds latency + log
    # noise + fails in CI without internet).
    async def _fake_subscribe(self, approval_code):
        return None
    monkeypatch.setattr(
        "app.approval.feishu_client.FeishuApprovalClient.ensure_subscribed",
        _fake_subscribe,
    )


@pytest.mark.asyncio
async def test_external_origin_synthesizes_and_dispatches_soc2(db_session):
    """Happy path: opted-in flow → row synthesized → SOC2 handler runs
    and asks the Google driver to add the applicant to the group."""
    await _seed_feishu(db_session, accept_external=True)
    await _seed_google_workspace(db_session)

    instance_code = "feishu_external_soc2_001"
    applicant_email = "applicant@pingcap.com"
    group_add_calls: list[tuple[str, str]] = []

    async def _fake_group_add(self, ctx):
        group_add_calls.append((ctx.user_email, str((ctx.extra or {}).get("group_email"))))
        return ProvisioningResult(ok=True, external_ref=f"ref:{ctx.user_email}")

    with (
        patch(
            "app.approval.service.FeishuApprovalClient.get_instance_detail",
            new=AsyncMock(return_value={
                "instance_code": instance_code,
                "approval_code": _SOC2_CODE,
                "status": "APPROVED",
                "open_id": "ou_FAKE_OPENID",
                "user_id": "FAKE_USERID",
            }),
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.get_user_with_custom_attrs",
            new=AsyncMock(return_value={
                "user_id": "FAKE_USERID",
                "name": "Test Applicant",
                "email": applicant_email,
            }),
        ),
        patch(
            "app.integrations.drivers.google_workspace.GoogleWorkspaceDriver._group_add",
            new=_fake_group_add,
        ),
    ):
        svc = ApprovalService(db_session)
        await svc.handle_feishu_callback({
            "header": {"event_type": "approval_instance"},
            "event": {
                "instance_code": instance_code,
                "approval_code": _SOC2_CODE,
                "status": "APPROVED",
            },
        })
        await db_session.commit()

    # Parent row synthesized + marked approved.
    row = (await db_session.execute(
        select(ApprovalRequest).where(ApprovalRequest.external_id == instance_code)
    )).scalar_one()
    assert row.approval_type == "soc2_email_group_grant"
    assert row.status == "approved"
    assert (row.source_payload or {}).get("external_origin") is True

    # Google driver was asked to add applicant to soc2@pingcap.com.
    assert group_add_calls == [(applicant_email, "soc2@pingcap.com")]


@pytest.mark.asyncio
async def test_external_origin_silent_drop_when_flow_not_opted_in(db_session):
    """A flow row exists for the approval_code but
    ``accept_external_origin=False`` → no synthesis, no row, no driver
    call. Preserves the strict contract for the 13 Helix-initiated
    flow_keys (default behavior must NOT change)."""
    await _seed_feishu(
        db_session, accept_external=False, flow_key="not_opted_in_flow",
    )

    instance_code = "feishu_external_soc2_002"
    with (
        patch(
            "app.approval.service.FeishuApprovalClient.get_instance_detail",
            new=AsyncMock(return_value={
                "instance_code": instance_code,
                "approval_code": _SOC2_CODE,
                "status": "APPROVED",
                "open_id": "ou_X",
            }),
        ),
        patch(
            "app.integrations.drivers.google_workspace.GoogleWorkspaceDriver._group_add",
            new=AsyncMock(),
        ) as group_add_mock,
    ):
        svc = ApprovalService(db_session)
        await svc.handle_feishu_callback({
            "header": {"event_type": "approval_instance"},
            "event": {
                "instance_code": instance_code,
                "approval_code": _SOC2_CODE,
                "status": "APPROVED",
            },
        })

    # No ApprovalRequest synthesized.
    row = (await db_session.execute(
        select(ApprovalRequest).where(ApprovalRequest.external_id == instance_code)
    )).scalar_one_or_none()
    assert row is None
    group_add_mock.assert_not_called()


@pytest.mark.asyncio
async def test_external_origin_no_matching_flow_silent_drop(db_session):
    """Webhook for an approval_code Helix has never heard of → no row,
    no errors. Matches the prior 'unknown instance_code' behavior."""
    # Deliberately seed nothing — no IntegrationApprovalFlow rows.
    instance_code = "feishu_external_soc2_003"
    with (
        patch(
            "app.approval.service.FeishuApprovalClient.get_instance_detail",
            new=AsyncMock(return_value={
                "instance_code": instance_code,
                "approval_code": "UNKNOWN-TEMPLATE-CODE",
                "status": "APPROVED",
            }),
        ),
        patch(
            "app.integrations.drivers.google_workspace.GoogleWorkspaceDriver._group_add",
            new=AsyncMock(),
        ) as group_add_mock,
    ):
        svc = ApprovalService(db_session)
        await svc.handle_feishu_callback({
            "header": {"event_type": "approval_instance"},
            "event": {
                "instance_code": instance_code,
                "approval_code": "UNKNOWN-TEMPLATE-CODE",
                "status": "APPROVED",
            },
        })

    row = (await db_session.execute(
        select(ApprovalRequest).where(ApprovalRequest.external_id == instance_code)
    )).scalar_one_or_none()
    assert row is None
    group_add_mock.assert_not_called()


@pytest.mark.asyncio
async def test_external_origin_rejected_status_no_group_add(db_session):
    """Rejected/cancelled approvals must NOT add anyone to the group.
    Only APPROVED is a write trigger; everything else is a state
    record on the synthesized row only."""
    await _seed_feishu(db_session, accept_external=True)
    await _seed_google_workspace(db_session)

    instance_code = "feishu_external_soc2_004"
    with (
        patch(
            "app.approval.service.FeishuApprovalClient.get_instance_detail",
            new=AsyncMock(return_value={
                "instance_code": instance_code,
                "approval_code": _SOC2_CODE,
                "status": "REJECTED",
                "open_id": "ou_Y",
            }),
        ),
        patch(
            "app.integrations.drivers.google_workspace.GoogleWorkspaceDriver._group_add",
            new=AsyncMock(),
        ) as group_add_mock,
    ):
        svc = ApprovalService(db_session)
        await svc.handle_feishu_callback({
            "header": {"event_type": "approval_instance"},
            "event": {
                "instance_code": instance_code,
                "approval_code": _SOC2_CODE,
                "status": "REJECTED",
            },
        })
        await db_session.commit()

    row = (await db_session.execute(
        select(ApprovalRequest).where(ApprovalRequest.external_id == instance_code)
    )).scalar_one()
    assert row.status == "rejected"
    group_add_mock.assert_not_called()


@pytest.mark.asyncio
async def test_external_origin_leaver_refused_with_audit(db_session, db_engine):
    """A leaver / leaver_pending applicant must NEVER be auto-added to
    the SOC2 group — mirrors the SSO / VPN / scope-request leaver gates.
    The handler raises ``SOC2_APPLICANT_LEFT`` and writes a
    ``soc2_email_group_grant.refused_leaver`` audit row."""
    from app.audit.models import AuditLog
    from app.identity.person_models import IdentityPerson

    await _seed_feishu(db_session, accept_external=True)
    await _seed_google_workspace(db_session)

    # Insert a leaver Person so the resolver finds them by work_email.
    leaver_email = "leaver@pingcap.com"
    leaver = IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id="P-LEAVER",
        work_email=leaver_email,
        display_name="Test Leaver",
        employment_status="leaver_pending",
    )
    db_session.add(leaver)
    await db_session.commit()

    instance_code = "feishu_external_soc2_leaver"
    with (
        patch(
            "app.approval.service.FeishuApprovalClient.get_instance_detail",
            new=AsyncMock(return_value={
                "instance_code": instance_code,
                "approval_code": _SOC2_CODE,
                "status": "APPROVED",
                "open_id": "ou_LEAVER",
            }),
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.get_user_with_custom_attrs",
            new=AsyncMock(return_value={
                "user_id": "LEAVER_USERID",
                "email": leaver_email,
            }),
        ),
        patch(
            "app.integrations.drivers.google_workspace.GoogleWorkspaceDriver._group_add",
            new=AsyncMock(),
        ) as group_add_mock,
        pytest.raises(Exception) as excinfo,  # AppException
    ):
        svc = ApprovalService(db_session)
        await svc.handle_feishu_callback({
            "header": {"event_type": "approval_instance"},
            "event": {
                "instance_code": instance_code,
                "approval_code": _SOC2_CODE,
                "status": "APPROVED",
            },
        })

    # Group add never called; refused-leaver audit row exists.
    group_add_mock.assert_not_called()
    assert excinfo.value.code == "SOC2_APPLICANT_LEFT"
    # The refused-leaver audit row is written via a SEPARATE session
    # (round-22 fresh-session pattern) so it survives the rollback
    # that the AppException triggers in ``get_db()``. Read it back
    # from a FRESH session bound to the SAME test engine —
    # ``db_session``'s REPEATABLE-READ snapshot pre-dates the
    # standalone-commit and won't see the row.
    from sqlalchemy.ext.asyncio import async_sessionmaker

    factory = async_sessionmaker(
        db_engine, class_=type(db_session), expire_on_commit=False,
    )
    async with factory() as audit_check_session:
        refused = (await audit_check_session.execute(
            select(AuditLog).where(
                AuditLog.action == "soc2_email_group_grant.refused_leaver",
            )
        )).scalars().first()
    assert refused is not None
    assert refused.detail["employment_status"] == "leaver_pending"


@pytest.mark.asyncio
async def test_external_origin_already_member_409_is_success(db_session):
    """Google's ``members.insert`` returns 409 when the user is already
    in the group; the driver maps that to ``ok=True`` with
    ``raw_response.already_member=True``. The audit row must record
    ``detail.already_member=True`` so admins can tell a fresh add
    from a redelivery-replay."""
    from app.audit.models import AuditLog

    await _seed_feishu(db_session, accept_external=True)
    await _seed_google_workspace(db_session)

    instance_code = "feishu_external_soc2_409"
    applicant_email = "already-in@pingcap.com"

    async def _fake_409(self, ctx):
        return ProvisioningResult(
            ok=True,
            external_ref=f"{ctx.extra['group_email']}:{ctx.user_email}",
            raw_response={"already_member": True},
        )

    with (
        patch(
            "app.approval.service.FeishuApprovalClient.get_instance_detail",
            new=AsyncMock(return_value={
                "instance_code": instance_code,
                "approval_code": _SOC2_CODE,
                "status": "APPROVED",
                "open_id": "ou_DUPE",
            }),
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.get_user_with_custom_attrs",
            new=AsyncMock(return_value={"user_id": "x", "email": applicant_email}),
        ),
        patch(
            "app.integrations.drivers.google_workspace.GoogleWorkspaceDriver._group_add",
            new=_fake_409,
        ),
    ):
        svc = ApprovalService(db_session)
        await svc.handle_feishu_callback({
            "header": {"event_type": "approval_instance"},
            "event": {
                "instance_code": instance_code,
                "approval_code": _SOC2_CODE,
                "status": "APPROVED",
            },
        })
        await db_session.commit()

    row = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "soc2_email_group_grant.member_added",
        )
    )).scalars().first()
    assert row is not None
    assert row.detail["already_member"] is True


@pytest.mark.asyncio
async def test_external_origin_applicant_not_in_helix_proceeds_with_warning(db_session):
    """When the Feishu applicant has no ``identity_persons`` row (rare —
    contractor pre-HRIS-sync), the leaver gate can't apply. Service
    proceeds on Feishu's approval (logs a WARNING; we don't fail
    closed because that would refuse legit day-zero contractor
    onboarding)."""
    from app.audit.models import AuditLog

    await _seed_feishu(db_session, accept_external=True)
    await _seed_google_workspace(db_session)

    instance_code = "feishu_external_soc2_no_helix_person"
    applicant_email = "fresh-contractor@pingcap.com"
    group_add_called: list[str] = []

    async def _fake_add(self, ctx):
        group_add_called.append(ctx.user_email)
        return ProvisioningResult(ok=True, external_ref="ok")

    with (
        patch(
            "app.approval.service.FeishuApprovalClient.get_instance_detail",
            new=AsyncMock(return_value={
                "instance_code": instance_code,
                "approval_code": _SOC2_CODE,
                "status": "APPROVED",
                "open_id": "ou_NEW",
            }),
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.get_user_with_custom_attrs",
            new=AsyncMock(return_value={"user_id": "x", "email": applicant_email}),
        ),
        patch(
            "app.integrations.drivers.google_workspace.GoogleWorkspaceDriver._group_add",
            new=_fake_add,
        ),
    ):
        svc = ApprovalService(db_session)
        await svc.handle_feishu_callback({
            "header": {"event_type": "approval_instance"},
            "event": {
                "instance_code": instance_code,
                "approval_code": _SOC2_CODE,
                "status": "APPROVED",
            },
        })
        await db_session.commit()

    assert group_add_called == [applicant_email]
    row = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "soc2_email_group_grant.member_added",
        )
    )).scalars().first()
    assert row is not None
    assert row.detail["applicant_person_id"] is None


@pytest.mark.asyncio
async def test_external_origin_leaver_gate_matches_by_email_alias(
    db_session, db_engine,
):
    """A leaver whose Feishu account uses an address listed in
    ``identity_persons.email_aliases`` (not the primary ``work_email``)
    must STILL be refused. Round-22 review (P1) — without the alias
    check the leaver-gate would fail-open via the
    ``person is None → proceed-with-warning`` path."""
    from app.audit.models import AuditLog
    from app.identity.person_models import IdentityPerson

    await _seed_feishu(db_session, accept_external=True)
    await _seed_google_workspace(db_session)

    # Leaver whose ALIAS (not work_email) matches the Feishu email.
    leaver = IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id="P-ALIAS-LEAVER",
        work_email="primary@pingcap.com",
        display_name="Alias Leaver",
        employment_status="left",
        email_aliases=["alias@pingcap.com"],
    )
    db_session.add(leaver)
    await db_session.commit()

    instance_code = f"feishu_external_soc2_alias_leaver_{uuid.uuid4().hex[:8]}"
    with (
        patch(
            "app.approval.service.FeishuApprovalClient.get_instance_detail",
            new=AsyncMock(return_value={
                "instance_code": instance_code,
                "approval_code": _SOC2_CODE,
                "status": "APPROVED",
                "open_id": "ou_ALIAS",
            }),
        ),
        patch(
            "app.approval.feishu_client.FeishuApprovalClient.get_user_with_custom_attrs",
            new=AsyncMock(return_value={
                "user_id": "x",
                # Feishu echoes the alias address (matches `email_aliases`).
                "email": "alias@pingcap.com",
            }),
        ),
        patch(
            "app.integrations.drivers.google_workspace.GoogleWorkspaceDriver._group_add",
            new=AsyncMock(),
        ) as group_add_mock,
        pytest.raises(Exception) as excinfo,
    ):
        svc = ApprovalService(db_session)
        await svc.handle_feishu_callback({
            "header": {"event_type": "approval_instance"},
            "event": {
                "instance_code": instance_code,
                "approval_code": _SOC2_CODE,
                "status": "APPROVED",
            },
        })

    group_add_mock.assert_not_called()
    assert excinfo.value.code == "SOC2_APPLICANT_LEFT"
    # Audit row persists via the fresh-session pattern.
    from sqlalchemy.ext.asyncio import async_sessionmaker

    factory = async_sessionmaker(
        db_engine, class_=AsyncSession, expire_on_commit=False,
    )
    async with factory() as audit_check_session:
        refused = (await audit_check_session.execute(
            select(AuditLog).where(
                AuditLog.action == "soc2_email_group_grant.refused_leaver",
                AuditLog.detail["applicant_person_id"].as_string()
                    == str(leaver.id),
            )
        )).scalars().first()
    assert refused is not None
    assert refused.detail["employment_status"] == "left"


@pytest.mark.asyncio
async def test_external_origin_synthesize_race_recovery_branch(
    db_session, db_engine,
):
    """Pin the SAVEPOINT/IntegrityError recovery branch of
    ``_synthesize_external_origin_request``.

    Real concurrent webhooks for the same unknown ``instance_code``
    would race on the ``UNIQUE(external_id)`` constraint inside the
    synthesizer's ``async with self.db.begin_nested()`` block. The
    loser must catch ``IntegrityError``, the SAVEPOINT auto-rolls
    back, re-SELECT FOR UPDATE pulls the winner's row, and the
    function returns ``(existing_row, instance_detail)``.

    Pin via two sessions both bound to the test engine: session A
    reserves the row + commits (simulating the winner). Session B then
    invokes the synthesizer. Session B's initial ``add`` + nested-flush
    hits the UNIQUE constraint on the real DB — exercising the actual
    recovery path end-to-end, NOT a mock. Both sessions are bound to
    ``db_engine`` (the per-test engine) so they share the same event
    loop; using the production ``async_session_factory`` would reach
    into a separate engine whose pool would carry over stale loops
    between tests."""
    from sqlalchemy.ext.asyncio import async_sessionmaker

    await _seed_feishu(db_session, accept_external=True)
    await _seed_google_workspace(db_session)
    # db_session keeps an implicit tx open with a snapshot from before
    # session_a's commit; commit it so session_b is free to operate.
    await db_session.commit()

    instance_code = f"feishu_external_soc2_race_{uuid.uuid4().hex[:8]}"

    test_session_factory = async_sessionmaker(
        db_engine, class_=type(db_session), expire_on_commit=False,
    )

    # Session A — the "winner". Inserts and commits BEFORE session B
    # opens, so session B's snapshot WILL see it on a fresh SELECT.
    async with test_session_factory() as session_a:
        winner = ApprovalRequest(
            external_id=instance_code,
            approval_type="soc2_email_group_grant",
            applicant_id=None,
            status="pending",
            source_payload={"external_origin": True, "winner_marker": True},
        )
        session_a.add(winner)
        await session_a.commit()

    # Session B — the "loser". Its synthesizer's begin_nested INSERT
    # collides on UNIQUE(external_id), the SAVEPOINT rolls back, and
    # the re-SELECT FOR UPDATE finds the winner's row.
    async with test_session_factory() as session_b:
        svc = ApprovalService(session_b)
        with patch(
            "app.approval.service.FeishuApprovalClient.get_instance_detail",
            new=AsyncMock(return_value={
                "instance_code": instance_code,
                "approval_code": _SOC2_CODE,
                "status": "APPROVED",
                "open_id": "ou_RACE",
            }),
        ):
            result = await svc._synthesize_external_origin_request(
                instance_code=instance_code,
                payload={"event": {"instance_code": instance_code}},
            )

        # Recovery branch fired: returned the winner's row, not a
        # newly-INSERTed one.
        assert result is not None
        row, detail = result
        assert row.external_id == instance_code
        assert (row.source_payload or {}).get("winner_marker") is True
        assert detail.get("approval_code") == _SOC2_CODE

    # No duplicate row exists in DB.
    async with test_session_factory() as verify_session:
        rows = (await verify_session.execute(
            select(ApprovalRequest).where(
                ApprovalRequest.external_id == instance_code,
            )
        )).scalars().all()
        assert len(rows) == 1