MINERVA 고급 테스트 패턴 — Property·Snapshot·동시성·Contract·Mutation·CI 분리

현재 진단 위에 얹는 테스트 패턴 라이브러리

12-0편이 현재 테스트 구조의 진단·검증된 영역·미검증 영역·보강 우선순위를 다뤘다면, 본 글은 그 보강에 적용할 6가지 고급 테스트 패턴(Property-Based, Snapshot, 동시성, Contract, Mutation, 비결정성 처리)과 CI 분리 전략, PG 마이그레이션 회귀 테스트를 정리한다.

Agent
저자

Kwangmin Kim

공개

2026년 05월 06일

1 본 글의 위치

12-0편 테스트 전략 분석이 현재 70+ 케이스의 검증 범위와 미검증 영역을 진단했다면, 본 글은 그 보강에 동원할 테스트 패턴 라이브러리를 정리한다. 6가지 고급 패턴 + CI 분리 전략 + PG 마이그레이션 회귀 테스트로 구성된다.

선행 학습

2 Property-Based 테스트 — Citation 파서

_extract_cited_indices()는 정규식 기반이라 엣지 케이스가 많다. Hypothesis로 속성 기반 테스트를 추가하면 unit 케이스가 놓치는 입력을 자동 탐색한다.

from hypothesis import given, strategies as st

@given(st.integers(min_value=1, max_value=999))
def test_single_citation_marker_always_extracted(n: int):
    """[N] 형식의 마커는 항상 인덱스 N으로 추출되어야 한다."""
    agent = _make_agent()
    text = f"본문 [{n}] 입니다."
    assert agent._extract_cited_indices(text) == {n}

@given(st.lists(st.integers(min_value=1, max_value=999), min_size=1, max_size=20, unique=True))
def test_multiple_citations_all_extracted(indices: list[int]):
    """[a] [b] [c]... 모든 인덱스가 추출되어야 한다."""
    agent = _make_agent()
    text = " ".join(f"[{n}]" for n in indices)
    assert agent._extract_cited_indices(text) == set(indices)

@given(st.text())
def test_extract_never_raises(text: str):
    """어떤 텍스트가 들어와도 예외가 나면 안 된다."""
    agent = _make_agent()
    result = agent._extract_cited_indices(text)
    assert isinstance(result, set)
    assert all(isinstance(n, int) for n in result)

test_extract_never_raises는 robustness 검증이다. Hypothesis가 control 문자, 유니코드, 빈 문자열, 매우 긴 문자열 등을 자동 생성한다.

3 Snapshot 테스트 — LLM 응답 회귀

LLM은 비결정적이지만 응답 구조(answer 형식, citation 마커 위치, 섹션 헤더)는 회귀를 탐지할 가치가 있다. 정확한 텍스트 일치가 아닌 구조 매칭으로 접근한다.

import pytest
from jsonschema import validate

ANSWER_SCHEMA = {
    "type": "object",
    "required": ["text", "citations", "run_id", "model"],
    "properties": {
        "text": {"type": "string", "minLength": 10},
        "citations": {
            "type": "array",
            "items": {
                "type": "object",
                "required": ["index", "content"],
                "properties": {
                    "index": {"type": "integer", "minimum": 1},
                    "content": {"type": "string"},
                    "score": {"type": ["number", "null"]},
                },
            },
        },
        "model": {"type": "string"},
        "latency_ms": {"type": ["integer", "null"], "minimum": 0},
    },
}

@pytest.mark.integration  # CI에서 별도 마커로 격리
def test_response_schema_for_canonical_query():
    """대표 질문의 응답이 스키마를 만족하는지."""
    agent = QnaChatbotAgent(documents=DEFAULT_DOCUMENTS, default_document="표준화", config=...)
    response = agent.run(Query(text="데이터 표준화의 핵심 원칙은?"))
    validate(instance=response.model_dump(), schema=ANSWER_SCHEMA)

@pytest.mark.integration
def test_citation_markers_match_citations():
    """답변 본문의 [N] 마커와 citations 배열이 일관성을 갖는지."""
    response = agent.run(Query(text="..."))
    cited_indices = re.findall(r'\[(\d+)\]', response.text)
    citation_indices = {c.index for c in response.citations}
    extracted = {int(n) for n in cited_indices}
    # 본문에서 인용된 모든 인덱스가 citations에 존재해야 함
    assert extracted.issubset(citation_indices)

이 테스트는 LLM 호출이 필요하므로 CI의 일반 단위 테스트 단계와 분리한다(pytest -m "not integration").

4 동시성 부하 테스트

_agent_cache_lock의 double-checked locking 정확성은 concurrent.futures로 직접 검증할 수 있다.

import concurrent.futures
from unittest.mock import patch

def test_agent_cache_creates_only_one_instance_per_key():
    """동시 N개 요청이 같은 키로 들어와도 agent는 1개만 생성되어야 한다."""
    from services.api.routers.qna_chatbot import _build_agent, _agent_cache, _agent_cache_lock

    _agent_cache.clear()
    creation_count = {"n": 0}
    original_init = QnaChatbotAgent.__init__

    def counting_init(self, *args, **kwargs):
        creation_count["n"] += 1
        time.sleep(0.05)  # 초기화 비용 시뮬레이션
        original_init(self, *args, **kwargs)

    with patch.object(QnaChatbotAgent, "__init__", counting_init):
        with concurrent.futures.ThreadPoolExecutor(max_workers=20) as ex:
            futures = [ex.submit(_build_agent, user_id=None) for _ in range(20)]
            results = [f.result() for f in futures]

    # 모든 결과가 같은 인스턴스를 가리켜야 함
    agents = {id(r[0]) for r in results}
    assert len(agents) == 1
    # 생성은 1회만 발생해야 함
    assert creation_count["n"] == 1


def test_agent_cache_multiple_keys_create_separate_instances():
    """다른 키 N개가 동시에 들어오면 N개의 agent가 생성된다."""
    # 글로벌 _active_config 경쟁 조건도 함께 검증할 수 있다

5 Contract 테스트 — Agent ↔︎ Router

라우터와 agent는 Query/Response 스키마로 결합되어 있다. 한쪽이 변경되어도 contract가 깨지지 않는지 검증한다.

from fastapi.testclient import TestClient
from services.api.main import app

client = TestClient(app)

def test_run_response_schema_matches_pydantic():
    """/run의 JSON 응답이 RunResponse Pydantic 모델에 역직렬화 가능한지."""
    with patch("services.api.routers.qna_chatbot._build_agent") as mock_build:
        mock_response = Response(
            text="답변",
            citations=[Citation(index=1, content="doc")],
            run_id="abc",
            model="gpt-4.1",
            latency_ms=100,
            ttft_ms=100,
        )
        mock_agent = MagicMock()
        mock_agent.run.return_value = mock_response
        mock_build.return_value = (mock_agent, None, None, {})

        resp = client.post("/agents/qna_chatbot/run", json={"text": "Q"})
        assert resp.status_code == 200

        # 역직렬화 검증
        run_response = RunResponse(**resp.json())
        assert run_response.response.text == "답변"
        assert run_response.response.citations[0].index == 1


def test_stream_event_schema():
    """/stream의 SSE 이벤트가 StreamEvent 스키마를 만족하는지."""
    with patch(...) as mock_build:
        mock_agent = MagicMock()
        mock_agent.stream.return_value = iter([
            StreamEvent(type="token", text="안"),
            StreamEvent(type="token", text="녕"),
            StreamEvent(type="done", response=mock_response),
        ])
        mock_build.return_value = (mock_agent, None, None, {})

        with client.stream("POST", "/agents/qna_chatbot/stream", json={"text": "Q"}) as resp:
            assert resp.headers["content-type"].startswith("text/event-stream")
            events = []
            for line in resp.iter_lines():
                if line.startswith("data: "):
                    events.append(StreamEvent(**json.loads(line[6:])))
            assert events[0].type == "token"
            assert events[-1].type == "done"

6 Mutation 테스트 도입 가능성

mutmut 같은 mutation testing 도구는 코드를 의도적으로 변형(== → !=, < → <=)한 뒤 테스트가 잡아내는지 측정한다.

poetry add --group dev mutmut
mutmut run --paths-to-mutate src/agents/qna_chatbot/agent.py
mutmut results

후보 함수 (현재 단위 테스트가 풍부한 함수): - _truncate_with_lead_tail() — 경계값 365/366 검증으로 mutation 잘 잡을 듯 - _filter_citations() — if not cited: 분기 mutation 검증 - _extract_cited_indices() — 정규식 변형 검증

mutation score(살아남은 mutation 비율)가 90% 이상이면 단위 테스트 충분, 80% 이하면 보강 필요.

7 비결정성 처리 전략

LLM 테스트에서 비결정성을 다루는 세 가지 접근을 정리한다.

전략 1 — 경계 격리 (현재 채택): LLM/RAG를 테스트에서 완전히 제외하고, 그 경계에서 입출력되는 순수 함수만 검증한다. _build_context_string(), _filter_citations() 등이 이에 해당한다.

전략 2 — 결정론적 mock: _FakeChain처럼 예측 가능한 출력을 생성하는 mock으로 스트리밍 흐름 전체를 검증한다. LLM 응답 품질이 아닌 이벤트 순서·형식을 검증한다.

전략 3 — 스냅샷 기반 회귀 (미구현): 실제 LLM을 실행하되, 응답을 스냅샷으로 저장하고 다음 실행과 비교한다. 프롬프트 변경이 응답 구조를 깨뜨리는 회귀를 탐지하는 데 유용하다. 비결정성으로 인해 완전 일치가 아닌 구조 검증(JSON 스키마, 인용 마커 존재 여부)이 현실적이다.

8 PG 마이그레이션 회귀 테스트

metrics_logger.py의 JSONL 스키마는 향후 PostgreSQL 컬럼과 1:1 매핑된다. 마이그레이션 시 record_from_response()/record_from_error()가 생성하는 모든 키가 PG 스키마와 호환되는지 회귀 테스트로 보장할 수 있다.

EXPECTED_PG_COLUMNS = {
    "timestamp", "run_id", "agent_name", "session_id", "user_id",
    "query", "answer", "total_time_ms", "ttft_ms", "success",
    "error_type", "error_message", "model_name",
    "input_tokens", "output_tokens", "citation_count",
    "documents_retrieved",
    "has_table", "has_code_block", "has_qna_format",
    "has_ds_format", "has_principle",
    "experiment_name", "arm_id", "extras",
}

def test_record_from_response_keys_match_pg_schema():
    query = Query(text="Q", user_id="u1")
    response = Response(
        text="A", citations=[],
        run_id="r1", model="gpt-4.1",
        latency_ms=100, ttft_ms=50,
    )
    record = record_from_response("qna_chatbot", query, response)
    assert set(record.keys()) == EXPECTED_PG_COLUMNS

def test_record_from_error_keys_match_pg_schema():
    query = Query(text="Q", user_id="u1")
    record = record_from_error("qna_chatbot", query, RuntimeError("fail"))
    assert set(record.keys()) == EXPECTED_PG_COLUMNS
    assert record["success"] is False
    assert record["error_type"] == "RuntimeError"

스키마 변경이 필요하면 이 테스트가 먼저 깨지도록 두고, 명시적으로 EXPECTED_PG_COLUMNS를 업데이트한다.

9 CI 분리 전략

테스트 단계를 외부 의존성 기준으로 분리한다.

단계 마커 환경 빈도
1. unit (none) 외부 의존 0 매 커밋
2. property @pytest.mark.property 외부 의존 0 매 PR
3. concurrency @pytest.mark.concurrency 외부 의존 0 매 PR
4. contract @pytest.mark.contract mock된 라우터 매 PR
5. integration @pytest.mark.integration Azure Search + OpenAI 일 1회
6. snapshot @pytest.mark.snapshot 실제 LLM 주 1회 
# pytest.ini
[pytest]
markers =
    property: Hypothesis 기반 속성 테스트
    concurrency: 동시성·thread safety 테스트
    contract: Agent <-> Router 계약
    integration: Azure/OpenAI 실제 호출
    snapshot: LLM 응답 회귀
# 매 커밋 (~3초)
pytest -m "not property and not concurrency and not contract and not integration and not snapshot"

# 매 PR (~30초)
pytest -m "not integration and not snapshot"

# 일간 (~5분, 외부 API 비용 발생)
pytest -m "integration"

# 주간 (~10분)
pytest -m "snapshot"

10 정리

패턴 대상 CI 마커
Property-Based 정규식·경계값 함수 (_extract_cited_indices) property
Snapshot LLM 응답 구조 (실제 호출 + JSON Schema 검증) integration / snapshot
동시성 부하 _agent_cache_lock double-checked locking concurrency
Contract Agent ↔︎ Router 스키마 호환성 contract
Mutation 단위 테스트 풍부한 함수의 mutation score (수동)
비결정성 처리 경계 격리 + 결정론 mock + 스냅샷 회귀 (전략)
PG 회귀 record_from_response/error 키 셋 (unit)

12-0편의 진단·보강 우선순위 위에 본 글의 패턴을 동원하면, MINERVA의 테스트 커버리지가 결정론 함수 → 라우터 통합 → 실제 LLM 회귀의 3계층으로 정합되게 확장된다.

11 관련 주제

선행 학습

Phase C-2 — LangGraph 전환 후 재적용

Phase C-9 (예정) 연결

  • C30 스킬 테스트와 품질 게이트 — 본 패턴들이 150+ 스킬 평가 파이프라인에 확장되는 형태

Subscribe

Enjoy this blog? Get notified of new posts by email: