用途:为 Temporal Python 工作流提供系统化测试方案,涵盖单元测试、集成测试、重放测试、本地开发与 CI/CD 集成,并给出覆盖率策略。
主要能力:使用 pytest 与异步 fixture;通过 WorkflowEnvironment.start_time_skipping() 让长达数月的工作流在秒级完成测试;用 ActivityEnvironment 单独测试活动;用模拟活动隔离工作流逻辑;通过重放生产历史验证确定性;支持信号与查询测试、错误注入、多活动工作流测试、版本兼容性测试。
使用方式:技能按需加载资源文件——单元测试加载 resources/unit-testing.md,集成测试加载 resources/integration-testing.md,重放测试加载 resources/replay-testing.md,本地环境加载 resources/local-setup.md。基本工作流测试示例: import pytest from temporalio.testing import WorkflowEnvironment from temporalio.worker import Worker
@pytest.fixture async def workflow_env(): env = await WorkflowEnvironment.start_time_skipping() yield env await env.shutdown()
@pytest.mark.asyncio async def test_workflow(workflow_env): async with Worker( workflow_env.client, task_queue="test-queue", workflows=[YourWorkflow], activities=[your_activity], ): result = await workflow_env.client.execute_workflow( YourWorkflow.run, args, id="test-wf-id", task_queue="test-queue", ) assert result == expected
活动测试示例: from temporalio.testing import ActivityEnvironment
async def test_activity(): env = ActivityEnvironment() result = await env.run(your_activity, "test-input") assert result == expected_output
适用场景:工作流与活动的单元测试、带模拟依赖的集成测试、部署前确定性验证、本地 Temporal 服务与 pytest 配置、自动化测试流水线。
限制:资料未提供具体安装命令、依赖版本、Docker Compose 配置细节及完整资源文件内容;覆盖率目标(工作流与活动各 ≥80%)为推荐值,非强制保证。