hyungi_document_server/tests/eid/test_eid_status_endpoint.py

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

★ 실행 환경: fastapi + httpx 필요 → test_eid_chat_endpoint.py 동일 idiom.
★ DB 0 / LLM 0: get_current_user 는 dependency_overrides 로 대체, gate 점유는
  llm_gate.gate_status monkeypatch (eid_chat 이 모듈 attribute 로 호출하므로 유효).
★ 무인증 케이스는 실제 auth 경로지만 decode 단계에서 거부돼 DB 접근 전 반환.
"""

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"))

from api.eid_chat import router as eid_chat_router  # noqa: E402
from core.auth import get_current_user  # noqa: E402
from services.search import llm_gate  # noqa: E402


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.get("/api/eid/status")
        assert r.status_code in (401, 403)
        # 위조 토큰 — decode_token 실패 → 401 (DB 접근 전 거부)
        r2 = await ac.get(
            "/api/eid/status", headers={"Authorization": "Bearer bogus-token"}
        )
        assert r2.status_code == 401


# ── 200 shape ────────────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_200_shape(client, monkeypatch):
    """응답 shape — daily 키 아래 busy/inflight/waiters 3필드, 타입 고정."""
    monkeypatch.setattr(
        llm_gate, "gate_status", lambda: {"inflight": False, "waiters": 0}
    )
    r = await client.get("/api/eid/status")
    assert r.status_code == 200, r.text
    js = r.json()
    assert set(js.keys()) == {"daily"}
    assert set(js["daily"].keys()) == {"busy", "inflight", "waiters"}
    assert isinstance(js["daily"]["busy"], bool)
    assert isinstance(js["daily"]["inflight"], bool)
    assert isinstance(js["daily"]["waiters"], int)


# ── busy 판정 — gate_status monkeypatch ──────────────────────────────────────


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "snap, expected",
    [
        # 유휴 — busy=false (근사: 외부 소비자 점유는 미포착)
        (
            {"inflight": False, "waiters": 0},
            {"busy": False, "inflight": False, "waiters": 0},
        ),
        # inflight 만 — busy=true (확실)
        (
            {"inflight": True, "waiters": 0},
            {"busy": True, "inflight": True, "waiters": 0},
        ),
        # waiters 만 — busy=true (inflight or waiters>0 의 or 분기)
        (
            {"inflight": False, "waiters": 3},
            {"busy": True, "inflight": False, "waiters": 3},
        ),
    ],
)
async def test_busy_from_gate_status(client, monkeypatch, snap, expected):
    monkeypatch.setattr(llm_gate, "gate_status", lambda: dict(snap))
    r = await client.get("/api/eid/status")
    assert r.status_code == 200, r.text
    assert r.json() == {"daily": expected}