"""Identity-consumer self-service request flow.

Pins:

* form-builder maps logical fields to template widget IDs by name keyword
  and resolves single-choice (radioV2) values to the option machine key
* submit validates the grant + PII justification + slug availability and
  persists a pending request (Feishu instance creation stubbed — it's an
  external call mocked platform-wide)
* approval callback provisions the live consumer (copying scopes +
  granted_fields), sets consumer_id, marks the request approved
* a slug taken between submit and approval marks the request FAILED, not
  a half-built consumer
* the no-login one-time token-issue link mints exactly one token, then
  refuses replay
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime

import pytest

from app.core.security import create_action_token
from app.identity.consumers import request_service as rsvc
from app.identity.consumers.models import (
    REQUEST_STATUS_APPROVED,
    REQUEST_STATUS_FAILED,
    REQUEST_STATUS_PENDING,
    REQUEST_STATUS_REJECTED,
    IdentityConsumer,
    IdentityConsumerRequest,
)
from app.identity.consumers.request_schemas import ConsumerRequestCreate
from app.identity.consumers.request_service import (
    ACTION_TOKEN_PURPOSE_ISSUE,
    ConsumerRequestService,
)
from app.identity.person_models import IdentityPerson

ISSUE = "/api/v1/open/consumer-token/issue"


# ── Pure unit: form builder + option resolution ──────────────────────


def test_match_field_disambiguates_scope_vs_population():
    assert rsvc._match_field("Requested Data Scope") == "scope"
    assert rsvc._match_field("Population Scope") == "population"
    assert rsvc._match_field("Includes Plaintext PII") == "pii_flag"
    assert rsvc._match_field("Token Validity") == "validity"
    assert rsvc._match_field("Totally Unknown") is None


def test_resolve_option_returns_machine_key():
    widget = {
        "type": "radioV2",
        "options": [
            {"value": "k_yes", "text": "Yes"},
            {"value": "k_no", "text": "No"},
        ],
    }
    assert rsvc._resolve_option(widget, "Yes") == "k_yes"
    assert rsvc._resolve_option(widget, "no") == "k_no"
    assert rsvc._resolve_option(widget, "maybe") is None


def test_build_form_payload_maps_and_resolves():
    import json

    widgets = [
        {"id": "w1", "type": "input", "name": "Consumer System Name"},
        {"id": "w2", "type": "input", "name": "System Slug"},
        {"id": "w3", "type": "radioV2", "name": "Includes Plaintext PII",
         "options": [{"value": "ky", "text": "Yes"}, {"value": "kn", "text": "No"}]},
        {"id": "w4", "type": "radioV2", "name": "Token Validity",
         "options": [{"value": "d90", "text": "90d"}, {"value": "nev", "text": "Never"}]},
        {"id": "w5", "type": "text", "name": "说明"},  # must be skipped
    ]
    values = {
        "name": "Huilianyi", "slug": "huilianyi",
        "pii_flag": "No", "validity": "90d",
    }
    payload = json.loads(ConsumerRequestService.__new__(
        ConsumerRequestService,
    )._build_form_payload(widgets, values))
    by_id = {p["id"]: p for p in payload}
    assert by_id["w1"]["value"] == "Huilianyi"
    assert by_id["w3"]["value"] == "kn"   # resolved to option key
    assert by_id["w4"]["value"] == "d90"
    assert "w5" not in by_id              # text widget skipped


# ── Helpers ───────────────────────────────────────────────────────────


async def _person(db, *, email: str, status: str = "active") -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(), work_email=email, display_name=email.split("@")[0],
        employment_status=status, source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:10]}", employee_no=f"P{uuid.uuid4().hex[:5]}",
    )
    db.add(p)
    await db.flush()
    return p


def _stub_open_approval(monkeypatch):
    """Replace the Feishu instance-creation step with an in-DB stub."""
    async def _stub(self, req, requester, owner, granted_fields, *, trace_id):
        from app.approval.models import ApprovalSubmissionIntent

        req.feishu_instance_code = f"inst-{uuid.uuid4().hex[:8]}"
        intent = ApprovalSubmissionIntent(
            scope="identity_consumer_request",
            applicant_id=requester.id,
            idempotency_key=str(uuid.uuid4()),
            payload_fingerprint=uuid.uuid4().hex,
            provider_request_uuid=str(uuid.uuid4()),
            state="provider_created",
            external_id=req.feishu_instance_code,
            payload={
                "trace_id": trace_id,
                "provider": {
                    "approval_code": "stub-code",
                    "identity_type": "user_id",
                    "identity_value": "stub-user",
                    "form": "[]",
                },
                "finalization": {
                    "consumer_request_id": str(req.id),
                    "approval_request_id": str(uuid.uuid4()),
                    "name": req.name,
                    "slug": req.slug,
                    "description": req.description,
                    "owner_person_id": str(owner.id),
                    "owner_work_email": owner.work_email,
                    "scopes": req.scopes,
                    "granted_fields": granted_fields,
                    "filter_config": req.filter_config,
                    "rate_limit_per_minute": req.rate_limit_per_minute,
                    "token_validity_days": req.token_validity_days,
                    "pii_justification": req.pii_justification,
                    "approval_code": "stub-code",
                },
            },
        )
        self.db.add(intent)
        await self.db.flush()
        return intent
    monkeypatch.setattr(ConsumerRequestService, "_open_approval", _stub)


def _body(**over) -> ConsumerRequestCreate:
    base = dict(
        name="Huilianyi", slug=f"huilianyi-{uuid.uuid4().hex[:6]}",
        scopes=["directory.persons:read"], description="expense pull",
    )
    base.update(over)
    return ConsumerRequestCreate(**base)


# ── Submit ────────────────────────────────────────────────────────────


async def test_submit_persists_pending_request(db_session, monkeypatch):
    _stub_open_approval(monkeypatch)
    requester = await _person(db_session, email="req@pingcap.com")
    await db_session.commit()
    req = await ConsumerRequestService(db_session).submit(
        requester=requester, body=_body(), trace_id="t",
    )
    assert req.status == REQUEST_STATUS_PENDING
    assert req.feishu_instance_code
    assert req.owner_person_id == requester.id   # defaults to requester
    assert "id" in req.granted_fields            # always-on present


@pytest.mark.asyncio
async def test_open_approval_creates_reaper_visible_pending(db_session, monkeypatch):
    """Regression (2026-06-16): the platform ApprovalRequest the consumer
    flow opens MUST carry the canonical ``"pending"`` status — NOT the
    consumer-row's ``"pending_approval"`` — or the lost-webhook reaper
    (``feishu.approval_pending_reaper`` filters ``status == "pending"``)
    silently skips it and the applicant is never told it expired.

    Unlike the stubbed tests above, this drives the REAL ``_open_approval``
    with only the Feishu client + flow lookup mocked, so the actual
    ApprovalRequest-creation line is exercised.
    """
    from app.approval.models import ApprovalRequest
    from app.identity.consumers.request_service import APPROVAL_REQUEST_PENDING

    class _FakeFlow:
        external_code = "CONSUMER_CODE"

    class _FakeClient:
        async def get_approval_definition(self, code):
            return {"form": "[]"}

        async def resolve_user_id_by_email(self, email):
            return "u-1"

        async def create_instance(self, **kw):
            return {"instance_code": f"inst-{uuid.uuid4().hex[:8]}"}

    async def _flow(*_a, **_k):
        return _FakeFlow()

    async def _client(self):
        return _FakeClient()

    monkeypatch.setattr(rsvc, "get_approval_flow", _flow)
    monkeypatch.setattr(ConsumerRequestService, "_client", _client)

    requester = await _person(db_session, email="reaper@pingcap.com")
    await db_session.commit()
    req = await ConsumerRequestService(db_session).submit(
        requester=requester, body=_body(), trace_id="t",
    )
    ar = await db_session.get(ApprovalRequest, req.approval_request_id)
    assert ar is not None
    assert ar.status == "pending", "reaper filters status=='pending'"
    assert APPROVAL_REQUEST_PENDING == "pending"


async def test_submit_requires_pii_justification(db_session, monkeypatch):
    _stub_open_approval(monkeypatch)
    requester = await _person(db_session, email="req2@pingcap.com")
    await db_session.commit()
    from app.core.exceptions import ValidationError
    with pytest.raises(ValidationError):
        await ConsumerRequestService(db_session).submit(
            requester=requester,
            body=_body(
                scopes=["directory.persons:read", "directory.pii:read"],
                granted_fields=["work_email", "mobile_phone"],
            ),
            trace_id="t",
        )


# ── Callback → provision ──────────────────────────────────────────────


async def _make_pending(db_session, monkeypatch, **body_over):
    _stub_open_approval(monkeypatch)
    requester = await _person(db_session, email=f"r-{uuid.uuid4().hex[:6]}@pingcap.com")
    await db_session.commit()
    req = await ConsumerRequestService(db_session).submit(
        requester=requester, body=_body(**body_over), trace_id="t",
    )
    from app.approval.models import ApprovalRequest
    ar = await db_session.get(ApprovalRequest, req.approval_request_id)
    return req, ar


async def test_callback_approved_provisions_consumer(db_session, monkeypatch):
    req, ar = await _make_pending(
        db_session, monkeypatch,
        scopes=["directory.persons:read", "directory.pii:read"],
        granted_fields=["work_email", "mobile_phone"],
        pii_justification="expense reconciliation needs phone",
    )
    await ConsumerRequestService(db_session).handle_feishu_callback(
        approval_request=ar, instance_detail={"status": "APPROVED"},
    )
    await db_session.flush()
    assert req.status == REQUEST_STATUS_APPROVED
    assert req.consumer_id is not None
    consumer = await db_session.get(IdentityConsumer, req.consumer_id)
    assert consumer.slug == req.slug
    assert "mobile_phone" in consumer.granted_fields
    assert consumer.scopes == req.scopes


async def test_callback_rejected_marks_rejected(db_session, monkeypatch):
    req, ar = await _make_pending(db_session, monkeypatch)
    await ConsumerRequestService(db_session).handle_feishu_callback(
        approval_request=ar, instance_detail={"status": "REJECTED"},
    )
    assert req.status == REQUEST_STATUS_REJECTED
    assert req.consumer_id is None


async def test_callback_slug_taken_marks_failed(db_session, monkeypatch):
    req, ar = await _make_pending(db_session, monkeypatch)
    # Someone registers the slug before approval lands.
    db_session.add(IdentityConsumer(
        name="squatter", slug=req.slug, owner_person_id=req.requester_person_id,
        scopes=["directory.org:read"], granted_fields=[], filter_config={},
    ))
    await db_session.flush()
    await ConsumerRequestService(db_session).handle_feishu_callback(
        approval_request=ar, instance_detail={"status": "APPROVED"},
    )
    assert req.status == REQUEST_STATUS_FAILED
    assert req.consumer_id is None
    assert "slug" in (req.reject_reason or "").lower()


# ── One-time token-issue link ─────────────────────────────────────────


async def test_issue_link_mints_once_then_refuses(client, db_session, monkeypatch):
    owner = await _person(db_session, email="owner-issue@pingcap.com")
    consumer = IdentityConsumer(
        name="Issue Co", slug=f"issue-{uuid.uuid4().hex[:6]}",
        owner_person_id=owner.id, scopes=["directory.persons:read"],
        granted_fields=["id", "updated_at", "work_email"], filter_config={},
    )
    db_session.add(consumer)
    await db_session.flush()
    cid = consumer.id
    await db_session.commit()

    token = create_action_token(
        purpose=ACTION_TOKEN_PURPOSE_ISSUE,
        subject_id=str(owner.id), resource_id=str(cid),
    )
    # GET renders the confirm page (no side effect).
    r = await client.get(ISSUE, params={"token": token})
    assert r.status_code == 200 and "生成" in r.text

    # POST mints the first token, shows plaintext once.
    r = await client.post(ISSUE, data={"token": token})
    assert r.status_code == 200
    assert "helix_svc_" in r.text

    # Replay → refused (already issued).
    r = await client.post(ISSUE, data={"token": token})
    assert r.status_code == 200
    assert "已领取" in r.text or "已发放" in r.text


async def test_issue_link_rejects_bad_token(client):
    r = await client.get(ISSUE, params={"token": "garbage"})
    assert r.status_code == 200 and "无效" in r.text


async def test_issue_link_single_use_survives_token_revocation(client, db_session):
    """Security regression: once the link mints, revoking that token must
    NOT re-open the no-login mint path (the request's token_issued_at flag
    is authoritative, independent of any active token)."""
    from sqlalchemy import select as _select

    from app.identity.consumers.models import (
        REQUEST_STATUS_APPROVED,
        IdentityConsumerToken,
    )

    owner = await _person(db_session, email="su-revoke@pingcap.com")
    consumer = IdentityConsumer(
        name="Revoke Co", slug=f"rev-{uuid.uuid4().hex[:6]}",
        owner_person_id=owner.id, scopes=["directory.persons:read"],
        granted_fields=["id", "updated_at", "work_email"], filter_config={},
    )
    db_session.add(consumer)
    await db_session.flush()
    cid = consumer.id
    # The originating request (carries the single-use marker).
    db_session.add(IdentityConsumerRequest(
        requester_person_id=owner.id, name=consumer.name, slug=consumer.slug,
        owner_person_id=owner.id, scopes=["directory.persons:read"],
        granted_fields=["id", "updated_at", "work_email"], filter_config={},
        status=REQUEST_STATUS_APPROVED, consumer_id=cid,
    ))
    await db_session.flush()
    await db_session.commit()

    token = create_action_token(
        purpose=ACTION_TOKEN_PURPOSE_ISSUE,
        subject_id=str(owner.id), resource_id=str(cid),
    )
    # First mint succeeds.
    r = await client.post(ISSUE, data={"token": token})
    assert r.status_code == 200 and "helix_svc_" in r.text

    # Admin revokes the freshly-minted token.
    tok = (await db_session.execute(
        _select(IdentityConsumerToken).where(
            IdentityConsumerToken.consumer_id == cid,
        )
    )).scalars().first()
    tok.revoked_at = datetime.now(UTC)
    await db_session.commit()

    # Replay the SAME link — must be refused (flag set), NOT re-minted,
    # even though the consumer now has zero active tokens.
    r = await client.post(ISSUE, data={"token": token})
    assert r.status_code == 200
    assert "helix_svc_" not in r.text
    assert "已发放" in r.text or "已领取" in r.text


# ── Admin firehose / detail visibility ────────────────────────────────


def test_is_admin_reader_uses_canonical_role_names():
    """Regression: ``_is_admin_reader`` once read a non-existent ``.roles``
    attribute via ``getattr`` and so ALWAYS returned False — silently
    fencing admins/auditors/management out of the org-wide firehose and the
    cross-requester detail endpoint. It must read the canonical
    ``role_names`` property (already alias-canonicalised)."""
    from app.identity.consumers.request_router import _is_admin_reader

    def _with_roles(*names: str) -> IdentityPerson:
        p = IdentityPerson(id=uuid.uuid4(), work_email="x@pingcap.com")
        # ``role_names`` honours this request-scoped override first.
        p.__dict__["_role_names_override"] = list(names)
        return p

    assert _is_admin_reader(_with_roles("global_admin")) is True
    assert _is_admin_reader(_with_roles("auditor")) is True
    assert _is_admin_reader(_with_roles("management")) is True
    # Legacy aliases canonicalise on the way in.
    assert _is_admin_reader(_with_roles("it_admin")) is True
    assert _is_admin_reader(_with_roles("security_auditor")) is True
    assert _is_admin_reader(_with_roles("employee")) is False
    assert _is_admin_reader(_with_roles()) is False


async def test_admin_sees_other_requester_via_all_and_detail(
    client, db_session, admin_headers,
):
    """End-to-end: a global admin must see another person's consumer
    request through ``?all=true`` and the ``/{id}`` detail endpoint —
    the exact paths the ``.roles`` bug broke."""
    requester = await _person(db_session, email="someone-else@pingcap.com")
    row = IdentityConsumerRequest(
        requester_person_id=requester.id, name="Other Co",
        slug=f"other-{uuid.uuid4().hex[:6]}", owner_person_id=requester.id,
        scopes=["directory.persons:read"], granted_fields=["id"],
        filter_config={}, status=REQUEST_STATUS_PENDING,
    )
    db_session.add(row)
    await db_session.flush()
    rid = row.id
    await db_session.commit()

    base = "/api/v1/identity/consumer-requests"
    r = await client.get(base, params={"all": "true"}, headers=admin_headers)
    assert r.status_code == 200
    ids = {item["id"] for item in r.json()["data"]}
    assert str(rid) in ids, "admin firehose must include another requester's row"

    r = await client.get(f"{base}/{rid}", headers=admin_headers)
    assert r.status_code == 200
    assert r.json()["data"]["id"] == str(rid)


async def test_non_admin_all_only_sees_own_and_detail_403(
    client, db_session, employee_headers, employee_user,
):
    """The non-admin contract is unchanged: ``?all=true`` collapses to own
    rows and another requester's detail is 403."""
    me, _ = employee_user
    other = await _person(db_session, email="not-me@pingcap.com")
    other_row = IdentityConsumerRequest(
        requester_person_id=other.id, name="Not Mine",
        slug=f"notmine-{uuid.uuid4().hex[:6]}", owner_person_id=other.id,
        scopes=["directory.persons:read"], granted_fields=["id"],
        filter_config={}, status=REQUEST_STATUS_PENDING,
    )
    db_session.add(other_row)
    await db_session.flush()
    other_id = other_row.id
    await db_session.commit()

    base = "/api/v1/identity/consumer-requests"
    r = await client.get(base, params={"all": "true"}, headers=employee_headers)
    assert r.status_code == 200
    ids = {item["id"] for item in r.json()["data"]}
    assert str(other_id) not in ids

    r = await client.get(f"{base}/{other_id}", headers=employee_headers)
    assert r.status_code == 403


async def test_callback_completed_provisions(db_session, monkeypatch):
    """Feishu COMPLETED is a terminal 'approved' state for some templates —
    it must provision, not leave the request stuck pending."""
    req, ar = await _make_pending(db_session, monkeypatch)
    await ConsumerRequestService(db_session).handle_feishu_callback(
        approval_request=ar, instance_detail={"status": "COMPLETED"},
    )
    await db_session.flush()
    assert req.status == REQUEST_STATUS_APPROVED
    assert req.consumer_id is not None