응답과 실패를 올바른 내부 계약으로 바꾸는가?

2. 전체 구조에서 담당하는 역할

내부 GenerationRequest → 전송 JSON → upstream 응답 → GenerationResult 또는 InferenceError의 번역 규칙을 검증한다. helper는 생성용과 probe용 client를 별도로 만든다. 생성 client에는 POST /v1/chat/completions만, probe client에는 두 GET 경로만 허용하므로 잘못된 client 선택도 잡는다.

3. 입력과 출력의 핵심 대응

4. 코드 구간별 해설

5. 데이터 상태 변화

MockTransport 테스트에서는 연결이 실제로 점유되지는 않는다. ReadTimeout을 직접 일으키는 기존 테스트도 오류 변환만 확인한다. 실제 timer·pool·취소는 다른 테스트에서 별도로 확인한다.

7. 자주 생기는 오류와 한계

새 upstream code를 추가하면서 매핑 표만 바꾸고 내부 Literal을 빠뜨리면 계약과 문서가 달라진다. base.py와 Adapter 및 테스트를 함께 갱신해야 한다. 알 수 없는 오류 문구를 그대로 외부로 반환해 디버깅하려 하면 prompt가 노출될 수 있다. 알려진 범주·status·요청 식별자처럼 안전한 정보로 관찰하는 정책을 유지한다.

현재 전체 코드

import asyncio
import json
import traceback
from collections.abc import AsyncGenerator, Callable
from contextlib import asynccontextmanager

import httpx2
import pytest

from app.engines.base import GenerationRequest, InferenceError, Message
from app.engines.turbofieldfare import TurboFieldfareAdapter


MODEL = "gemma-4-26b-a4b-it"


@asynccontextmanager
async def make_adapter(
    handler: Callable[[httpx2.Request], httpx2.Response],
) -> AsyncGenerator[TurboFieldfareAdapter, None]:
    def generate_only(request: httpx2.Request) -> httpx2.Response:
        assert (request.method, request.url.path) == ("POST", "/v1/chat/completions")
        return handler(request)

    def probe_only(request: httpx2.Request) -> httpx2.Response:
        assert request.method == "GET"
        assert request.url.path in {"/health", "/v1/models"}
        return handler(request)

    async with (
        httpx2.AsyncClient(
            base_url="http://inference.test",
            transport=httpx2.MockTransport(generate_only), trust_env=False,
        ) as generation_client,
        httpx2.AsyncClient(
            base_url="http://inference.test",
            transport=httpx2.MockTransport(probe_only), trust_env=False,
        ) as probe_client,
    ):
        yield TurboFieldfareAdapter(
            generation_client, probe_client=probe_client,
            model=MODEL, probe_timeout=1, generation_timeout=1,
        )


def test_check_ready_checks_health_and_configured_model() -> None:
    paths: list[str] = []

    def handler(request: httpx2.Request) -> httpx2.Response:
        paths.append(request.url.path)
        if request.url.path == "/health":
            return httpx2.Response(200, json={"status": "ok"})
        return httpx2.Response(200, json={
            "object": "list",
            "data": [{"id": MODEL, "object": "model", "created": 0,
                      "owned_by": "turbofieldfare"}],
        })

    async def scenario() -> None:
        async with make_adapter(handler) as adapter:
            model = await adapter.check_ready()
        assert model.id == MODEL

    asyncio.run(scenario())
    assert paths == ["/health", "/v1/models"]


def test_generate_translates_request_and_normalizes_response() -> None:
    observed_payload: dict = {}

    def handler(request: httpx2.Request) -> httpx2.Response:
        observed_payload.update(json.loads(request.content))
        return httpx2.Response(200, json={
            "id": "chatcmpl-test",
            "object": "chat.completion",
            "created": 0,
            "model": MODEL,
            "choices": [{
                "index": 0,
                "message": {"role": "assistant", "content": "READY"},
                "finish_reason": "stop",
            }],
            "usage": {
                "prompt_tokens": 12,
                "completion_tokens": 1,
                "total_tokens": 13,
                "prompt_tokens_details": {"cached_tokens": 5},
            },
        })

    async def scenario() -> None:
        async with make_adapter(handler) as adapter:
            result = await adapter.generate(GenerationRequest(
                messages=(Message(role="user", content="Say READY"),),
                max_output_tokens=16,
                temperature=0,
            ))
        assert result.text == "READY"
        assert result.usage.input_tokens == 12
        assert result.usage.output_tokens == 1
        assert result.usage.cached_input_tokens == 5

    asyncio.run(scenario())
    assert observed_payload == {
        "model": MODEL,
        "messages": [{"role": "user", "content": "Say READY"}],
        "max_completion_tokens": 16,
        "temperature": 0.0,
        "stream": False,
    }


@pytest.mark.parametrize(
    ("status_code", "expected_code"),
    [(400, "request_rejected"), (429, "busy"), (500, "upstream_error")],
)
def test_status_codes_become_stable_errors(
    status_code: int, expected_code: str,
) -> None:
    def handler(_: httpx2.Request) -> httpx2.Response:
        return httpx2.Response(status_code, json={
            "error": {"message": "upstream detail", "code": "some_code"},
        })

    async def scenario() -> None:
        with pytest.raises(InferenceError) as caught:
            async with make_adapter(handler) as adapter:
                await adapter.list_models()
        assert caught.value.code == expected_code
        assert caught.value.upstream_status == status_code
        assert "upstream detail" not in str(caught.value)

    asyncio.run(scenario())


def test_connection_failure_becomes_unavailable() -> None:
    def handler(request: httpx2.Request) -> httpx2.Response:
        raise httpx2.ConnectError("connection refused", request=request)

    async def scenario() -> None:
        with pytest.raises(InferenceError) as caught:
            async with make_adapter(handler) as adapter:
                await adapter.list_models()
        assert caught.value.code == "unavailable"

    asyncio.run(scenario())


def test_http_timeout_becomes_timeout() -> None:
    def handler(request: httpx2.Request) -> httpx2.Response:
        raise httpx2.ReadTimeout("too slow", request=request)

    async def scenario() -> None:
        with pytest.raises(InferenceError) as caught:
            async with make_adapter(handler) as adapter:
                await adapter.list_models()
        assert caught.value.code == "timeout"

    asyncio.run(scenario())


@pytest.mark.parametrize(
    ("status_code", "wire_code", "expected_code"),
    [
        (400, "context_length_exceeded", "context_too_long"),
        (400, "invalid_message", "invalid_message"),
        (404, "model_not_found", "model_mismatch"),
        (429, "queue_full", "busy"),
        (400, "future_error", "request_rejected"),
        (404, "not_found", "request_rejected"),
        (401, "model_not_found", "request_rejected"),
        (400, "model_not_found", "request_rejected"),
        (500, "context_length_exceeded", "upstream_error"),
        (429, "invalid_message", "busy"),
    ],
)
def test_known_error_codes_are_normalized_without_leaking_details(
    status_code: int, wire_code: str, expected_code: str, caplog,
) -> None:
    secret = "PRIVATE_PROMPT_AND_UPSTREAM_DETAIL"
    caplog.set_level("INFO")

    def handler(_: httpx2.Request) -> httpx2.Response:
        return httpx2.Response(status_code, json={
            "error": {"type": "invalid_request_error", "code": wire_code,
                      "message": secret, "param": secret},
        })

    async def scenario() -> None:
        async with make_adapter(handler) as adapter:
            with pytest.raises(InferenceError) as caught:
                await adapter.generate(GenerationRequest(
                    messages=(Message(role="user", content=secret),),
                ))
        error = caught.value
        assert error.code == expected_code
        assert error.upstream_status == status_code
        assert vars(error) == {"code": expected_code, "upstream_status": status_code}
        assert secret not in str(error)
        assert secret not in "".join(traceback.format_exception(error))

    asyncio.run(scenario())
    assert secret not in caplog.text


@pytest.mark.parametrize("body", [
    b"not JSON: PRIVATE_DETAIL", b"null", b"[]", b"{}",
    b'{"error":null}', b'{"error":[]}', b'{"error":{}}',
    b'{"error":{"code":123}}', b'{"error":{"code":[]}}',
    b'{"error":{"code":"PRIVATE_DETAIL","message":"PRIVATE_DETAIL"}}',
])
def test_malformed_or_unknown_error_body_uses_safe_status_fallback(body: bytes) -> None:
    def handler(_: httpx2.Request) -> httpx2.Response:
        return httpx2.Response(400, content=body)

    async def scenario() -> None:
        async with make_adapter(handler) as adapter:
            with pytest.raises(InferenceError) as caught:
                await adapter.list_models()
        assert caught.value.code == "request_rejected"
        assert caught.value.upstream_status == 400
        assert "PRIVATE_DETAIL" not in "".join(traceback.format_exception(caught.value))

    asyncio.run(scenario())


def test_missing_configured_model_becomes_model_mismatch() -> None:
    def handler(request: httpx2.Request) -> httpx2.Response:
        if request.url.path == "/health":
            return httpx2.Response(200, json={"status": "ok"})
        return httpx2.Response(200, json={
            "object": "list",
            "data": [{"id": "some-other-model", "object": "model"}],
        })

    async def scenario() -> None:
        with pytest.raises(InferenceError) as caught:
            async with make_adapter(handler) as adapter:
                await adapter.check_ready()
        assert caught.value.code == "model_mismatch"

    asyncio.run(scenario())


def test_inconsistent_token_counts_are_rejected() -> None:
    def handler(_: httpx2.Request) -> httpx2.Response:
        return httpx2.Response(200, json={
            "id": "chatcmpl-invalid",
            "object": "chat.completion",
            "model": MODEL,
            "choices": [{
                "index": 0,
                "message": {"role": "assistant", "content": "secret text"},
                "finish_reason": "stop",
            }],
            "usage": {
                "prompt_tokens": 3,
                "completion_tokens": 2,
                "total_tokens": 999,
                "prompt_tokens_details": {"cached_tokens": 0},
            },
        })

    async def scenario() -> None:
        with pytest.raises(InferenceError) as caught:
            async with make_adapter(handler) as adapter:
                await adapter.generate(GenerationRequest(
                    messages=(Message(role="user", content="hello"),),
                ))
        assert caught.value.code == "invalid_response"
        assert "secret text" not in str(caught.value)

    asyncio.run(scenario())

실제 테스트 소스와 동일한 코드다. 실행 명령: cd be 후 .venv/bin/python -m pytest -q tests/test_turbofieldfare_adapter.py.

8. 다음 파일과의 연결

Adapter 소스로 돌아가기에서 다른 경계의 보장을 확인한다. 전체 구조는 Backend 개요, client 조립은 main.py, 오류 번역은 Adapter 구현과 연결된다.