temporal-python-testing

Temporal Python テスト戦略

pytest を使用した 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