"""gRPC 服务形态集成测试：双形态 golden 一致（ADR-017）。

在进程内启动服务器（随机端口），验证：
- GetEngineInfo 版本协商
- TransformMarkdown 与进程内 lingtai.transform_markdown 输出一致
- OpenStream 双向流 envelope 键序与序号单调
- EvaluatePolicy / ValidateUrl / CheckStateTransition / RerankRecall 行为
"""

import sys
from pathlib import Path

import grpc
import pytest

_SERVER_ROOT = Path(__file__).resolve().parents[2] / "engine" / "server"
sys.path.insert(0, str(_SERVER_ROOT / "generated"))
_SERVER_MODULE = _SERVER_ROOT / "lingtai_server.py"

import liushen.v1.liushen_pb2 as pb  # noqa: E402
import liushen.v1.liushen_pb2_grpc as pb_grpc  # noqa: E402

from app.core import lingtai  # noqa: E402


@pytest.fixture()
async def grpc_channel():
    import importlib.util as _ilu

    from grpc import aio as grpc_aio

    _spec = _ilu.spec_from_file_location(
        "lingtai_server",
        _SERVER_MODULE,
    )
    _mod = _ilu.module_from_spec(_spec)
    _spec.loader.exec_module(_mod)
    LingtaiEngineServicer = _mod.LingtaiEngineServicer

    server = grpc_aio.server()
    pb_grpc.add_LingtaiEngineServicer_to_server(LingtaiEngineServicer(), server)
    port = server.add_insecure_port("127.0.0.1:0")
    await server.start()
    channel = grpc_aio.insecure_channel(f"127.0.0.1:{port}")
    yield channel
    await channel.close()
    await server.stop(0)


async def test_engine_info(grpc_channel) -> None:
    stub = pb_grpc.LingtaiEngineStub(grpc_channel)
    info = await stub.GetEngineInfo(pb.GetEngineInfoRequest())
    assert info.abi_major == 1
    assert info.parser_version == "hunjing-markdown-v1"
    assert "transform" in info.capabilities


async def test_transform_dual_form_golden(grpc_channel) -> None:
    """双形态一致：服务输出 == 进程内引擎输出（ADR-017 核心不变量）。"""
    import json

    stub = pb_grpc.LingtaiEngineStub(grpc_channel)
    markdown = "说明\n```mermaid\nA-->B\n```\n$$\nz\n$$\n"
    resp = await stub.TransformMarkdown(pb.TransformMarkdownRequest(markdown=markdown))
    served = json.loads(resp.canonical_json)
    inproc = lingtai.transform_markdown(markdown)
    assert served == inproc  # 服务形态与进程内输出一致


async def test_open_stream_envelope(grpc_channel) -> None:
    import json

    stub = pb_grpc.LingtaiEngineStub(grpc_channel)

    async def frames():
        yield pb.StreamFrame(block_started="block_01")
        yield pb.StreamFrame(text_delta="你")
        yield pb.StreamFrame(text_delta="好")

    events = []
    async for ev in stub.OpenStream(frames()):
        events.append(json.loads(ev.event_json))
    assert len(events) == 2
    assert all(e["type"] == "content.block.delta" for e in events)
    seqs = [e["sequence"] for e in events]
    assert seqs == sorted(seqs)
    assert events[-1]["data"]["delta"]["text"] == "好"


async def test_policy_and_state(grpc_channel) -> None:
    stub = pb_grpc.LingtaiEngineStub(grpc_channel)
    # 决策表
    assert (await stub.EvaluatePolicy(pb.EvaluatePolicyRequest(
        autonomy_level=1, risk_level=0))).decision == 0
    assert (await stub.EvaluatePolicy(pb.EvaluatePolicyRequest(
        autonomy_level=0, risk_level=2))).decision == 2
    # URL 白名单
    assert (await stub.ValidateUrl(pb.ValidateUrlRequest(
        url="https://ok"))).allowed is True
    assert (await stub.ValidateUrl(pb.ValidateUrlRequest(
        url="file:///etc"))).allowed is False
    # 状态机
    assert (await stub.CheckStateTransition(pb.CheckStateTransitionRequest(
        **{"from": pb.RUN_STATUS_QUEUED}, to=pb.RUN_STATUS_RUNNING))).valid is True
    assert (await stub.CheckStateTransition(pb.CheckStateTransitionRequest(
        **{"from": pb.RUN_STATUS_COMPLETED}, to=pb.RUN_STATUS_RUNNING))).valid is False


async def test_rerank(grpc_channel) -> None:
    import json

    stub = pb_grpc.LingtaiEngineStub(grpc_channel)
    resp = await stub.RerankRecall(pb.RerankRecallRequest(
        candidates_json='{"query_terms":["魂镜"],"candidates":['
        '{"id":"a","text":"魂镜 系统"},{"id":"b","text":"无"}]}'
    ))
    ranked = json.loads(resp.ranked_json)["ranked"]
    assert ranked[0] == {"id": "a", "score": 10}
    assert ranked[1] == {"id": "b", "score": 0}


async def test_rebuild_export_unimplemented(grpc_channel) -> None:
    stub = pb_grpc.LingtaiEngineStub(grpc_channel)
    with pytest.raises(grpc.aio.AioRpcError) as exc:
        await stub.RebuildExport(pb.RebuildExportRequest(canonical_json="{}"))
    assert exc.value.code() == grpc.StatusCode.UNIMPLEMENTED