temporal-python-testing은(는) 무엇을 하나요?
pytest, 시간 스킵, 목킹 전략을 사용하여 Temporal 워크플로우를 테스트합니다. 단위 테스트, 통합 테스트, 재생 테스트 및 로컬 개발 환경 설정을 다룹니다. Temporal 워크플로우 테스트를 구현하거나 테스트 실패를 디버깅할 때 사용하세요.
개요
Temporal 워크플로우는 시간 종속적이고 상태를 유지하는 비즈니스 로직을 포함하기 때문에 테스트가 어려울 수 있습니다. 이 스킬은 pytest 기반의 테스트 패턴을 제공하여 워크플로우의 정확성을 보장합니다.
주요 기능
- 단위 테스트: 워크플로우를 격리된 환경에서 테스트 (예:
WorkflowEnvironment사용) - 통합 테스트: 실제 Temporal 서버와 상호작용하여 워크플로우 테스트
- 시간 스킵:
time_skipping유틸리티를 사용하여 장기 실행 워크플로우를 빠르게 테스트 - 재생 테스트: 기존 실행 기록을 기반으로 워크플로우 코드의 비결정적 변경 사항 감지
- 목킹: 활동(Activity) 및 외부 호출을 목킹하여 테스트 격리성 향상
지원 도구
- pytest: Python 테스트 프레임워크
- Temporal Python SDK:
temporalio패키지 - Docker Compose: 로컬 Temporal 서버 설정 (선택적)
사용 방법
설치
pip install pytest temporalio
로컬 Temporal 서버 실행 (선택 사항)
docker compose up -d
예제 테스트 코드
다음은 시간 스킵을 사용하는 단위 테스트 예제입니다:
import pytest
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
from your_workflow import YourWorkflow
from your_activities import your_activity
@pytest.mark.asyncio
async def test_your_workflow():
async with await WorkflowEnvironment(time_skipping=True) as env:
async with Worker(
env.client,
task_queue="your-task-queue",
workflows=[YourWorkflow],
activities=[your_activity],
):
result = await env.client.execute_workflow(
YourWorkflow.run,
"input_data",
id="test-workflow-id",
task_queue="your-task-queue",
)
assert result == "expected_output"