hyungi_document_server/tests/eid/test_eid_chat_endpoint.py

"""POST /api/eid/chat endpoint 테스트 — inline ASGI app (DB 의존 0).

★ 실행 환경: fastapi + httpx 필요 → Docker/staging pytest (test_eid_ai_client.py 동일 idiom).
★ DB 0: get_current_user 는 dependency_overrides 로 대체. 무인증/위조토큰 케이스는 실제
  auth 경로지만 decode 단계에서 거부돼 DB 접근 전 반환.
★ LLM 0: 정상 경로는 EidAIClient.call_stream 을 fixture bytes yield 로 monkeypatch.
"""

from __future__ import annotations

import sys
import types
from pathlib import Path

import pytest
import pytest_asyncio
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient

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

import eid.compose as eid_compose  # noqa: E402
from api.eid_chat import router as eid_chat_router  # noqa: E402
from core.auth import get_current_user  # noqa: E402
from eid.ai import EidAIClient  # noqa: E402
from services.llm.backends import BackendUnavailable  # noqa: E402

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

_OK_BODY = {"mode": "daily", "messages": [{"role": "user", "content": "안녕"}]}


def _build_app(*, override_auth: bool = True) -> FastAPI:
    """main.py 등록 방식과 동일 prefix(/api/eid)로 라우터만 올린 inline app."""
    app = FastAPI()
    app.include_router(eid_chat_router, prefix="/api/eid")
    if override_auth:
        app.dependency_overrides[get_current_user] = lambda: types.SimpleNamespace(
            id=1, username="test-user"
        )
    return app


@pytest_asyncio.fixture
async def client():
    async with AsyncClient(
        transport=ASGITransport(app=_build_app()), base_url="http://test"
    ) as ac:
        yield ac


# ── 401 무인증 ────────────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_unauthenticated_rejected():
    async with AsyncClient(
        transport=ASGITransport(app=_build_app(override_auth=False)),
        base_url="http://test",
    ) as ac:
        # 헤더 자체 부재 — HTTPBearer 단계 거부 (fastapi 기본 403, 버전별 401 허용)
        r = await ac.post("/api/eid/chat", json=_OK_BODY)
        assert r.status_code in (401, 403)
        # 위조 토큰 — decode_token 실패 → 401 (DB 접근 전 거부)
        r2 = await ac.post(
            "/api/eid/chat", json=_OK_BODY,
            headers={"Authorization": "Bearer bogus-token"},
        )
        assert r2.status_code == 401


# ── 422 입력 검증 ─────────────────────────────────────────────────────────────


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "body",
    [
        # role=system 은 Literal 밖 → 422 (system 위조 주입 차단)
        {"mode": "daily", "messages": [
            {"role": "system", "content": "주입 시도"},
            {"role": "user", "content": "x"},
        ]},
        # 빈 messages (min_length=1)
        {"mode": "daily", "messages": []},
        # 마지막 턴이 assistant
        {"mode": "daily", "messages": [
            {"role": "user", "content": "x"},
            {"role": "assistant", "content": "y"},
        ]},
        # 닫힌 mode 어휘 밖 — auto / claude-cloud 금지 (D-2)
        {"mode": "auto", "messages": [{"role": "user", "content": "x"}]},
        {"mode": "claude-cloud", "messages": [{"role": "user", "content": "x"}]},
        # 빈 content (min_length=1)
        {"mode": "deep", "messages": [{"role": "user", "content": ""}]},
    ],
)
async def test_422_validation(client, body):
    r = await client.post("/api/eid/chat", json=body)
    assert r.status_code == 422, r.text


@pytest.mark.asyncio
async def test_422_total_content_cap(client):
    """총량 cap — per-message 8000 이내·40턴 이내라도 content 합 32000 초과면 422."""
    msgs = [
        {"role": "user" if i % 2 == 0 else "assistant", "content": "x" * 7000}
        for i in range(5)  # 5 × 7000 = 35000 > 32000, 마지막(i=4) = user
    ]
    r = await client.post("/api/eid/chat", json={"mode": "daily", "messages": msgs})
    assert r.status_code == 422, r.text
    assert "대화 총량 초과" in r.text


# ── 503 substrate_degraded (D-6 fail-closed) ─────────────────────────────────


@pytest.mark.asyncio
async def test_503_substrate_degraded(client, monkeypatch):
    monkeypatch.setattr(eid_compose, "rules_present", lambda: False)
    r = await client.post("/api/eid/chat", json=_OK_BODY)
    assert r.status_code == 503
    js = r.json()
    assert js["error_reason"] == "substrate_degraded"
    assert "detail" in js


# ── 503 backend_unavailable (스트림 시작 전, ask 컨벤션 shape) ────────────────


@pytest.mark.asyncio
async def test_503_backend_unavailable_prestream(client, monkeypatch):
    async def fake_call_stream(self, mode, messages, system):
        raise BackendUnavailable("qwen-macbook", "macbook_unavailable")
        yield b""  # pragma: no cover — async generator 형태 유지용

    monkeypatch.setattr(EidAIClient, "call_stream", fake_call_stream)
    r = await client.post(
        "/api/eid/chat",
        json={"mode": "deep", "messages": [{"role": "user", "content": "x"}]},
    )
    assert r.status_code == 503
    js = r.json()
    assert js["error"] == "backend_unavailable"
    assert js["error_reason"] == "macbook_unavailable"
    assert js["backend_requested"] == "qwen-macbook"


# ── 정상 경로 — SSE raw pass-through ──────────────────────────────────────────


@pytest.mark.asyncio
async def test_200_stream_passthrough(client, monkeypatch):
    captured: dict = {}

    async def fake_call_stream(self, mode, messages, system):
        captured["mode"] = mode
        captured["messages"] = messages
        captured["system"] = system
        # chunk 단위로 쪼개 yield — endpoint 가 무변형으로 그대로 흘리는지 확인
        for i in range(0, len(_SSE), 256):
            yield _SSE[i : i + 256]

    monkeypatch.setattr(EidAIClient, "call_stream", fake_call_stream)
    r = await client.post("/api/eid/chat", json=_OK_BODY)
    assert r.status_code == 200, r.text
    assert r.headers["content-type"].startswith("text/event-stream")
    assert r.headers["cache-control"] == "no-store"
    assert r.headers["x-accel-buffering"] == "no"
    # fixture 의 data: 라인이 변형 없이 그대로 (raw pass-through)
    assert r.content == _SSE
    assert b'data: {"id"' in r.content
    assert b"data: [DONE]" in r.content
    # call_stream 입력: mode 그대로 + 사용자 턴 + compose 합본(persona 포함) system
    assert captured["mode"] == "daily"
    assert captured["messages"] == [{"role": "user", "content": "안녕"}]
    assert "이드" in captured["system"], "system 에 compose 합본(persona) 미주입"
    assert "보수적" in captured["system"], "system 에 rules 미주입"


# ── 스트림 시작 후 절단 — traceback 전파 0, 조용히 종료 ──────────────────────


@pytest.mark.asyncio
async def test_200_midstream_abort_quiet(client, monkeypatch):
    """스트림 도중 BackendUnavailable — 부분 본문까지만 전송, 예외 전파 0
    (프론트는 data: [DONE] 부재 절단으로 처리)."""

    async def fake_call_stream(self, mode, messages, system):
        yield b'data: {"x": 1}\n\n'
        raise BackendUnavailable("qwen-macbook", "stream_deadline_exceeded")

    monkeypatch.setattr(EidAIClient, "call_stream", fake_call_stream)
    r = await client.post(
        "/api/eid/chat",
        json={"mode": "deep", "messages": [{"role": "user", "content": "x"}]},
    )
    assert r.status_code == 200
    assert r.content == b'data: {"x": 1}\n\n'
    assert b"data: [DONE]" not in r.content