temporal-python-testing

Temporal Python のテスト戦略

Temporal ワークフローの包括的なテスト手法です。特定のテストシナリオに応じて段階的に情報を開示するリソースを提供します。

このスキルを使用する場面

• ワークフローの単体テスト - 時間スキップを使った高速なテスト
• 統合テスト - モックしたアクティビティを使ったワークフローのテスト
• リプレイテスト - 本番環境の履歴を使って決定性を検証
• ローカル開発 - Temporal サーバーと pytest をセットアップ
• CI/CD との統合 - テストパイプラインを自動化
• カバレッジ戦略 - ≥80% のテストカバレッジを達成

テストの基本方針

推奨されるアプローチ（出典：docs.temporal.io/develop/python/testing-suite）：

• テストの大半を統合テストとして作成する
• 非同期フィクスチャを使って pytest を使用する
• 時間スキップによって迅速なフィードバックを得る（月単位のワークフローを数秒でテスト）
• アクティビティをモックしてワークフローのロジックを分離する
• リプレイテストで決定性を検証する

3 種類のテスト：

1. 単体テスト：時間スキップを使ったワークフロー、ActivityEnvironment を使ったアクティビティ
2. 統合テスト：モックしたアクティビティを使ったワーカー
3. エンドツーエンドテスト：実際のアクティビティを使った完全な Temporal サーバーのテスト（必要な場合にのみ使用）

利用可能なリソース

このスキルでは、段階的な情報開示を通じて詳細なガイダンスを提供します。テストのニーズに応じて、特定のリソースを読み込んでください。

単体テストのリソース

ファイル：resources/unit-testing.md 読み込むタイミング：個々のワークフローまたはアクティビティを分離してテストする場合 内容：

• 時間スキップを使った WorkflowEnvironment
• アクティビティテスト用の ActivityEnvironment
• 長時間実行されるワークフローの高速実行
• 手動で時間を進めるパターン
• pytest のフィクスチャとパターン

統合テストのリソース

ファイル：resources/integration-testing.md 読み込むタイミング：外部依存関係をモックしたワークフローをテストする場合 内容：

• アクティビティのモック戦略
• エラー注入のパターン
• 複数のアクティビティを含むワークフローのテスト
• シグナルとクエリのテスト
• カバレッジ戦略

リプレイテストのリソース

ファイル：resources/replay-testing.md 読み込むタイミング：決定性を検証する場合、またはワークフローの変更をデプロイする場合 内容：

• 決定性の検証
• 本番環境の履歴のリプレイ
• CI/CD との統合パターン
• バージョン互換性テスト

ローカル開発のリソース

ファイル：resources/local-setup.md 読み込むタイミング：開発環境をセットアップする場合 内容：

• Docker Compose の設定
• pytest のセットアップと設定
• カバレッジツールとの統合
• 開発ワークフロー

クイックスタートガイド

基本的なワークフローテスト

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

カバレッジ目標

推奨されるカバレッジ（出典：docs.temporal.io のベストプラクティス）：

• ワークフロー：ロジックカバレッジ ≥80%
• アクティビティ：ロジックカバレッジ ≥80%
• 統合テスト：モックしたアクティビティを使って重要なパスをテスト
• リプレイ：デプロイ前にすべてのワークフローバージョンをテスト

テストの主な原則

1. 時間スキップ - 数か月にわたるワークフローを数秒でテスト
2. アクティビティのモック - 外部依存関係からワークフローのロジックを分離
3. リプレイテスト - デプロイ前に決定性を検証
4. 高いカバレッジ - 本番環境のワークフローでは ≥80% を目標にする
5. 迅速なフィードバック - 単体テストはミリ秒単位で実行

リソースの使用方法

必要に応じて特定のリソースを読み込んでください：

• 「単体テストのパターンを見せてください」→ resources/unit-testing.md を読み込む
• 「アクティビティをモックするにはどうすればよいですか？」→ resources/integration-testing.md を読み込む
• 「ローカルの Temporal サーバーをセットアップする」→ resources/local-setup.md を読み込む
• 「決定性を検証する」→ resources/replay-testing.md を読み込む

参考資料

• Python SDK のテスト：docs.temporal.io/develop/python/testing-suite
• テストパターン：github.com/temporalio/temporal/blob/main/docs/development/testing.md
• Python サンプル：github.com/temporalio/samples-python