Field Log · Entry

RAG Agent 테스트: Fake·Contract·Record/Replay 설계 (9/10)

오늘의 결론

• RAG Agent test의 기본 경로는 실제 LLM이 아니라 deterministic fake입니다. Clock·ID·random·network를 port로 주입해야 같은 입력에서 같은 trajectory를 재현할 수 있습니다.
• Fake가 그럴듯하게 동작하는지만 검사하지 않습니다. Fake adapter와 실제 HTTP adapter에 같은 contract suite를 통과시켜 interface drift를 막습니다.
• Network failure는 function mock보다 HTTP transport 경계에서 재현합니다. Timeout, 429, malformed JSON, 잘린 SSE, cancellation처럼 parser·retry·cleanup을 함께 검증합니다.
• Record/replay fixture는 raw traffic 보관소가 아닙니다. Authorization·cookie·PII·prompt·검색 문서를 allowlist 방식으로 제거하고 schema version과 provenance를 남깁니다.
• Exact answer string보다 불변식을 검사합니다. Citation이 실제 evidence를 가리키는지, terminal이 하나인지, budget과 authorization을 지켰는지가 핵심입니다.

앞 글에서는 bounded SSE stream과 disconnect·resume protocol을 만들었습니다. 이번에는 81~88편의 경계를 빠르고 재현 가능하게 검증하는 test pyramid를 구성합니다.

이 글의 대상과 학습 시간

• 대상: LLM·retrieval API 때문에 test가 느리거나 flaky한 Python 개발자
• 선수 지식: pytest, async Python, port/adapter 구조, 82~88편의 protocol
• 빠르게 읽기: 약 15분
• Fake·fault suite까지 따라 하기: 약 70분
• 산출물: deterministic test harness, shared contract suite, sanitized replay fixture

1. “실제 모델로 물어봤더니 맞았다”는 회귀 test가 아니다

Live LLM test 하나에 모든 기대를 걸면 실패 원인을 분리하기 어렵습니다.

• Model snapshot이나 sampling이 바뀜
• Retrieval index가 갱신됨
• Provider rate limit·regional outage가 발생
• Prompt version이 달라짐
• Clock·UUID·retry jitter 때문에 trace가 달라짐
• Network 비용과 latency 때문에 PR마다 실행하지 못함

이 test가 실패했을 때 application bug인지 upstream 변화인지 알 수 없습니다. Production RAG Agent는 서로 다른 질문을 서로 다른 층에서 검증해야 합니다.

층	질문	외부 network	실행 빈도
Pure unit	Context·citation·state transition이 맞는가	없음	모든 commit
Fake component	Runtime이 scripted trajectory를 지키는가	없음	모든 commit
Contract	모든 adapter가 같은 의미를 제공하는가	기본 없음	모든 PR
Transport integration	실제 HTTP serialization·parser·retry가 맞는가	Mock transport	모든 PR
Record/replay	알려진 provider wire shape를 다시 처리하는가	없음	모든 PR
Live smoke	Credential·endpoint·현재 provider가 연결되는가	있음	제한적
Offline eval	검색·근거·답변 품질 gate를 넘는가	선택	merge/release

위로 갈수록 비싸고 느리며 실패 원인이 많아집니다. 아래층을 넓게 만들고 live test는 작게 유지합니다.

2. Test 대상은 prose가 아니라 contract다

다음 assertion은 쉽게 깨집니다.

assert answer == "서울은 대한민국의 수도입니다."

같은 의미의 문장 변화도 실패합니다. 반대로 그 문장이 잘못된 evidence에서 나왔는지는 확인하지 못합니다.

RAG Agent에서 더 중요한 불변식은 다음과 같습니다.

• Unauthorized document가 retrieval result와 prompt에 없음
• Context token budget을 초과하지 않음
• 모든 citation의 evidence ID와 source span이 존재함
• Unsupported claim에서는 답변을 보류하거나 한계를 표시함
• Tool call argument가 schema·policy validation을 통과함
• Retryable error만 budget 안에서 재시도함
• Event sequence가 단조 증가하고 terminal이 정확히 하나임
• 같은 idempotency key와 request hash가 같은 run을 반환함
• Run deadline 이후 새 external call을 시작하지 않음

Semantic quality는 별도 eval dataset으로 측정하되 software correctness와 섞지 않습니다.

3. Determinism을 dependency로 주입한다

Application 코드가 datetime.now(), uuid.uuid4(), random.random(), asyncio.sleep()을 직접 부르면 test가 wall clock과 운에 의존합니다.

from collections.abc import Awaitable
from datetime import datetime, timedelta, timezone
from typing import Protocol


class Clock(Protocol):
    def now(self) -> datetime: ...
    def monotonic(self) -> float: ...
    async def sleep(self, seconds: float) -> None: ...


class IdFactory(Protocol):
    def new(self, prefix: str) -> str: ...


class RandomSource(Protocol):
    def uniform(self, low: float, high: float) -> float: ...


class FakeClock:
    def __init__(self) -> None:
        self._wall = datetime(2026, 7, 16, tzinfo=timezone.utc)
        self._mono = 100.0

    def now(self) -> datetime:
        return self._wall

    def monotonic(self) -> float:
        return self._mono

    async def sleep(self, seconds: float) -> None:
        if seconds < 0:
            raise ValueError("sleep must be non-negative")
        self.advance(seconds)

    def advance(self, seconds: float) -> None:
        self._mono += seconds
        self._wall += timedelta(seconds=seconds)


class SequentialIds:
    def __init__(self) -> None:
        self._next = 1

    def new(self, prefix: str) -> str:
        value = f"{prefix}_test_{self._next:04d}"
        self._next += 1
        return value

Fake clock은 retry backoff 단위 test를 빠르게 만들지만, event loop scheduling과 실제 socket timeout까지 증명하지는 않습니다. 실제 clock을 쓰는 소수 integration test도 남깁니다.

Global patch보다 constructor injection

monkeypatch는 environment나 legacy global을 임시 변경할 때 유용합니다. 그러나 직접 통제하는 application dependency는 constructor나 function argument로 받는 편이 의존성과 lifetime이 더 분명합니다.

runtime = AgentRuntime(
    llm=FakeLlm(...),
    retrieval=FakeRetrieval(...),
    clock=FakeClock(),
    ids=SequentialIds(),
    random=FixedRandom(0.5),
)

4. Fake·Stub·Mock·Spy를 구분한다

용어 자체보다 test가 무엇을 보장하는지 분명해야 합니다.

• Stub: 정해 둔 값이나 오류를 반환
• Fake: 단순하지만 동작하는 구현. 예: memory event store
• Spy: 받은 call을 기록해 나중에 검사
• Mock: 기대한 interaction이 발생했는지 자체적으로 판정

한 객체가 여러 역할을 할 수 있지만, production semantics를 흉내 내지 못하는 fake가 test를 통과시키는 위험을 항상 봐야 합니다.

Scripted LLM fake

from collections import deque
from dataclasses import dataclass


@dataclass(frozen=True)
class Emit:
    event: LlmEvent


@dataclass(frozen=True)
class Raise:
    error: Exception


ScriptStep = Emit | Raise


class ScriptedLlm:
    def __init__(self, scripts: list[list[ScriptStep]]) -> None:
        self._scripts = deque(deque(script) for script in scripts)
        self.requests: list[LlmRequest] = []

    async def stream(self, request: LlmRequest):
        self.requests.append(request)
        if not self._scripts:
            raise AssertionError("unexpected LLM call")

        script = self._scripts.popleft()
        while script:
            match script.popleft():
                case Emit(event):
                    yield event
                case Raise(error):
                    raise error

    def assert_consumed(self) -> None:
        if self._scripts:
            raise AssertionError(f"unused LLM scripts: {len(self._scripts)}")

Script는 “이 prompt 문자열이면 이 답”보다 call 순서와 protocol event를 표현합니다. 예상하지 않은 추가 LLM call과 남은 script도 test 실패로 만들어 runaway loop를 잡습니다.

Scripted retrieval fake

class FakeRetrieval:
    def __init__(self, results: list[Evidence]) -> None:
        self._results = results
        self.calls: list[SearchRequest] = []

    async def search(self, request: SearchRequest) -> list[Evidence]:
        self.calls.append(request)
        return [item.model_copy(deep=True) for item in self._results]

Deep copy를 반환해 application이 fixture를 실수로 mutate해도 다음 test에 누출되지 않게 합니다.

5. 하나의 vertical slice를 fake로 검증한다

import pytest


@pytest.mark.asyncio
async def test_answer_uses_only_authorized_evidence_and_terminates():
    retrieval = FakeRetrieval([
        evidence("ev_a", acl=["team-a"], text="허용된 근거"),
    ])
    llm = ScriptedLlm([[
        Emit(text_delta("허용된 ")),
        Emit(text_delta("근거입니다.")),
        Emit(citation("ev_a", answer_start=0, answer_end=7)),
        Emit(message_done(input_tokens=120, output_tokens=8)),
    ]])
    runtime = make_runtime(retrieval=retrieval, llm=llm)

    events = [
        event async for event in runtime.stream(
            query="근거를 알려줘",
            principal=principal(groups={"team-a"}),
        )
    ]

    assert events[-1].type == "done"
    assert sum(event.terminal for event in events) == 1
    assert citation_ids(events) == {"ev_a"}
    assert all(call.principal_groups == {"team-a"} for call in retrieval.calls)
    assert "허용된 근거" in llm.requests[0].rendered_context
    llm.assert_consumed()

이 test에는 실제 HTTP도 vector DB도 없습니다. 하지만 authorization filter가 request에 전달되고, evidence가 prompt에 들어가며, citation과 terminal이 연결되는 application contract는 검증합니다.

6. Fake와 real adapter에 같은 contract suite를 적용한다

Fake가 production adapter와 다른 의미를 제공하면 unit test가 거짓 안전감을 줍니다. Contract suite는 구현체가 지켜야 할 공통 의미를 함수로 만듭니다.

from collections.abc import Callable

LlmFactory = Callable[[list[ProviderFixture]], LlmPort]


async def assert_llm_contract(factory: LlmFactory) -> None:
    client = factory([fixture_text("안녕"), fixture_done(3, 2)])
    events = [event async for event in client.stream(minimal_request())]

    assert [event.type for event in events] == ["text_delta", "message_done"]
    assert join_text(events) == "안녕"
    assert events[-1].usage.output_tokens == 2


async def assert_error_contract(factory: LlmFactory) -> None:
    client = factory([fixture_http_error(429, retry_after="2")])

    with pytest.raises(RateLimited) as caught:
        _ = [event async for event in client.stream(minimal_request())]

    assert caught.value.retryable is True
    assert caught.value.retry_after_seconds == 2.0


@pytest.mark.asyncio
async def test_scripted_fake_contract():
    await assert_llm_contract(scripted_llm_factory)
    await assert_error_contract(scripted_llm_factory)


@pytest.mark.asyncio
async def test_http_adapter_contract():
    await assert_llm_contract(mock_transport_llm_factory)
    await assert_error_contract(mock_transport_llm_factory)

Contract는 implementation detail이 아니라 consumer가 의존하는 의미를 검사합니다.

• Empty delta를 무시하는가
• Usage가 terminal에서 한 번 확정되는가
• 429가 typed RateLimited로 변환되는가
• Malformed terminal이 ProtocolError인가
• Cancellation을 retryable failure로 바꾸지 않는가
• Provider ID가 trace metadata로 보존되는가

실제 provider live contract는 소수 case만 별도 schedule로 실행합니다.

7. HTTPX MockTransport로 wire boundary를 검증한다

LLM method 자체를 patch하면 request serialization, header, status mapping, stream parser를 건너뜁니다. HTTPX는 request를 받아 미리 정한 response를 반환하는 MockTransport를 제공합니다.

import json

import httpx
import pytest


@pytest.mark.asyncio
async def test_adapter_serializes_request_and_maps_429():
    seen: list[httpx.Request] = []

    def handler(request: httpx.Request) -> httpx.Response:
        seen.append(request)
        return httpx.Response(
            status_code=429,
            headers={
                "retry-after": "2",
                "x-request-id": "provider_req_test",
            },
            json={"error": {"code": "rate_limit"}},
        )

    transport = httpx.MockTransport(handler)
    async with httpx.AsyncClient(
        transport=transport,
        base_url="https://provider.invalid",
    ) as http:
        adapter = ProviderLlmAdapter(http=http, api_key="test-secret")
        with pytest.raises(RateLimited) as caught:
            _ = [event async for event in adapter.stream(minimal_request())]

    assert len(seen) == 1
    assert seen[0].method == "POST"
    assert seen[0].url.path == "/v1/messages"
    assert json.loads(seen[0].content)["stream"] is True
    assert caught.value.provider_request_id == "provider_req_test"

Test failure output에 authorization header를 출력하지 않습니다. Credential 내용보다 header가 존재하고 adapter가 redaction policy를 지키는지 검사합니다.

잘린 stream 재현

class ChunkStream(httpx.AsyncByteStream):
    def __init__(self, chunks: list[bytes], *, fail_after: int | None = None) -> None:
        self._chunks = chunks
        self._fail_after = fail_after
        self.closed = False

    async def __aiter__(self):
        for index, chunk in enumerate(self._chunks):
            if self._fail_after == index:
                raise httpx.ReadError("scripted truncated stream")
            yield chunk

    async def aclose(self) -> None:
        self.closed = True


def truncated_handler(request: httpx.Request) -> httpx.Response:
    return httpx.Response(
        200,
        headers={"content-type": "text/event-stream"},
        stream=ChunkStream([
            b'event: content_block_delta\n',
            b'data: {"delta":{"text":"partial"}}\n\n',
            b'event: message_stop\ndata: {}\n\n',
        ], fail_after=2),
    )

이 fixture로 다음을 함께 확인합니다.

• Partial text 뒤 network error가 성공 done으로 바뀌지 않음
• Response stream이 aclose() 됨
• Retry policy가 이미 노출한 delta를 무작정 중복 전송하지 않음
• Public error와 provider exception detail이 분리됨

8. Fault matrix를 표로 관리한다

Happy path fixture만 늘리면 복원력 코드는 실제로 실행되지 않습니다.

Fault	주입 위치	기대 결과
Connect timeout	HTTP transport	retry budget 내 backoff
Read timeout before delta	stream open/read	정책에 따라 retry 가능
Read error after delta	stream parser	terminal error, 중복 text 금지
HTTP 429 + Retry-After	response mapping	server hint와 deadline 반영
HTTP 500	response mapping	bounded retry 또는 breaker count
Malformed JSON	decoder	typed protocol error
Unknown event type	parser	forward-compatible ignore 또는 명시 거부
Missing terminal	parser/runtime	protocol error
Duplicate terminal	runtime	invariant violation
Tool timeout	tool adapter	cancellation·receipt 기록
Client disconnect	SSE consumer	cancel 또는 unsubscribe policy
DB version conflict	persistence	reload·conflict response
Stale ACL	retrieval adapter	result reject·audit metric

각 row에는 owner test file과 최소 한 개 assertion을 연결합니다. “Retry test 있음”이 아니라 어떤 side effect가 중복되지 않는지도 봅니다.

9. Retry test는 clock과 call count를 함께 본다

@pytest.mark.asyncio
async def test_retry_stops_at_shared_deadline():
    clock = FakeClock()
    llm = ScriptedLlm([
        [Raise(UpstreamTimeout("first"))],
        [Raise(UpstreamTimeout("second"))],
        [Emit(text_delta("must not happen"))],
    ])
    client = ResilientLlm(
        inner=llm,
        clock=clock,
        random=FixedRandom(0.0),
        retry=RetryPolicy(max_attempts=5, base_seconds=1.0),
    )

    with pytest.raises(DeadlineExceeded):
        _ = [
            event async for event in client.stream(
                minimal_request(deadline_monotonic=101.5)
            )
        ]

    assert len(llm.requests) == 2
    assert clock.monotonic() <= 101.5

단순히 exception type만 보지 않고 attempt 수, elapsed budget, request ID, side effect count를 함께 확인합니다.

10. Record/replay는 redacted contract fixture다

실제 provider response shape를 한 번 capture해 offline에서 replay하면 SDK·parser regression을 찾는 데 유용합니다. 그러나 raw HTTP dump를 repository에 넣으면 다음이 유출될 수 있습니다.

• Authorization·cookie·signed URL
• User query와 session identifier
• Retrieved document와 개인·기밀 정보
• Prompt의 system instruction
• Provider request ID와 account metadata

따라서 cassette는 allowlist schema로 다시 만듭니다.

{
  "schema_version": 2,
  "case_id": "text_then_tool_then_done",
  "recorded_at": "2026-07-16T00:00:00Z",
  "provider_family": "example-messages-api",
  "request": {
    "method": "POST",
    "path": "/v1/messages",
    "body_fixture": "minimal_tool_request_v1"
  },
  "response": {
    "status": 200,
    "headers": {"content-type": "text/event-stream"},
    "chunks_base64": ["ZXZlbnQ6Li4u"]
  },
  "redaction": {
    "policy_version": 3,
    "verified_by": "manual-review"
  }
}

Raw request body 대신 synthetic fixture name을 저장하고, response도 실제 사용자 내용이 없는 synthetic account에서 기록합니다.

Redaction은 denylist보다 allowlist

SAFE_RESPONSE_HEADERS = {
    "content-type",
    "retry-after",
    "x-request-id",  # 값은 synthetic token으로 치환
}


def sanitize_headers(headers: httpx.Headers) -> dict[str, str]:
    safe: dict[str, str] = {}
    for name in SAFE_RESPONSE_HEADERS:
        if name in headers:
            safe[name] = (
                "provider_req_redacted"
                if name == "x-request-id"
                else headers[name]
            )
    return safe

api-key, authorization 같은 알려진 이름만 지우는 denylist는 새 custom header를 놓칠 수 있습니다. 보관 목적에 필요한 field만 선택합니다.

Cassette gate

• Schema validation 통과
• Secret scanner 통과
• Email·phone·token pattern scan
• 최대 file size 제한
• Manual review metadata
• 만료일·provider API version 기록
• Binary/base64 decode 후에도 scan

Capture mode는 developer가 명시적으로 켤 때만 동작하고 CI 기본값에서는 network를 차단합니다.

11. Replay가 증명하지 못하는 것

Record/replay가 성공해도 다음은 보장되지 않습니다.

• 현재 provider가 같은 schema를 반환함
• Credential와 endpoint가 유효함
• Rate limit·latency가 production과 같음
• Model 품질이 유지됨
• HTTP/2, proxy, TLS, cancellation timing이 같음

Replay는 과거에 관찰한 wire fixture를 현재 parser가 처리한다는 회귀 test입니다. 최신 compatibility를 확인하는 sparse live smoke와 역할이 다릅니다.

12. Live test는 작고 명시적으로 격리한다

import os

import pytest


live = pytest.mark.skipif(
    os.getenv("RUN_LIVE_LLM_TESTS") != "1",
    reason="live provider tests are opt-in",
)


@live
@pytest.mark.asyncio
async def test_provider_live_minimal_stream():
    adapter = build_live_adapter_from_environment()
    events = [event async for event in adapter.stream(live_smoke_request())]

    assert events[-1].type == "message_done"
    assert sum(event.type == "message_done" for event in events) == 1
    assert join_text(events).strip()

Live test 원칙:

• Synthetic prompt와 public document만 사용
• 낮은 token·cost ceiling
• 별도 low-privilege credential
• Retry를 적게 두고 outage를 product bug와 구분
• PR 필수 gate가 아니라 nightly 또는 release candidate gate
• 실패 시 sanitized request ID와 provider status page 연결

13. Property와 stateful test로 순서 조합을 탐색한다

Idempotency·event replay·bounded state machine은 단일 example보다 action sequence에서 깨집니다.

from hypothesis import given, strategies as st


@given(st.lists(st.sampled_from(["delta", "citation"]), max_size=30))
def test_terminal_event_is_final(prefix_types: list[str]):
    events = [make_event(kind, index + 1) for index, kind in enumerate(prefix_types)]
    events.append(make_event("done", len(events) + 1))

    validate_stream(events)
    assert events[-1].terminal
    assert all(not event.terminal for event in events[:-1])

Hypothesis rule-based state machine은 다음 action 조합을 생성할 수 있습니다.

• create(key, hash)
• duplicate(key, same_hash)
• conflict(key, different_hash)
• append(expected_version)
• worker_crash()
• lease_expire()
• takeover()
• complete(owner_token)

항상 유지할 invariant:

• 한 key·hash의 completed response는 하나
• Session sequence에 gap·duplicate 없음
• Old owner가 takeover 뒤 complete하지 못함
• Terminal 뒤 append 불가

Shrinking은 실패를 만든 긴 sequence를 작은 재현 case로 줄이는 데 유용합니다. 단순한 함수라면 stateful test보다 일반 @given이 더 읽기 쉬울 수 있습니다.

14. Offline eval과 software test를 연결하되 섞지 않는다

Software test가 모두 통과해도 retrieval quality는 나쁠 수 있습니다. 반대로 model 문장이 조금 달라져도 software contract는 정상일 수 있습니다.

Local golden dataset 예시

{
  "case_id": "policy_retention_001",
  "query": "삭제 요청 후 백업 보존 기간은?",
  "principal_fixture": "privacy-team",
  "expected_evidence_ids": ["policy-v7:p12:s3"],
  "forbidden_evidence_ids": ["policy-v6:p9:s1"],
  "must_abstain": false,
  "required_claims": ["백업은 별도 만료 절차를 따른다"]
}

Release gate를 분리합니다.

• Retrieval gate: Recall@k, stale/forbidden result rate
• Grounding gate: Citation validity, claim support
• Behavior gate: Abstention, tool policy violation
• System gate: p95 latency, timeout/error budget, cost ceiling

Exact prose snapshot은 UI formatting처럼 정말 동일해야 하는 부분에만 씁니다.

15. CI에서 network를 기본 거부한다

Test가 실수로 production endpoint를 호출하면 flaky와 비용뿐 아니라 data leak 위험도 생깁니다.

방어선:

1. Domain port를 fake로 주입
2. HTTP client는 test transport만 사용
3. Credential environment variable 제거
4. 예상하지 않은 host request는 즉시 실패
5. Live marker는 별도 job·별도 secret scope에서만 허용

def deny_unexpected_network(request: httpx.Request) -> httpx.Response:
    raise AssertionError(f"unexpected network host: {request.url.host}")


@pytest.fixture
def offline_http() -> httpx.AsyncClient:
    return httpx.AsyncClient(
        transport=httpx.MockTransport(deny_unexpected_network),
        base_url="https://offline.invalid",
    )

Fixture teardown에서 client를 닫도록 실제 코드는 async fixture 또는 context manager로 감쌉니다.

16. Flaky test를 retry로 숨기지 않는다

Test retry plugin으로 초록색을 만들면 race가 남습니다. Flaky test에는 최소 다음 metadata를 붙입니다.

• 최초 발견 commit과 owner
• 실패 seed·cassette·trace ID
• 재현 command
• Clock/network/random 의존성
• Quarantine 만료일

Retry 횟수보다 왜 nondeterministic한지를 고칩니다. 동시성 test는 arbitrary sleep(0.1) 대신 barrier, event, queue depth처럼 관찰 가능한 synchronization point를 사용합니다.

17. 자주 실패하는 테스트 전략

1. 모든 것을 MagicMock으로 만든다

Async iterator·context manager·stream cleanup contract가 사라집니다. Protocol을 구현한 작은 fake와 transport fixture를 사용합니다.

2. Fake만 test하고 real adapter에 같은 contract를 적용하지 않는다

Fake가 production과 다른 event type이나 error semantics를 가질 수 있습니다.

3. 실제 답변 전문을 snapshot으로 고정한다

무해한 문장 변화에 깨지고 citation·ACL 오류는 놓칩니다. 의미 불변식과 구조를 검사합니다.

4. Raw cassette를 commit한다

Header를 지워도 body·chunk·base64 안에 prompt와 문서가 남을 수 있습니다. Synthetic capture와 allowlist redaction을 사용합니다.

5. Replay를 현재 provider compatibility로 해석한다

Replay는 과거 fixture test입니다. 소수 live smoke를 별도로 둡니다.

6. Timeout test에서 실제로 30초 기다린다

Clock을 주입해 policy를 빠르게 test하고 실제 socket timeout은 작은 integration case로 확인합니다.

7. Test가 예상치 못한 network를 허용한다

Offline job은 기본 deny로 구성합니다.

직접 적용 체크리스트

• Clock·ID·random·network가 주입 가능한 port다.
• 예상하지 않은 call과 남은 script를 fake가 실패시킨다.
• Fake와 HTTP adapter가 같은 contract suite를 통과한다.
• Request serialization·status mapping·stream cleanup을 transport에서 검사한다.
• Timeout·429·malformed JSON·truncated stream·cancel fault가 있다.
• Exact prose 대신 ACL·citation·terminal·budget 불변식을 검사한다.
• Cassette는 synthetic input과 allowlist redaction을 사용한다.
• Base64 decode 후까지 secret·PII scan을 한다.
• Offline CI는 예상하지 않은 network를 거부한다.
• Live smoke는 opt-in·low-cost·low-privilege다.
• Offline eval gate와 software correctness test가 분리돼 있다.
• Flaky test의 seed·owner·재현 command를 보존한다.

스스로 확인하기

1. Function mock보다 HTTP transport fake가 더 넓게 검증하는 경계는 무엇인가?
2. Fake adapter에만 있는 편리한 동작이 왜 위험한가?
3. Partial delta 뒤 read error를 무작정 retry하면 어떤 중복이 생길 수 있는가?
4. Cassette에서 header만 redaction해도 부족한 이유는 무엇인가?
5. Software test와 RAG offline eval의 실패가 각각 무엇을 뜻하는가?

자주 묻는 질문

Q1. Mock을 전혀 쓰지 말아야 하나요?

아닙니다. 좁은 interaction을 확인할 때 유용합니다. 다만 domain 경계 전체를 loose mock으로 대체하면 실제 protocol과 멀어지므로 typed fake와 contract test를 함께 둡니다.

Q2. Provider SDK 자체를 fake하면 충분한가요?

SDK event를 domain event로 변환하는 adapter는 검증할 수 있지만 raw HTTP serialization·status·stream parser를 SDK가 소유한다면 그 경계는 SDK update와 sparse live test로 확인합니다. 직접 HTTP adapter를 만들었다면 transport test가 필요합니다.

Q3. Record fixture는 얼마나 자주 갱신하나요?

Provider API version·SDK version이 바뀌거나 live contract가 drift를 감지할 때 갱신합니다. 이유 없이 golden을 덮어쓰지 말고 diff와 manual review를 남깁니다.

Q4. LLM 응답의 의미를 deterministic하게 어떻게 검사하나요?

핵심 claim·citation·abstention 같은 rubric을 structured evaluator로 검사합니다. Judge model을 쓰더라도 calibration dataset, version, threshold, human audit sample이 필요합니다.

Q5. Test pyramid 비율은 몇 퍼센트가 적당한가요?

고정 비율보다 feedback time과 failure localization이 기준입니다. 모든 commit에서 수초~수분 안에 핵심 contract를 검증하고, 비싼 live/eval은 merge·release cadence에 맞춥니다.

마무리

RAG Agent test harness는 다음 네 질문에 답해야 합니다.

1. 같은 입력과 script를 다시 실행하면 같은 state transition을 얻는가?
2. Fake와 real adapter가 같은 error·stream semantics를 제공하는가?
3. 실제로 발생하기 어려운 failure를 빠르고 의도적으로 만들 수 있는가?
4. 외부 서비스 없이도 release를 막을 핵심 불변식을 검증할 수 있는가?

다음 글에서는 지금까지 만든 application을 container로 묶고, CI gate·OpenTelemetry·canary·rollback과 release manifest까지 연결해 실제 배포 가능한 capstone으로 완결합니다.

검증 정보

• 글의 성격: HTTPX·pytest·Python·Hypothesis 공식 문서 해설 + 작성자 test architecture 제안 + 가상 Python fixture
• 마지막 검증일: 2026-07-16
• 검증 기준: HTTPX MockTransport/AsyncByteStream, pytest fixture·monkeypatch, Hypothesis rule-based state machine
• 실행 주의: pytest.mark.asyncio와 async fixture는 선택한 pytest async plugin version에 맞춰 설정
• 수치 표기: queue·timeout·case 수와 CI 빈도는 설명용이며 repository feedback SLO와 비용으로 결정
• 경험 표기: Provider path·request ID·golden case는 실제 운영 정보가 아닌 synthetic 예시

참고자료

• HTTPX Transports and MockTransport
• HTTPX Async Support and streaming
• pytest fixtures
• pytest monkeypatch
• Python unittest.mock autospeccing
• Hypothesis stateful testing
• Python asyncio synchronization primitives