"""EidAIClient.call_stream 단위 테스트 — mode 닫힌 매핑·egress 차단·SSE 라인 단위 중계.

★ 실행 환경: httpx + config(settings) 필요 → Docker/staging pytest
   (tests/eid/test_eid_ai_client.py 와 동일 idiom, MacBook 로컬 deps 없으면 hard-fail).
★ httpx 호출은 MockTransport 로 대체 — 실제 네트워크 0 (DB 의존 0).
★ 차단 대상 host 문자열은 런타임 분할 조립 — 차단을 *테스트*하는 코드지 호출 아님
   (meter-guard 오탐 회피, test_eid_ai_client.py 동일).
★ 스트림 검증 = byte-equal 아님: call_stream 이 data: JSON 의 model 을 mode 어휘로
   치환 + usage 제거(머신 경로/텔레메트리 비노출) — content 누적·프레이밍 보존을 본다.
"""

from __future__ import annotations

import asyncio
import json
import sys
from pathlib import Path

import httpx
import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "app"))

import eid.ai as eid_ai  # noqa: E402
from eid.ai import EidAIClient, EidEgressBlocked  # noqa: E402
from services.llm.backends import BackendUnavailable  # noqa: E402
from services.search.llm_gate import _reset_for_test  # noqa: E402

_FIXTURES = Path(__file__).resolve().parents[1] / "fixtures"
_SSE_MACMINI = (_FIXTURES / "router_sse_chat_macmini_26b.txt").read_bytes()
_SSE_QWEN = (_FIXTURES / "router_sse_chat_qwen_27b.txt").read_bytes()

_BLOCKED_HOST = "anthropic" + ".com"

_MSG = [{"role": "user", "content": "안녕"}]


@pytest.fixture(autouse=True)
def _reset_gate():
    """daily(mac-mini-default) 경로가 mlx gate 를 잡으므로 fresh event loop 마다 reset."""
    _reset_for_test()
    yield
    _reset_for_test()


def _patch_transport(monkeypatch, handler):
    """eid.ai 내부 httpx.AsyncClient 생성에 MockTransport 주입 (생성 인자는 보존)."""
    real = httpx.AsyncClient

    def _factory(*args, **kwargs):
        kwargs["transport"] = httpx.MockTransport(handler)
        return real(*args, **kwargs)

    monkeypatch.setattr(eid_ai.httpx, "AsyncClient", _factory)


def _data_objs(raw: bytes) -> list[dict]:
    """SSE bytes → data: JSON 객체 목록 ([DONE] 제외)."""
    objs = []
    for line in raw.split(b"\n"):
        if line.startswith(b"data: ") and line[len(b"data: "):].strip() != b"[DONE]":
            objs.append(json.loads(line[len(b"data: "):]))
    return objs


def _content_concat(raw: bytes) -> str:
    """delta.content 누적 — 본문 무손실 검증용."""
    return "".join(
        (o["choices"][0]["delta"].get("content") or "") for o in _data_objs(raw)
    )


# ── mode 닫힌 매핑 / egress 차단 ──────────────────────────────────────────────


@pytest.mark.asyncio
@pytest.mark.parametrize("bad_mode", ["auto", "claude-cloud", "mac-mini-default", "bogus"])
async def test_unknown_mode_blocked(bad_mode):
    """미지 mode = EidEgressBlocked — alias 직접 지정 포함 닫힌 매핑(daily/deep) 밖 전부 차단."""
    c = EidAIClient()
    try:
        stream = c.call_stream(bad_mode, _MSG, "sys")
        with pytest.raises(EidEgressBlocked):
            await anext(stream)
    finally:
        await c.close()


@pytest.mark.asyncio
async def test_anthropic_router_url_blocked(monkeypatch):
    """router URL 이 외부로 오결선돼도 call_stream 이 차단 (기존 _request 패턴 미러)."""
    monkeypatch.setattr(eid_ai, "_router_url", lambda: "https://api." + _BLOCKED_HOST)
    c = EidAIClient()
    try:
        stream = c.call_stream("deep", _MSG, "sys")
        with pytest.raises(EidEgressBlocked):
            await anext(stream)
    finally:
        await c.close()


# ── alias 매핑 + payload shape + 라인 단위 중계(model 치환·usage 제거) ────────


@pytest.mark.asyncio
async def test_deep_mode_alias_and_sse_line_rewrite(monkeypatch):
    """deep → mac-mini-default alias (맥북 백지화 2026-06-11), system 은 messages[0] 단일 주입, 라인 단위 정화 중계."""
    seen: dict = {}

    def handler(request: httpx.Request) -> httpx.Response:
        seen["url"] = str(request.url)
        seen["json"] = json.loads(request.content)
        return httpx.Response(
            200, content=_SSE_QWEN, headers={"content-type": "text/event-stream"}
        )

    _patch_transport(monkeypatch, handler)
    c = EidAIClient()
    try:
        chunks = [b async for b in c.call_stream("deep", _MSG, "SYS_SENTINEL")]
    finally:
        await c.close()

    joined = b"".join(chunks)
    # (a) content 누적 = fixture 와 동일 (델타 본문 무손실)
    assert _content_concat(joined) == _content_concat(_SSE_QWEN) != ""
    assert len(_data_objs(joined)) == len(_data_objs(_SSE_QWEN))
    # (b) model 필드 = mode 어휘 치환 — 맥북 파일시스템 절대경로/실모델명 비노출
    assert all(o["model"] == "deep" for o in _data_objs(joined))
    assert b"mlx-models" not in joined and b"Qwen" not in joined
    # (c) usage(머신 텔레메트리) 부재
    assert all("usage" not in o for o in _data_objs(joined))
    assert b"peak_memory" not in joined
    # (d) data: [DONE] 보존
    assert b"data: [DONE]" in joined
    # (e) 빈 줄 프레이밍 보존 — 라인 수·빈 줄 위치가 fixture 와 동일
    assert [bool(l) for l in joined.split(b"\n")] == [
        bool(l) for l in _SSE_QWEN.split(b"\n")
    ]
    assert seen["url"].endswith("/v1/chat/completions")
    body = seen["json"]
    assert body["model"] == "mac-mini-default"
    assert body["stream"] is True
    assert body["max_tokens"] == 2048
    assert body["temperature"] == 0.4
    assert body["messages"][0] == {"role": "system", "content": "SYS_SENTINEL"}
    assert body["messages"][1:] == _MSG


@pytest.mark.asyncio
async def test_daily_mode_alias_macmini(monkeypatch):
    """daily → mac-mini-default alias (mlx gate 경유) + 라인 단위 정화 중계."""

    class _TinyChunks(httpx.AsyncByteStream):
        """청크 경계가 라인/JSON 중간에 오도록 7B 씩 방출 — 라인 버퍼링 검증."""

        async def __aiter__(self):
            for i in range(0, len(_SSE_MACMINI), 7):
                yield _SSE_MACMINI[i : i + 7]

        async def aclose(self):
            return None

    def handler(request: httpx.Request) -> httpx.Response:
        assert json.loads(request.content)["model"] == "mac-mini-default"
        return httpx.Response(
            200, stream=_TinyChunks(), headers={"content-type": "text/event-stream"}
        )

    _patch_transport(monkeypatch, handler)
    c = EidAIClient()
    try:
        chunks = [b async for b in c.call_stream("daily", _MSG, "sys")]
    finally:
        await c.close()
    joined = b"".join(chunks)
    # (a) content 누적 동일 / (b) model 치환 / (c) usage 부재 / (d) [DONE] / (e) 프레이밍
    assert _content_concat(joined) == _content_concat(_SSE_MACMINI) != ""
    assert all(o["model"] == "daily" for o in _data_objs(joined))
    assert b"gemma" not in joined
    assert all("usage" not in o for o in _data_objs(joined))
    assert b"data: [DONE]" in joined
    assert [bool(l) for l in joined.split(b"\n")] == [
        bool(l) for l in _SSE_MACMINI.split(b"\n")
    ]


# ── 스트림 시작 전 에러 → BackendUnavailable (ask 어휘 일치) ──────────────────


@pytest.mark.asyncio
async def test_prestream_503_maps_reason(monkeypatch):
    """router 503 body 의 error.type 을 error_reason 으로 추출 (ask 와 동일 어휘)."""

    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(503, json={"error": {"type": "macbook_unavailable"}})

    _patch_transport(monkeypatch, handler)
    c = EidAIClient()
    try:
        stream = c.call_stream("deep", _MSG, "sys")
        with pytest.raises(BackendUnavailable) as ei:
            await anext(stream)
        assert ei.value.reason == "macbook_unavailable"
        assert ei.value.backend_name == "mac-mini-default"
    finally:
        await c.close()


@pytest.mark.asyncio
async def test_prestream_503_no_body_falls_back_router_503(monkeypatch):
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(503, content=b"oops not json")

    _patch_transport(monkeypatch, handler)
    c = EidAIClient()
    try:
        stream = c.call_stream("deep", _MSG, "sys")
        with pytest.raises(BackendUnavailable) as ei:
            await anext(stream)
        assert ei.value.reason == "router_503"
    finally:
        await c.close()


@pytest.mark.asyncio
async def test_prestream_connect_error_maps_router_prefix(monkeypatch):
    """연결 실패 → router_<예외명> (RouterBackend._post 어휘 일치)."""

    def handler(request: httpx.Request) -> httpx.Response:
        raise httpx.ConnectError("connection refused")

    _patch_transport(monkeypatch, handler)
    c = EidAIClient()
    try:
        stream = c.call_stream("deep", _MSG, "sys")
        with pytest.raises(BackendUnavailable) as ei:
            await anext(stream)
        assert ei.value.reason == "router_ConnectError"
    finally:
        await c.close()


@pytest.mark.asyncio
async def test_prestream_400_raises_valueerror_failloud(monkeypatch):
    """router 400 = 닫힌 매핑에서 alias drift 코드 버그 — BackendUnavailable 아닌
    ValueError fail-loud (RouterBackend._post 컨벤션 미러)."""

    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(400, json={"error": "unknown_alias"})

    _patch_transport(monkeypatch, handler)
    c = EidAIClient()
    try:
        stream = c.call_stream("deep", _MSG, "sys")
        with pytest.raises(ValueError, match="router rejected alias='mac-mini-default'"):
            await anext(stream)
    finally:
        await c.close()


# ── wall-clock deadline (게이트 점유 무한화 차단) ─────────────────────────────


@pytest.mark.asyncio
async def test_stream_deadline_exceeded(monkeypatch):
    """업스트림 진입~종료 deadline 초과 → BackendUnavailable(stream_deadline_exceeded)."""

    class _StallStream(httpx.AsyncByteStream):
        """첫 chunk 후 정체 — per-chunk read timeout 으론 안 잡히는 패턴 모사."""

        async def __aiter__(self):
            yield b'data: {"choices": []}\n\n'
            await asyncio.sleep(30)

        async def aclose(self):
            return None

    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(
            200, stream=_StallStream(), headers={"content-type": "text/event-stream"}
        )

    _patch_transport(monkeypatch, handler)
    monkeypatch.setattr(eid_ai, "_STREAM_DEADLINE_S", 0.05)
    c = EidAIClient()
    try:
        stream = c.call_stream("deep", _MSG, "sys")
        with pytest.raises(BackendUnavailable) as ei:
            async for _ in stream:
                pass
        assert ei.value.reason == "stream_deadline_exceeded"
        assert ei.value.backend_name == "mac-mini-default"
    finally:
        await c.close()


# ── error_reason allowlist sanitize ──────────────────────────────────────────


def test_stream_error_reason_sanitized():
    """최종 reason 은 [a-z0-9_]{1,64} allowlist — 불일치(대문자/공백/dict 파편)는
    upstream_502(502)/router_error(그 외) 로 일반화, dict 직렬화 파편 비노출."""
    from eid.ai import _stream_error_reason

    # 정상 어휘는 그대로 (ask 와 동일)
    assert (
        _stream_error_reason(503, b'{"error": {"type": "macbook_unavailable"}}')
        == "macbook_unavailable"
    )
    assert _stream_error_reason(503, b"oops not json") == "router_503"
    assert _stream_error_reason(418, b"{}") == "router_http_418"
    # 502 + 추출 실패 → upstream_502 (기존 upstream_502_{dict...} 파편 제거)
    assert _stream_error_reason(502, b'{"error": {"detail": "x"}}') == "upstream_502"
    # allowlist 밖(대문자/공백/특수문자) → 일반화
    assert _stream_error_reason(502, b'{"error": {"type": "Bad Gateway!"}}') == "upstream_502"
    assert _stream_error_reason(503, b'{"error": {"type": "Weird Reason"}}') == "router_error"
    assert _stream_error_reason(503, b'{"error": {"type": "' + b"a" * 80 + b'"}}') == "router_error"