서비스 학습
SOURCE · be/tests/test_chat.py · 2026-09-12

추론 서버 없이
채팅 API 전체를 검증한다

29개 테스트가 입력 검증·오류 변환·상한을 붙잡는다. 실제 Gemma를 한 번도 부르지 않고, 0.4초 안에 끝난다.

이 파일을 만든 이유

채팅 API에는 사람이 기억하기 어려운 규칙이 많다. 오류 코드 9개가 각각 다른 status로 나가야 하고, 공백 메시지는 거부해야 하고, 생성 옵션은 client가 못 정해야 한다. 하나라도 조용히 바뀌면 브라우저가 잘못 동작한다.

이 파일은 그 규칙을 코드로 고정한다. 핵심은 추론 서버 없이 전부 검증한다는 점이다. 실제 Gemma는 한 번 생성에 30초가 걸리고 8GB 메모리를 쓴다. 29개를 실제로 돌리면 15분이 넘고 개발 중에는 아무도 실행하지 않게 된다.

전체 코드

import logging
from typing import get_args

import httpx2
import pytest
from fastapi.testclient import TestClient

from app.api.dependencies import get_inference_engine
from app.core.config import Settings
from app.engines.base import (
    GenerationRequest,
    GenerationResult,
    InferenceError,
    InferenceErrorCode,
    TokenUsage,
)
from app.main import create_app
from app.schemas.chat import (
    _MAX_MESSAGE_CHARS as MAX_MESSAGE_CHARS,
    _MAX_MESSAGES as MAX_MESSAGES,
    _MAX_TOTAL_CHARS as MAX_TOTAL_CHARS,
)


class RecordingEngine:
    """Records what the route asked for and returns a fixed answer."""

    def __init__(self) -> None:
        self.received: GenerationRequest | None = None

    async def generate(self, request: GenerationRequest) -> GenerationResult:
        self.received = request
        return GenerationResult(
            id="chatcmpl-test", model="gemma-4-26b-a4b-it",
            text="MoE는 전문가 여러 개 중 일부만 실행하는 구조입니다.",
            finish_reason="stop",
            usage=TokenUsage(
                input_tokens=12, output_tokens=20,
                total_tokens=32, cached_input_tokens=8,
            ),
        )


def make_client(engine: RecordingEngine) -> TestClient:
    application = create_app(Settings(environment="test", log_level="WARNING"))
    application.dependency_overrides[get_inference_engine] = lambda: engine
    return TestClient(application)


def test_chat_returns_the_answer_and_usage() -> None:
    engine = RecordingEngine()
    with make_client(engine) as client:
        response = client.post("/api/v1/chat", json={
            "messages": [{"role": "user", "content": "MoE가 뭐야?"}],
        })

    assert response.status_code == 200
    assert response.json() == {
        "id": "chatcmpl-test",
        "model": "gemma-4-26b-a4b-it",
        "text": "MoE는 전문가 여러 개 중 일부만 실행하는 구조입니다.",
        "finish_reason": "stop",
        "usage": {
            "input_tokens": 12, "output_tokens": 20,
            "total_tokens": 32, "cached_input_tokens": 8,
        },
    }


def test_whole_conversation_reaches_the_engine_in_order() -> None:
    engine = RecordingEngine()
    with make_client(engine) as client:
        client.post("/api/v1/chat", json={"messages": [
            {"role": "system", "content": "짧게 답하라"},
            {"role": "user", "content": "안녕"},
            {"role": "assistant", "content": "반가워요"},
            {"role": "user", "content": "MoE가 뭐야?"},
        ]})

    assert engine.received is not None
    assert [(m.role, m.content) for m in engine.received.messages] == [
        ("system", "짧게 답하라"),
        ("user", "안녕"),
        ("assistant", "반가워요"),
        ("user", "MoE가 뭐야?"),
    ]


def test_server_owns_the_generation_options() -> None:
    engine = RecordingEngine()
    with make_client(engine) as client:
        response = client.post("/api/v1/chat", json={
            "messages": [{"role": "user", "content": "안녕"}],
            "max_output_tokens": 4096,
        })

    assert response.status_code == 422
    assert engine.received is None


def test_generation_options_are_not_client_controlled() -> None:
    engine = RecordingEngine()
    with make_client(engine) as client:
        client.post("/api/v1/chat", json={
            "messages": [{"role": "user", "content": "안녕"}],
        })

    assert engine.received is not None
    assert engine.received.max_output_tokens == 128
    assert engine.received.temperature == 0.2


def test_malformed_requests_are_rejected_before_the_engine() -> None:
    engine = RecordingEngine()
    bad_bodies = (
        {"messages": []},
        {"messages": [{"role": "bot", "content": "안녕"}]},
        {"messages": [{"role": "user", "content": ""}]},
        {"messages": [{"role": "user"}]},
        {"message": "안녕"},
    )
    with make_client(engine) as client:
        for body in bad_bodies:
            assert client.post("/api/v1/chat", json=body).status_code == 422, body

    assert engine.received is None


def test_openapi_documents_the_chat_contract() -> None:
    with make_client(RecordingEngine()) as client:
        document = client.get("/openapi.json").json()

    assert "/api/v1/chat" in document["paths"]
    schema = document["components"]["schemas"]["ChatResponse"]
    assert set(schema["required"]) == {"id", "model", "text", "finish_reason", "usage"}


class FailingEngine:
    """Raises one chosen InferenceError instead of generating."""

    def __init__(self, code: str, upstream_status: int | None = None) -> None:
        self.code = code
        self.upstream_status = upstream_status

    async def generate(self, request: GenerationRequest) -> GenerationResult:
        raise InferenceError(self.code, upstream_status=self.upstream_status)


# The contract agreed for step 3, written out independently of the route module.
EXPECTED_STATUS = {
    "invalid_message": 400,
    "request_rejected": 400,
    "context_too_long": 413,
    "busy": 429,
    "timeout": 504,
    "unavailable": 503,
    "model_mismatch": 503,
    "upstream_error": 502,
    "invalid_response": 502,
}


def post_one(engine: object) -> httpx2.Response:
    application = create_app(Settings(environment="test", log_level="WARNING"))
    application.dependency_overrides[get_inference_engine] = lambda: engine
    with TestClient(application) as client:
        return client.post("/api/v1/chat", json={
            "messages": [{"role": "user", "content": "MoE가 뭐야?"}],
        })


@pytest.mark.parametrize(("code", "expected"), sorted(EXPECTED_STATUS.items()))
def test_each_failure_maps_to_its_own_status(code: str, expected: int) -> None:
    response = post_one(FailingEngine(code))

    assert response.status_code == expected
    assert response.json() == {"error_code": code}


def test_every_error_code_has_a_decided_status() -> None:
    assert set(get_args(InferenceErrorCode)) == set(EXPECTED_STATUS)


def test_busy_tells_the_client_to_wait() -> None:
    response = post_one(FailingEngine("busy", upstream_status=429))

    assert response.status_code == 429
    assert int(response.headers["retry-after"]) > 0


def test_upstream_status_never_reaches_the_browser() -> None:
    response = post_one(FailingEngine("upstream_error", upstream_status=500))

    assert response.status_code == 502
    assert response.json() == {"error_code": "upstream_error"}
    assert "500" not in response.text


def test_failures_are_logged_without_the_prompt(caplog) -> None:
    with caplog.at_level(logging.WARNING):
        post_one(FailingEngine("timeout"))

    logged = "\n".join(record.getMessage() for record in caplog.records)
    assert "code=timeout" in logged
    assert "MoE가 뭐야?" not in logged


def post_messages(engine: object, messages: list) -> httpx2.Response:
    application = create_app(Settings(environment="test", log_level="WARNING"))
    application.dependency_overrides[get_inference_engine] = lambda: engine
    with TestClient(application) as client:
        return client.post("/api/v1/chat", json={"messages": messages})


@pytest.mark.parametrize(("label", "messages"), [
    ("공백만", [{"role": "user", "content": "   "}]),
    ("탭과 줄바꿈만", [{"role": "user", "content": "\t\n "}]),
    ("한 메시지가 상한 초과", [{"role": "user", "content": "x" * (MAX_MESSAGE_CHARS + 1)}]),
    ("메시지 개수 초과", [{"role": "user", "content": "x"}] * (MAX_MESSAGES + 1)),
    ("대화 전체가 상한 초과",
     [{"role": "user", "content": "x" * MAX_MESSAGE_CHARS}] * (MAX_TOTAL_CHARS // MAX_MESSAGE_CHARS + 1)),
])
def test_oversized_input_is_rejected_before_the_engine(label: str, messages: list) -> None:
    engine = RecordingEngine()

    assert post_messages(engine, messages).status_code == 422, label
    assert engine.received is None, label


def test_surrounding_whitespace_is_trimmed_before_the_engine() -> None:
    engine = RecordingEngine()
    post_messages(engine, [{"role": "user", "content": "  MoE가 뭐야?\n\n"}])

    assert engine.received is not None
    assert engine.received.messages[0].content == "MoE가 뭐야?"


@pytest.mark.parametrize(("label", "messages"), [
    ("메시지 개수 상한", [{"role": "user", "content": "x"}] * MAX_MESSAGES),
    ("한 메시지 길이 상한", [{"role": "user", "content": "x" * MAX_MESSAGE_CHARS}]),
    ("대화 전체 길이 상한",
     [{"role": "user", "content": "x" * MAX_MESSAGE_CHARS}] * (MAX_TOTAL_CHARS // MAX_MESSAGE_CHARS)),
])
def test_input_exactly_at_the_limit_is_accepted(label: str, messages: list) -> None:
    engine = RecordingEngine()

    assert post_messages(engine, messages).status_code == 200, label
    assert engine.received is not None, label


def test_the_real_context_limit_still_comes_from_the_server() -> None:
    """Our caps are a coarse guard; the model decides what actually fits."""
    response = post_messages(FailingEngine("context_too_long"), [
        {"role": "user", "content": "짧지만 문맥을 넘긴 대화"},
    ])

    assert response.status_code == 413
    assert response.json() == {"error_code": "context_too_long"}

가짜 엔진 두 개

class RecordingEngine:
    """Records what the route asked for and returns a fixed answer."""

    def __init__(self) -> None:
        self.received: GenerationRequest | None = None

    async def generate(self, request: GenerationRequest) -> GenerationResult:
        self.received = request
        return GenerationResult(...)


class FailingEngine:
    """Raises one chosen InferenceError instead of generating."""

    def __init__(self, code: str, upstream_status: int | None = None) -> None:
        ...
    async def generate(self, request: GenerationRequest) -> GenerationResult:
        raise InferenceError(self.code, upstream_status=self.upstream_status)
InferenceEngine을 상속하지 않는다

base.py의 계약이 Protocol이라 상속 선언이 필요 없다. 두 class 모두 generate 하나만 구현했고 check_ready·list_models는 없다. 채팅 route가 그것만 부르기 때문이다.

RecordingEngine.received가 하는 일

route가 엔진에게 무엇을 넘겼는지 붙잡아 둔다. 응답만 보면 “대화가 순서대로 전달됐는지”, “생성 옵션이 128/0.2로 붙었는지”를 확인할 수 없다. 거부되어야 할 요청에서 received is None인지 보는 것도 같은 장치다 — 엔진에 닿기 전에 막혔다는 증거가 된다.

FailingEngine이 코드를 인수로 받는 이유

9개 오류 코드마다 class를 만들 필요가 없다. 하나로 모든 실패를 흉내 낸다.

의존성 교체 — 추론 서버 자리에 대역을 넣는다

def post_one(engine: object) -> httpx2.Response:
    application = create_app(Settings(environment="test", log_level="WARNING"))
    application.dependency_overrides[get_inference_engine] = lambda: engine
    with TestClient(application) as client:
        return client.post("/api/v1/chat", json={
            "messages": [{"role": "user", "content": "MoE가 뭐야?"}],
        })

dependencies.py의 함수를 열쇠로 써서 대역을 등록한다. route 코드는 전혀 몰라도 되고 app.state를 직접 조작하지도 않는다. test_health.py가 /readyz에 쓰던 방식과 같다.

create_app(Settings(...))로 테스트마다 새 application을 만든다. 전역 app에 override를 걸면 다른 테스트에 남는다.

오류 변환 — 계약을 두 번 적는다

# The contract agreed for step 3, written out independently of the route module.
EXPECTED_STATUS = {
    "invalid_message": 400,   "request_rejected": 400,
    "context_too_long": 413,  "busy": 429,
    "timeout": 504,           "unavailable": 503,
    "model_mismatch": 503,    "upstream_error": 502,
    "invalid_response": 502,
}


@pytest.mark.parametrize(("code", "expected"), sorted(EXPECTED_STATUS.items()))
def test_each_failure_maps_to_its_own_status(code: str, expected: int) -> None:
    response = post_one(FailingEngine(code))

    assert response.status_code == expected
    assert response.json() == {"error_code": code}

routes/chat.py의 _ERROR_STATUS를 import하지 않고 표를 여기에 다시 적었다. 구현을 가져다 쓰면 “구현이 구현과 같다”는 무의미한 검사가 된다. 따로 적어야 합의한 계약을 검사하는 것이 된다.

parametrize가 9개 테스트로 갈라진다

함수는 하나인데 pytest가 코드마다 별도 테스트로 실행한다. 하나가 깨져도 나머지는 계속 돌고, 실패 목록에 어느 코드인지 이름으로 나온다.

새 오류 코드를 감시하는 한 줄

이 파일에서 가장 중요한 테스트다.

def test_every_error_code_has_a_decided_status() -> None:
    assert set(get_args(InferenceErrorCode)) == set(EXPECTED_STATUS)

get_args가 base.py의 Literal에서 값 9개를 꺼낸다. 10번째 코드를 추가하면 이 테스트가 즉시 실패한다. 누군가 status를 정하기 전까지는 통과하지 못한다. 4단계에서 streaming 관련 코드가 늘어날 가능성이 높아 미리 걸어 두었다.

정보가 새지 않는지도 검사한다

upstream_status가 본문에 없는지, 로그에 prompt가 없는지 확인한다. base.py의 “코드만 담는다”는 규칙이 route 계층까지 지켜지는지 보는 것이다.

assert "500" not in response.text                 # upstream_status 는 응답에 없다
assert "MoE가 뭐야?" not in logged                  # 로그에 prompt 가 없다
assert int(response.headers["retry-after"]) > 0   # 429 에는 대기 힌트가 붙는다

입력 상한 — 경계값까지 확인한다

@pytest.mark.parametrize(("label", "messages"), [
    ("공백만", [{"role": "user", "content": "   "}]),
    ("탭과 줄바꿈만", [{"role": "user", "content": "\t\n "}]),
    ("한 메시지가 상한 초과", [{"role": "user", "content": "x" * (MAX_MESSAGE_CHARS + 1)}]),
    ("메시지 개수 초과", [{"role": "user", "content": "x"}] * (MAX_MESSAGES + 1)),
    ...
])
def test_oversized_input_is_rejected_before_the_engine(label, messages) -> None:
    engine = RecordingEngine()

    assert post_messages(engine, messages).status_code == 422, label
    assert engine.received is None, label
상한값을 숫자로 적지 않고 import한다

MAX_MESSAGE_CHARS + 1처럼 schemas/chat.py의 상수를 가져와 계산한다. 상한을 바꿔도 테스트가 따라가므로, “상한이 있다”는 성질을 검사하지 특정 숫자를 검사하지 않는다.

경계값 세 개를 따로 본다

정확히 상한인 입력은 통과해야 한다. 메시지 100개, 16,000자, 총 64,000자가 각각 200을 받는지 확인한다. 부등호를 >=로 잘못 쓰는 off-by-one을 잡는다.

engine.received is None이 핵심

422가 나왔다는 것만으로는 부족하다. 엔진이 호출되지 않았음까지 확인해야 “8GB 환경에서 30초를 낭비하지 않는다”가 증명된다.

역할 분담을 고정하는 테스트

마지막 테스트가 두 층의 경계를 못 박는다.

def test_the_real_context_limit_still_comes_from_the_server() -> None:
    """Our caps are a coarse guard; the model decides what actually fits."""
    response = post_messages(FailingEngine("context_too_long"), [
        {"role": "user", "content": "짧지만 문맥을 넘긴 대화"},
    ])

    assert response.status_code == 413

메시지가 짧아 우리 상한에는 걸리지 않는데도 엔진이 context_too_long을 던지면 413이 나간다. 진짜 문맥 판정은 서버가 한다는 설계가 코드로 남는다.

실제 실행 결과

cd be
.venv/bin/python -m pytest tests/test_chat.py -q

.............................                                            [100%]
29 passed in 0.40s
이 파일29 passed0.4초
전체 backend76 passed2026-09-12 기준
추론 서버불필요가짜 엔진으로 대체

TurboFieldfareServer가 꺼져 있어도 29개 모두 통과한다. 반대로 이 테스트들은 실제 Gemma가 답을 만드는지 증명하지 않는다. 그 확인은 smoke.py와 실제 curl 요청의 몫이다.

이 파일이 증명하지 않는 것

항목왜 못 하나어디서 확인하나
실제 Gemma 답변 품질가짜 엔진은 고정 문자열을 돌려준다실제 서버로 curl
busy(429)가 실제로 발생동시 요청 5개 이상이 필요하고 8GB에서 위험미검증
context_too_long이 실제로 발생16,384 토큰을 넘기는 대화가 필요미검증
prompt cache 재사용서버 내부 동작실제 2번째 턴 요청 — 99/124 토큰 확인됨
streaming 중간 실패계약 자체가 아직 없다4단계

자주 발생하는 오류

증상원인진단
DID NOT RAISE 대신 200상한 검사가 schema에서 빠짐schemas/chat.py의 validator를 확인한다
누락 감시 테스트 실패base.py에 코드를 추가하고 매핑을 안 정함의도된 실패다. status를 정하면 통과한다
AttributeError: inference_engineoverride 없이 with만 씀대역을 등록하거나 실제 lifespan을 쓴다
다른 테스트에 영향전역 app에 override 등록create_app으로 매번 새로 만든다
received가 이전 값RecordingEngine을 재사용테스트마다 새로 만든다

이전 단계와 다음 파일

여기서 검증하는 구현은 routes/chat.py, 입력 계약은 schemas/chat.py, 오류 코드의 출처는 engines/base.py다. 상태 endpoint를 같은 방식으로 검증하는 파일은 test_health.py다.

← 개요Backend 검증 목록다음 →상태 endpoint 검증