생성 연결이 꽉 차도 상태를 확인할 수 있는가?

2. 전체 구조에서 검증하는 경계

가짜 서버는 Gemma의 계산 대신 미리 정한 JSON을 반환한다. 여기서 필요한 것은 토큰 품질이 아니라 “응답을 기다리는 연결이 실제로 점유되는가”이기 때문이다.

3. 입력과 예상 출력

두 경우 모두 /healthz는 200이어야 하며, 생성 응답을 해제하면 네 요청이 모두 OK로 완료되어야 한다. 모델 ID가 틀린 경우도 검사해야 “연결 분리 대신 readiness를 무조건 200으로 바꾸는 잘못된 수정”을 잡을 수 있다.

4. 중요한 코드를 구간별로 읽기

5. 화면 대신 작업 상태가 바뀌는 순서

이 파일은 UI 코드를 바꾸지 않는다. 관찰하는 state는 생성 Task의 완료 여부, 서버의 카운터, Event와 기록된 요청 경로다.

7. 자주 생길 수 있는 오류

샌드박스에서 PermissionError가 나면 127.0.0.1 임시 포트 bind 권한을 확인한다. all_occupied 대기 초과는 네 요청이 서버까지 도착하지 못했다는 뜻이므로 연결 수 설정이나 앞선 Task 오류를 확인한다. HTTP 503과 busy가 재발하면 생성·probe client가 같은 풀을 공유하도록 바뀌지 않았는지 확인한다.

현재 전체 코드

"""Exercise real connection pools against a model-free loopback HTTP server."""

import asyncio
import json
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager

import httpx2
import pytest

from app.core.config import Settings
from app.engines.base import GenerationRequest, Message
from app.main import create_app


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


class HeldGenerationServer:
    """Keep four POST responses pending while still answering GET probes."""

    def __init__(self, advertised_model: str) -> None:
        self.advertised_model = advertised_model
        self.release = asyncio.Event()
        self.all_occupied = asyncio.Event()
        self.generation_count = 0
        self.probe_paths: list[str] = []
        self.handlers: set[asyncio.Task] = set()

    async def handle(
        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter,
    ) -> None:
        task = asyncio.current_task()
        assert task is not None
        self.handlers.add(task)
        try:
            while True:
                head = await reader.readuntil(b"\r\n\r\n")
                first, *headers = head.decode("ascii").split("\r\n")
                method, path, _ = first.split()
                size = next((
                    int(header.split(":", 1)[1])
                    for header in headers
                    if header.lower().startswith("content-length:")
                ), 0)
                await reader.readexactly(size)
                if method == "POST" and path == "/v1/chat/completions":
                    self.generation_count += 1
                    if self.generation_count == 4:
                        self.all_occupied.set()
                    await self.release.wait()
                    body = {
                        "id": "chatcmpl-local-test", "object": "chat.completion",
                        "model": MODEL,
                        "choices": [{
                            "index": 0,
                            "message": {"role": "assistant", "content": "OK"},
                            "finish_reason": "stop",
                        }],
                        "usage": {
                            "prompt_tokens": 1, "completion_tokens": 1,
                            "total_tokens": 2,
                            "prompt_tokens_details": {"cached_tokens": 0},
                        },
                    }
                else:
                    self.probe_paths.append(path)
                    body = {"status": "ok"} if path == "/health" else {
                        "object": "list",
                        "data": [{"id": self.advertised_model, "object": "model"}],
                    }
                payload = json.dumps(body).encode()
                writer.write(
                    b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n"
                    + f"Content-Length: {len(payload)}\r\n\r\n".encode()
                    + payload
                )
                await writer.drain()
        except (asyncio.IncompleteReadError, ConnectionError):
            pass
        finally:
            writer.close()
            await writer.wait_closed()
            self.handlers.discard(task)

    @asynccontextmanager
    async def running(self) -> AsyncGenerator[str, None]:
        server = await asyncio.start_server(self.handle, "127.0.0.1", 0)
        async with server:
            try:
                port = server.sockets[0].getsockname()[1]
                yield f"http://127.0.0.1:{port}"
            finally:
                self.release.set()
                tasks = tuple(self.handlers)
                for task in tasks:
                    task.cancel()
                await asyncio.gather(*tasks, return_exceptions=True)


@pytest.mark.parametrize(
    ("advertised_model", "expected_status", "expected_error"),
    [(MODEL, 200, None), ("different-model", 503, "model_mismatch")],
)
def test_readiness_probes_run_while_all_generation_connections_are_held(
    advertised_model: str, expected_status: int, expected_error: str | None,
) -> None:
    async def scenario() -> None:
        upstream = HeldGenerationServer(advertised_model)
        async with upstream.running() as url:
            application = create_app(Settings(
                _env_file=None, environment="test", log_level="WARNING",
                inference_base_url=url, inference_model=MODEL,
                inference_connect_timeout=0.25,
                inference_probe_timeout=2, inference_generation_timeout=10,
            ))
            # Use the real factory, lifespan, adapter and /readyz dependency.
            async with application.router.lifespan_context(application):
                engine = application.state.inference_engine
                request = GenerationRequest(
                    messages=(Message(role="user", content="Synthetic test"),),
                    max_output_tokens=1,
                )
                generations = [asyncio.create_task(engine.generate(request)) for _ in range(4)]
                try:
                    await asyncio.wait_for(upstream.all_occupied.wait(), timeout=3)
                    async with httpx2.AsyncClient(
                        transport=httpx2.ASGITransport(app=application),
                        base_url="http://backend.test", trust_env=False,
                    ) as browser:
                        assert (await browser.get("/healthz")).status_code == 200
                        response = await asyncio.wait_for(browser.get("/readyz"), timeout=3)
                    assert all(not task.done() for task in generations)
                    assert response.status_code == expected_status
                    assert response.json() == {
                        "status": "ready" if expected_status == 200 else "not_ready",
                        "dependency": "inference", "error_code": expected_error,
                    }
                    assert upstream.probe_paths == ["/health", "/v1/models"]
                    upstream.release.set()
                    results = await asyncio.wait_for(asyncio.gather(*generations), timeout=3)
                    assert [result.text for result in results] == ["OK"] * 4
                finally:
                    upstream.release.set()
                    for task in generations:
                        task.cancel()
                    await asyncio.gather(*generations, return_exceptions=True)

    asyncio.run(scenario())

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

8. 다음 파일과의 연결

실제 시간 초과·취소 검증에서 다른 경계의 보장을 확인한다. 전체 구조는 Backend 개요, client 조립은 main.py, 오류 번역은 Adapter 구현과 연결된다.