Temporal 工作流和活动单元测试使用 WorkflowEnvironment 和 ActivityEnvironment 隔离测试单个工作流和活动的重点指南。带时间跳跃的 WorkflowEnvironment目的: 在隔离环境中测试工作流时间即时推进数月的长工作流 → 数秒基本设置模式importpytestfromtemporalio.testingimportWorkflowEnvironmentfromtemporalio.workerimportWorkerpytest.fixtureasyncdefworkflow_env():Reusable time-skipping test environmentenvawaitWorkflowEnvironment.start_time_skipping()yieldenvawaitenv.shutdown()pytest.mark.asyncioasyncdeftest_workflow_execution(workflow_env):Test workflow with time-skippingasyncwithWorker(workflow_env.client,task_queuetest-queue,workflows[YourWorkflow],activities[your_activity],):resultawaitworkflow_env.client.execute_workflow(YourWorkflow.run,test-input,idtest-wf-id,task_queuetest-queue,)assertresultexpected-output关键优势workflow.sleep(timedelta(days30))立即完成快速反馈循环毫秒 vs 小时确定性的测试执行时间跳跃示例睡眠推进pytest.mark.asyncioasyncdeftest_workflow_with_delays(workflow_env):Workflow sleeps are instant in time-skipping modeworkflow.defnclassDelayedWorkflow:workflow.runasyncdefrun(self)-str:awaitworkflow.sleep(timedelta(hours24))# Instant in testsreturncompletedasyncwithWorker(workflow_env.client,task_queuetest,workflows[DelayedWorkflow],):resultawaitworkflow_env.client.execute_workflow(DelayedWorkflow.run,iddelayed-wf,task_queuetest,)assertresultcompleted手动时间控制pytest.mark.asyncioasyncdeftest_workflow_manual_time(workflow_env):Manually advance time for precise controlhandleawaitworkflow_env.client.start_workflow(TimeBasedWorkflow.run,idtime-wf,task_queuetest,)# Advance time by specific amountawaitworkflow_env.sleep(timedelta(hours1))# Verify intermediate state via querystateawaithandle.query(TimeBasedWorkflow.get_state)assertstateprocessing# Advance to completionawaitworkflow_env.sleep(timedelta(hours23))resultawaithandle.result()assertresultcompleted测试工作流逻辑决策测试pytest.mark.asyncioasyncdeftest_workflow_branching(workflow_env):Test different execution pathsworkflow.defnclassConditionalWorkflow:workflow.runasyncdefrun(self,condition:bool)-str:ifcondition:returnpath-areturnpath-basyncwithWorker(workflow_env.client,task_queuetest,workflows[ConditionalWorkflow],):# Test true pathresult_aawaitworkflow_env.client.execute_workflow(ConditionalWorkflow.run,True,idcond-wf-true,task_queuetest,)assertresult_apath-a# Test false pathresult_bawaitworkflow_env.client.execute_workflow(ConditionalWorkflow.run,False,idcond-wf-false,task_queuetest,)assertresult_bpath-bActivityEnvironment 测试目的: 在没有工作流或 Temporal 服务器的情况下隔离测试活动基本活动测试fromtemporalio.testingimportActivityEnvironmentasyncdeftest_activity_basic():Test activity without workflow contextactivity.defnasyncdefprocess_data(input:str)-str:returninput.upper()envActivityEnvironment()resultawaitenv.run(process_data,test)assertresultTEST测试活动上下文心跳测试asyncdeftest_activity_heartbeat():Verify heartbeat callsactivity.defnasyncdeflong_running_activity(total_items:int)-int:foriinrange(total_items):activity.heartbeat(i)# Report progressawaitasyncio.sleep(0.1)returntotal_items envActivityEnvironment()resultawaitenv.run(long_running_activity,10)assertresult10取消测试asyncdeftest_activity_cancellation():Test activity cancellation handlingactivity.defnasyncdefcancellable_activity()-str:try:whileTrue:ifactivity.is_cancelled():returncancelledawaitasyncio.sleep(0.1)exceptasyncio.CancelledError:returncancelledenvActivityEnvironment(cancellation_reasontest-cancel)resultawaitenv.run(cancellable_activity)assertresultcancelled测试错误处理异常传播asyncdeftest_activity_error():Test activity error handlingactivity.defnasyncdeffailing_activity(should_fail:bool)-str:ifshould_fail:raiseApplicationError(Validation failed,non_retryableTrue)returnsuccessenvActivityEnvironment()# Test success pathresultawaitenv.run(failing_activity,False)assertresultsuccess# Test error pathwithpytest.raises(ApplicationError)asexc_info:awaitenv.run(failing_activity,True)assertValidation failedinstr(exc_info.value)pytest 集成模式共享夹具# conftest.pyimportpytestfromtemporalio.testingimportWorkflowEnvironmentpytest.fixture(scopemodule)asyncdefworkflow_env():Module-scoped environment (reused across tests)envawaitWorkflowEnvironment.start_time_skipping()yieldenvawaitenv.shutdown()pytest.fixturedefactivity_env():Function-scoped environment (fresh per test)returnActivityEnvironment()参数化测试pytest.mark.parametrize(input,expected,[(test,TEST),(hello,HELLO),(123,123),])asyncdeftest_activity_parameterized(activity_env,input,expected):Test multiple input scenariosresultawaitactivity_env.run(process_data,input)assertresultexpected最佳实践快速执行: 所有工作流测试都使用时间跳跃隔离: 分别测试工作流和活动共享夹具: 在相关测试之间重用 WorkflowEnvironment覆盖率目标: 工作流逻辑 ≥80%模拟活动: 使用 ActivityEnvironment 测试活动特定逻辑确定性: 确保测试结果在多次运行间一致错误场景: 同时测试成功和失败场景常见模式测试重试逻辑pytest.mark.asyncioasyncdeftest_workflow_with_retries(workflow_env):Test activity retry behaviorcall_count0activity.defnasyncdefflaky_activity()-str:nonlocalcall_count call_count1ifcall_count3:raiseException(Transient error)returnsuccessworkflow.defnclassRetryWorkflow:workflow.runasyncdefrun(self)-str:returnawaitworkflow.execute_activity(flaky_activity,start_to_close_timeouttimedelta(seconds10),retry_policyRetryPolicy(initial_intervaltimedelta(milliseconds1),maximum_attempts5,),)asyncwithWorker(workflow_env.client,task_queuetest,workflows[RetryWorkflow],activities[flaky_activity],):resultawaitworkflow_env.client.execute_workflow(RetryWorkflow.run,idretry-wf,task_queuetest,)assertresultsuccessassertcall_count3# Verify retry attempts其他资源Python SDK 测试: docs.temporal.io/develop/python/testing-suitepytest 文档: docs.pytest.orgTemporal 示例: github.com/temporalio/samples-python