워크플로 테스트 — 이벤트 기반으로 검증하기

워크플로 테스트 — 이벤트 기반으로 검증하기

워크플로는 이벤트 기반 시스템으로 테스트하는 게 가장 쉬워요. 워크플로를 실행하고, 스트리밍된 이벤트를 모으고, 최종 결과를 검증하며, 상태가 중요할 땐 컨텍스트 상태를 들여다보면 됩니다. workflows.testing 모듈에 이를 위한 작은 러너가 있어요.

출처: 공식문서

from llama_index.core.workflow.testing import WorkflowTestRunner

엔드투엔드 테스트

WorkflowTestRunner는 워크플로를 시작하고, 이벤트 스트림을 소진하며, 최종 결과를 기다린 다음, 그 모든 것을 하나의 객체로 돌려줘요.

import pytest
from llama_index.core.workflow import Workflow, Context, step
from llama_index.core.workflow.events import Event, StartEvent, StopEvent
from llama_index.core.workflow.testing import WorkflowTestRunner

class Progress(Event):
    message: str
class Done(Event):
    value: str

class ExampleWorkflow(Workflow):
    @step
    async def start(self, ctx: Context, ev: StartEvent) -> Done:
        ctx.write_event_to_stream(Progress(message="started"))
        return Done(value=ev.topic.upper())

    @step
    async def finish(self, ev: Done) -> StopEvent:
        return StopEvent(result=ev.value)

@pytest.mark.asyncio
async def test_workflow_streams_progress_and_returns_result() -> None:
    result = await WorkflowTestRunner(ExampleWorkflow()).run(
        start_event=StartEvent(topic="docs")
    )
    assert result.result == "DOCS"
    assert result.event_types[Progress] == 1
    assert any(isinstance(ev, Progress) and ev.message == "started"
               for ev in result.collected)

돌려받는 객체의 주요 필드는 이래요.

필드 의미
result 워크플로 핸들을 await해서 얻은 최종 값
collected 제외되지 않은, 스트리밍된 모든 이벤트
event_types 이벤트 클래스별 수집 이벤트 개수
ctx 최종 Context. 상태 검증이나 스냅샷에 유용

내부 이벤트 다루기

기본적으로 러너는 StepStateChanged를 포함한 내부 이벤트까지 노출해요. 실행 형태를 검증하고 싶을 때 유용하죠. 사용자 이벤트만 신경 쓴다면 expose_internal=False로 내부 이벤트를 끄거나, exclude_events=[StepStateChanged]로 소음이 많은 이벤트만 수집 목록에서 빼면 됩니다.

result = await WorkflowTestRunner(ExampleWorkflow()).run(
    start_event=StartEvent(topic="docs"),
    expose_internal=False,
)
# 또는
result = await WorkflowTestRunner(ExampleWorkflow()).run(
    start_event=StartEvent(topic="docs"),
    exclude_events=[StepStateChanged],
)

상태 검증

워크플로가 ctx.store에 뭔가를 쓴다면 반환된 컨텍스트를 사용해요.

class CounterState(BaseModel):
    count: int = Field(default=0)

@pytest.mark.asyncio
async def test_workflow_updates_state() -> None:
    result = await WorkflowTestRunner(CounterWorkflow()).run()
    state = await result.ctx.store.get_state()
    assert state.count == 1

내구성 있는 워크플로 코드라면, 실제 스냅샷과 복원을 거친 뒤의 동작을 검증하는 편이 좋아요. first.ctx.to_dict()로 스냅샷을 만들고, 새 워크플로 인스턴스에 Context.from_dict로 복원해 이어서 실행하세요. 이렇게 하면 단위 테스트가 놓치기 쉬운 실수들, 즉 직렬화할 수 없는 상태, 복원 시 import되지 않는 이벤트, 반복해도 안전하지 않은 부수 효과를 잡아냅니다.

더 알아보기