"""安全 FAQ MCP tools mirror the HTTP self surface exactly.

Two claims: any employee may call them (scope IS the authorisation, same
as ``/api/v1/me/security-faq``), and they carry the same narrowing — an
entry the security team retracted must be unreachable through a PAT too,
and no governance field may ride along.
"""
from __future__ import annotations

import uuid

import pytest

from app.core.exceptions import NotFoundError
from app.domains.auditor import answer_library_service as lib
from app.mcp import context as _mcp_ctx
from app.mcp import tools as _tools

pytestmark = pytest.mark.asyncio


@pytest.fixture()
def _patch_factory(db_engine, monkeypatch):
    from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

    from app.core import database as _db

    monkeypatch.setattr(
        _db, "async_session_factory",
        async_sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False),
    )


def _as_user(user_id):
    return _mcp_ctx.mcp_user_id.set(user_id)


async def _entry(db, *, zh, en, category="general", visible=True):
    e = await lib.create_entry(
        db,
        payload={
            "category": category,
            "question_zh": zh, "question_en": en,
            "answer_zh": "答案内容", "answer_en": "Answer body",
            "employee_visible": visible,
        },
        actor_id=None,
    )
    await db.commit()
    return e


async def test_an_employee_may_search_and_gets_narrowed_rows(
    employee_user, db_session, _patch_factory,
):
    user, _ = employee_user
    e = await _entry(
        db_session, zh="能用个人手机收公司邮件吗?", en="Personal phone for mail?",
        category="endpoint_security",
    )

    token = _as_user(user.id)
    try:
        found = await _tools.search_security_faq(query="个人手机")
        cats = await _tools.list_security_faq_categories()
        detail = await _tools.get_security_faq_entry(entry_id=str(e.id))
    finally:
        _mcp_ctx.mcp_user_id.reset(token)

    assert [row["id"] for row in found["items"]] == [str(e.id)]
    assert found["lang"] == "zh"
    for leaked in ("owner", "usage_count", "is_stale", "source_doc_ids"):
        assert leaked not in found["items"][0]
    assert any(c["key"] == "endpoint_security" for c in cats["categories"])
    assert detail["answer"] == "答案内容"


async def test_english_keywords_find_chinese_entries(
    employee_user, db_session, _patch_factory,
):
    user, _ = employee_user
    await _entry(
        db_session, zh="是否管控 USB 存储?", en="Do you control USB storage?",
    )

    token = _as_user(user.id)
    try:
        found = await _tools.search_security_faq(query="USB storage", lang="zh")
    finally:
        _mcp_ctx.mcp_user_id.reset(token)

    assert found["total"] == 1
    assert found["items"][0]["question"] == "是否管控 USB 存储?"


async def test_a_retracted_entry_is_unreachable_through_a_token(
    employee_user, db_session, _patch_factory,
):
    user, _ = employee_user
    hidden = await _entry(
        db_session, zh="内部拓扑", en="Internal topology", visible=False,
    )

    token = _as_user(user.id)
    try:
        listed = await _tools.search_security_faq()
        with pytest.raises(NotFoundError):
            await _tools.get_security_faq_entry(entry_id=str(hidden.id))
    finally:
        _mcp_ctx.mcp_user_id.reset(token)

    assert str(hidden.id) not in {row["id"] for row in listed["items"]}


async def test_a_malformed_id_is_a_validation_error_not_a_crash(
    employee_user, _patch_factory,
):
    from app.core.exceptions import AppException

    user, _ = employee_user
    token = _as_user(user.id)
    try:
        with pytest.raises(AppException) as exc:
            await _tools.get_security_faq_entry(entry_id="not-a-uuid")
    finally:
        _mcp_ctx.mcp_user_id.reset(token)
    assert exc.value.status_code == 400


async def test_the_limit_is_capped(employee_user, db_session, _patch_factory):
    user, _ = employee_user
    for i in range(3):
        await _entry(db_session, zh=f"问题{i}", en=f"Question {i}")

    token = _as_user(user.id)
    try:
        # A caller asking for the whole library gets a page, not the library.
        found = await _tools.search_security_faq(limit=10_000)
    finally:
        _mcp_ctx.mcp_user_id.reset(token)
    assert found["returned"] <= 50
    assert uuid.UUID(found["items"][0]["id"])


async def test_ask_runs_retrieval_but_neither_composes_nor_records_a_gap(
    employee_user, db_session, _patch_factory, monkeypatch,
):
    """The ask twin exists because an MCP caller cannot reproduce the
    semantic channel from the keyword tool's output — it has no access to
    the corpus vectors. What it does NOT need is a second model writing
    prose it will rewrite, and what it must not do is put a person's name
    against a question in the gap ledger for a probe they did not make.
    """
    from sqlalchemy import select

    from app.domains.auditor import security_faq_service as faq
    from app.domains.auditor.models import AuditorSecurityFaqGap

    user, _ = employee_user
    entry = await _entry(
        db_session, zh="能用生成式 AI 吗?", en="Generative AI?",
    )

    # Counted, not raised: the compose path catches every exception as part
    # of its degradation contract, so a raising double would be swallowed and
    # this test would pass even if the twin did call the model.
    calls: list = []

    async def _client(_db, **_kwargs):
        calls.append(1)
        raise RuntimeError("should not get here")

    monkeypatch.setattr(faq, "build_llm_client", _client)

    token = _as_user(user.id)
    try:
        hit = await _tools.ask_security_faq(question="生成式 AI")
        # No prose was composed for a hit — the caller writes its own.
        composed_on_hit = list(calls)
        miss = await _tools.ask_security_faq(question="完全没有的问题")
    finally:
        _mcp_ctx.mcp_user_id.reset(token)

    assert composed_on_hit == []
    assert [row["id"] for row in hit["items"]] == [str(entry.id)]
    assert hit["answer"] is None
    # The zero-retrieval id-picker is NOT disabled by ``compose=False``: it
    # selects ids and never writes a sentence, and on a deployment with no
    # vectors it is the only thing that can answer at all.
    assert calls == [1]
    assert miss["matched_by"] == "none"
    # Read-only: the tool_runtime session never commits, and the service was
    # told not to write in the first place.
    assert (await db_session.execute(select(AuditorSecurityFaqGap))).first() is None