"""Shared fixtures for the assistant channel tests."""
from __future__ import annotations

import asyncio
import uuid
from typing import Any

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

from app.assistant.settings import SETTING_ENABLED
from app.identity.person_models import IdentityExternalIdentity, IdentityPerson
from app.integrations.models import PlatformSetting

UNION_ID = "on_assistant_union_0001"
OPEN_ID = "ou_assistant_open_0001"


def message_event(
    *,
    text: str = "我的电脑保修到什么时候",
    event_id: str = "evt-0001",
    open_id: str = OPEN_ID,
    union_id: str | None = UNION_ID,
    chat_type: str = "p2p",
    message_type: str = "text",
    sender_type: str = "user",
    content: str | None = None,
) -> dict[str, Any]:
    """A realistic ``im.message.receive_v1`` v2 envelope."""
    import json

    return {
        "schema": "2.0",
        "header": {
            "event_id": event_id,
            "event_type": "im.message.receive_v1",
            "token": "verification-token",
            "app_id": "cli_test",
        },
        "event": {
            "sender": {
                "sender_type": sender_type,
                "sender_id": {
                    "open_id": open_id,
                    **({"union_id": union_id} if union_id else {}),
                },
            },
            "message": {
                "message_id": "om-0001",
                "chat_id": "oc-0001",
                "chat_type": chat_type,
                "message_type": message_type,
                "content": (
                    content
                    if content is not None
                    else json.dumps({"text": text}, ensure_ascii=False)
                ),
            },
        },
    }


async def set_gate(db: AsyncSession, value: Any) -> None:
    row = await db.get(PlatformSetting, SETTING_ENABLED)
    if row is None:
        db.add(PlatformSetting(key=SETTING_ENABLED, value=value))
    else:
        row.value = value
    await db.commit()


@pytest_asyncio.fixture
async def enabled_gate(db_session: AsyncSession):
    await set_gate(db_session, True)


async def make_feishu_person(
    db: AsyncSession,
    *,
    subject: str = UNION_ID,
    email: str = "assistant.user@test.com",
    name: str = "Assistant User",
    employment_status: str = "active",
    is_login_identity: bool = True,
) -> IdentityPerson:
    """A person plus their Feishu external identity, in one call.

    ``employment_status`` is set at INSERT rather than flipped afterwards: the
    Leaver Blocked before_flush listener rejects a live flip to ``left`` while
    blockers remain, which is production behaviour worth not fighting in tests.
    """
    person = IdentityPerson(
        id=uuid.uuid4(),
        work_email=email,
        display_name=name,
        employment_status=employment_status,
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:12]}",
    )
    db.add(person)
    await db.flush()
    db.add(
        IdentityExternalIdentity(
            person_id=person.id,
            provider="feishu",
            provider_subject=subject,
            provider_email=email,
            is_login_identity=is_login_identity,
            identity_status="active",
        ),
    )
    await db.commit()
    return person


@pytest_asyncio.fixture
async def feishu_person(db_session: AsyncSession) -> IdentityPerson:
    """An active employee with a Feishu LOGIN identity (union_id subject)."""
    return await make_feishu_person(db_session)


@pytest.fixture
def sent(monkeypatch) -> list[dict[str, Any]]:
    """Capture outbound replies instead of calling Feishu."""
    from app.assistant import channel

    captured: list[dict[str, Any]] = []

    async def _fake_send_text(db, *, to, text, idempotency_key=None):
        captured.append({
            # 保留 ``open_id`` 这个键名:几十条既有断言按它读。群消息在这里
            # 是空串,而 ``chat_id`` 有值 —— 一条断言因此能同时表达「发出去了」
            # 和「发到哪个面」。
            "open_id": to.open_id,
            "chat_id": to.chat_id,
            "to": to,
            "text": text,
            "idempotency_key": idempotency_key,
        })
        return True

    monkeypatch.setattr(channel, "send_text", _fake_send_text)
    return captured


class FakeCopilotResult:
    """Stand-in for ``CopilotResult`` — same attributes the channel reads."""

    def __init__(
        self,
        reply: str = "你的电脑保修到 2027-03-01。",
        *,
        available: bool = True,
        tool_calls: list[str] | None = None,
        provider: str | None = "bailian",
        turn_failed: bool = False,
    ) -> None:
        self.available = available
        self.reply = reply
        self.tool_calls = tool_calls or []
        self.suggest_escalate = False
        self.provider = provider
        # Mirrors the real ``CopilotResult``: a stand-in that omits a field the
        # channel reads would make the channel's own branch untestable, and
        # would blow up with AttributeError the moment it is exercised.
        self.turn_failed = turn_failed


@pytest.fixture
def kernel(monkeypatch) -> dict[str, Any]:
    """Replace the model kernel and record the prompt it was handed."""
    import app.domains.support.copilot as copilot_module

    calls: list[list[dict[str, str]]] = []
    state: dict[str, Any] = {
        "result": FakeCopilotResult(), "raises": False, "delay": 0.0,
        # Tool calls the fake kernel should make before answering, as
        # ``[(name, arguments), …]``. The real provider loop dispatches tools
        # and feeds the results to the observer; a fake that only returns a
        # string cannot exercise anything downstream of a tool — which is how a
        # test for "a turn that sent a card" ended up asserting on a hand-built
        # dataclass instead of on the turn.
        "dispatch": [],
    }

    async def _fake_turn(
        db, person, history, *,
        extra_tools=None, extra_dispatch=None, system_prompt=None, observer=None,
    ):
        calls.append([dict(t) for t in history])
        state["last_extra_tools"] = extra_tools
        state["last_extra_dispatch"] = extra_dispatch
        state["last_system_prompt"] = system_prompt
        state["last_observer"] = observer
        for name, arguments in state["dispatch"]:
            out = await extra_dispatch(name, arguments, db, person)
            if observer is not None:
                observer(name, out)
        if state["delay"]:
            await asyncio.sleep(state["delay"])
        if state["raises"]:
            raise RuntimeError("model exploded")
        return state["result"]

    monkeypatch.setattr(copilot_module, "run_copilot_turn", _fake_turn)
    return {"calls": calls, "state": state}


@pytest.fixture
def cards(monkeypatch) -> list[dict[str, Any]]:
    """Capture outbound interactive cards instead of calling Feishu."""
    from app.assistant import channel

    captured: list[dict[str, Any]] = []

    async def _fake_send_card(db, *, to, card, idempotency_key=None):
        captured.append({
            "open_id": to.open_id, "chat_id": to.chat_id, "to": to, "card": card,
        })
        return True

    monkeypatch.setattr(channel, "send_card", _fake_send_card)
    return captured


@pytest.fixture
def spawned(monkeypatch) -> list[Any]:
    """Hand back every fire-and-forget task the code under test starts.

    A card tap answers Feishu inside 3 seconds and does the real work after, so
    the work is a task nobody awaits. A test that does not await it either
    asserts on a turn that has not run — green today, green tomorrow, and green
    the day the turn stops being started at all. Collecting the tasks lets the
    test say ``await asyncio.gather(*spawned)`` and then assert on the outcome.
    """
    from app.core import background

    tasks: list[Any] = []
    real_spawn = background.spawn

    def _spawn(coro, *, name=None):
        task = real_spawn(coro, name=name)
        tasks.append(task)
        return task

    monkeypatch.setattr(background, "spawn", _spawn)
    return tasks


@pytest_asyncio.fixture(autouse=True)
async def background_sessions(db_engine):
    """Point ``async_session_factory`` at the test database.

    Code that outlives its request opens its OWN session through the
    module-level factory — the card-action budget (``card_actions``) and the
    message-turn runner (``feishu_events``) both do, and both must. Left
    unpatched, that factory is the one built from ``.env``, so an assistant test
    would silently drive a different database than the one it set up: the
    ``client`` fixture patches it for exactly this reason, and dragging the
    whole ASGI app in just to reach one line of it is a heavy way to get there.

    The background session is a SEPARATE transaction on the same engine, which
    is production's shape too: anything a test wants it to see has to be
    committed, not merely flushed.
    """
    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