"""Assistant write path: propose, show, confirm, execute.

These assert the four properties that make a confirmation mean something —
the server owns the parameters, the press is bound to what was shown, the link
belongs to one person for a short while, and one press submits exactly once.
"""
from __future__ import annotations

import json
import uuid
from datetime import UTC, datetime, timedelta

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

from app.assistant.actions import ACTION_SUPPORT_TICKET, get_action
from app.assistant.models import (
    STATUS_PENDING,
    AssistantActionProposal,
)
from app.assistant.proposals import (
    create_proposal,
)
from app.domains.support.models import SupportTicket
from tests.test_assistant.conftest import make_feishu_person

pytestmark = pytest.mark.asyncio

SHOW_URL = "/api/v1/assistant/proposals"

VALID_PARAMS = {
    "title": "笔记本无法开机",
    "body": "按电源键没有反应，充电器换过了。",
    "category": "hardware",
    "priority": "high",
}


def _as(person) -> dict[str, str]:
    """Headers for the proposal's OWNER.

    The confirm route no longer accepts the signed token alone: a Feishu card
    is forwardable and the token rides inside it, so the link was a credential
    anyone the card reached could use. The signed-in person must BE the
    proposal's person.
    """
    from sqlalchemy import inspect as sa_inspect

    from app.core.security import create_access_token

    # ``inspect(...).identity`` rather than ``person.id``: a request that
    # rolled back (the failed-action test does exactly that) leaves the ORM
    # instance expired, and touching an attribute would emit IO outside the
    # greenlet — surfacing as MissingGreenlet instead of the failure under
    # test. The identity is held on the instance state and needs no refresh.
    identity = sa_inspect(person).identity
    person_id = str(identity[0]) if identity else str(person.id)
    return {
        "Authorization": f"Bearer {create_access_token(person_id, ['employee'])}",
    }


async def _issue(db: AsyncSession, person, params=None):
    action = get_action(ACTION_SUPPORT_TICKET)
    assert action is not None
    issued = await create_proposal(
        db,
        person=person,
        action_key=ACTION_SUPPORT_TICKET,
        params=action.normalise(params or VALID_PARAMS),
        conversation_id=None,
        trace_id="t-propose",
    )
    await db.commit()
    return issued


# ── showing the offer ───────────────────────────────────────────────


async def test_show_renders_every_row_without_consuming_the_token(
    client: AsyncClient, db_session: AsyncSession, enabled_gate, feishu_person,
):
    issued = await _issue(db_session, feishu_person)

    first = await client.get(SHOW_URL, params={"token": issued.token, "lang": "zh"}, headers=_as(feishu_person))
    second = await client.get(SHOW_URL, params={"token": issued.token, "lang": "zh"}, headers=_as(feishu_person))

    assert first.status_code == 200
    assert second.status_code == 200
    body = first.json()["data"]
    # No ``actionable``: the page has no button. A live offer simply reads as
    # pending, and the employee confirms from the card.
    assert body["status"] == "pending"
    assert "actionable" not in body
    assert body["params_hash"] == issued.proposal.params_hash
    assert any(VALID_PARAMS["title"] in row["value"] for row in body["rows"])


async def test_show_is_localized(
    client: AsyncClient, db_session: AsyncSession, enabled_gate, feishu_person,
):
    issued = await _issue(db_session, feishu_person)

    zh = await client.get(SHOW_URL, params={"token": issued.token, "lang": "zh"}, headers=_as(feishu_person))
    en = await client.get(SHOW_URL, params={"token": issued.token, "lang": "en"}, headers=_as(feishu_person))

    assert zh.json()["data"]["title"] != en.json()["data"]["title"]


async def test_show_refuses_a_forged_token(
    client: AsyncClient, db_session: AsyncSession, enabled_gate, feishu_person,
):
    response = await client.get(SHOW_URL, params={"token": "not-a-token"})
    assert response.status_code == 401


async def test_gate_closed_hides_the_surface_entirely(
    client: AsyncClient, db_session: AsyncSession, feishu_person,
):
    issued = await _issue(db_session, feishu_person)
    response = await client.get(SHOW_URL, params={"token": issued.token}, headers=_as(feishu_person))
    assert response.status_code == 404


# ── confirming ──────────────────────────────────────────────────────


async def test_proposal_starts_pending_and_nothing_happens_until_confirmed(
    db_session: AsyncSession, enabled_gate, feishu_person,
):
    issued = await _issue(db_session, feishu_person)

    assert issued.proposal.status == STATUS_PENDING
    assert (await db_session.execute(select(SupportTicket))).scalars().first() is None
    assert uuid.UUID(str(issued.proposal.id))


async def test_an_expired_proposal_reads_as_lapsed(
    client: AsyncClient, db_session: AsyncSession, enabled_gate, feishu_person,
):
    """Rendering an enabled button on something confirm will refuse is a trap."""
    issued = await _issue(db_session, feishu_person)
    proposal = await db_session.get(AssistantActionProposal, issued.proposal.id)
    assert proposal is not None
    proposal.expires_at = datetime.now(UTC) - timedelta(minutes=1)
    await db_session.commit()

    body = (
        await client.get(SHOW_URL, params={"token": issued.token}, headers=_as(feishu_person))
    ).json()["data"]

    assert body["status"] == "expired"


async def test_a_colleagues_session_cannot_read_your_proposal(
    client: AsyncClient, db_session: AsyncSession, enabled_gate, feishu_person,
):
    """The render route leaked more than the card it claimed to mirror: 613
    characters of the employee's free-text body against the card's 120, plus
    the execution ``result``, which the card never carried at all.

    Fixed by making the claim TRUE rather than by adding a login wall — the
    page has to open in the Feishu in-app browser, where silent re-auth is
    known to fail on iOS.
    """

    issued = await _issue(db_session, feishu_person)
    colleague = await make_feishu_person(
        db_session, subject="on_colleague_read", email="colleague.read@test.com",
    )

    long_body = "工单正文 " * 80
    issued = await _issue(
        db_session, feishu_person, params={**VALID_PARAMS, "body": long_body},
    )

    # A reader with no session still gets the page — the Feishu in-app browser
    # cannot silently re-auth on iOS — but sees the card's own truncation.
    anonymous = await client.get(SHOW_URL, params={"token": issued.token})
    assert anonymous.status_code == 200
    anon_body = json.dumps(anonymous.json()["data"], ensure_ascii=False)
    assert "…" in anon_body
    assert long_body.strip() not in anon_body
    assert anonymous.json()["data"]["result"] is None

    # A colleague WITH a session is still just a non-owner.
    other = await client.get(
        SHOW_URL, params={"token": issued.token}, headers=_as(colleague),
    )
    assert json.dumps(other.json()["data"], ensure_ascii=False) == anon_body

    # The owner sees the whole thing, which is what the page is for.
    owner = await client.get(
        SHOW_URL, params={"token": issued.token}, headers=_as(feishu_person),
    )
    assert owner.status_code == 200
    assert long_body.strip() in json.dumps(owner.json()["data"], ensure_ascii=False)


async def test_the_token_does_not_outlive_the_offer(
    db_session: AsyncSession, feishu_person,
):
    """A 24h token on a 60-minute proposal is a standing credential: the
    forwarded card kept reading the confirmation page — including the
    post-execution result — for 23 hours after the offer was dead."""
    import jwt

    from app.assistant.proposals import PROPOSAL_TTL_MINUTES

    issued = await _issue(db_session, feishu_person)
    payload = jwt.decode(issued.token, options={"verify_signature": False})
    assert (payload["exp"] - payload["iat"]) / 60 <= PROPOSAL_TTL_MINUTES