temporal-python-testing

Temporal Python テスト戦略

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

このスキルを使用しない場合

• タスクが Temporal Python のテスト戦略に関係しない場合
• この範囲外の異なるドメインやツールが必要な場合

手順

• 目標、制約、必要な入力を明確にします。
• 関連するベストプラクティスを適用し、結果を検証します。
• 実行可能な手順と検証方法を提示します。

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

• ワークフローのユニットテスト - 時間スキップを使った高速なテスト
• 統合テスト - モック化したアクティビティを使ったワークフローのテスト
• リプレイテスト - 本番環境の履歴を使った決定性の検証
• ローカル開発 - 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 のセットアップと設定
• カバレッジツールとの統合
• 開発ワークフロー

実行時の前提条件

対象プロジェクトで互換性のある temporalio、pytest、pytest-asyncio のバージョンを固定し、asyncio_mode = auto を設定するか、pytest_asyncio.fixture を明示的に使用してください。YourWorkflow、アクティビティ、期待値、インポートを使ったスニペットは、適用時の参考例であり、実行可能なテストスイート一式ではありません。サンドボックス化されたワークフローとアクティビティは、インポート可能なモジュールのモジュールスコープで定義してください。スニペット内のネストされたクラスや関数はロジックの例示であり、サンドボックスの再インポート検証で失敗する場合があります。一意のワークフロー ID とタスクキューを使用し、実行時間に上限を設定してください。時間スキップ機能では、対応している環境でテストサーバーを起動またはダウンロードします。本番環境から分離して使用してください。

クイックスタートガイド

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

import pytest
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker

@pytest.fixture
async def workflow_env():
    env = await WorkflowEnvironment.start_time_skipping()
    try:
        yield env
    finally:
        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,
            "test-input", # or args=[arg1, arg2] for multiple arguments
            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 の公式なしきい値ではありません）：

• ワークフロー：プロジェクトで定義したしきい値（ここでは例として 80%）のロジックカバレッジ
• アクティビティ：プロジェクトで定義したしきい値（ここでは例として 80%）のロジックカバレッジ
• 統合：モック化したアクティビティを使った重要なパス
• リプレイ：デプロイ前にすべてのワークフローバージョンを検証

主なテスト原則

1. 時間のスキップ - タイマーをスキップしても、実際のアクティビティの I/O は高速化されません
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

制限事項

• タスクが上記の説明にある範囲に明確に一致する場合に限り、このスキルを使用してください。
• 出力を、環境固有の検証、テスト、または専門家によるレビューの代わりとして扱わないでください。
• 必要な入力、権限、安全上の境界、または成功基準が不足している場合は、作業を中断して確認を求めてください。