6

私は次の設定をしています：

  • Django-Celery プロジェクト A はタスクを登録しますfoo
  • プロジェクト B: Celery の send_task を使用して呼び出しますfoo
  • プロジェクト A とプロジェクト B の構成は同じです: SQS、シリアライズ用の msgpack、gzip など。
  • 各プロジェクトは異なる github リポジトリに存在します

Celery をまったく使用せずに、プロジェクト A で「foo」への呼び出しを単体テストしfoo(1,2,3)、結果をアサートしました。私はそれが機能することを知っています。

プロジェクト B の send_task が正しいパラメーターを送信することを単体テストしました。

私がテストしておらず、アドバイスが必要なのは、2 つのプロジェクト間の統合です。次のような単体テストが必要です。

  • プロジェクト A のコンテキストでワーカーを開始する
  • プロジェクト B のコードを使用してタスクを送信する
  • 最初のステップで開始されたワーカーが、2 番目のステップで送信したパラメーターを使用してタスクを取得し、foo関数が期待される結果を返したことをアサートします。

これは、python のサブプロセスを使ってワーカーの出力を解析することでハッキングできそうですが、それは醜いです。このような場合の単体テストの推奨されるアプローチは何ですか? 共有できるコード スニペットはありますか? ありがとう！

4

1 に答える 1

1

単体テストを使用してトランスポート メカニズム (つまり、セロリを介したタスク パラメーターの送信) を明示的にテストする価値があるかどうかはわかりません。個人的には、次のようにテストを記述します (複数の単体テストに分割できます)。

  • プロジェクト B のコードを使用して、サンプル パラメーターを含むタスクを生成します。
  • Celery で使用されるのと同じ方法を使用して、タスク パラメーターをエンコードします (つまり、パラメーターをピクルするか、JSON としてエンコードします)。
  • タスク パラメーターを再度デコードし、破損が発生していないことを確認します。
  • タスク関数を呼び出して、正しい結果が生成されることを確認します。
  • タスク関数の結果に対して同じエンコード/デコード シーケンスを実行します。

このメソッドを使用すると、それをテストできます

  • タスク生成は意図したとおりに機能します
  • タスクのパラメーターと結果のエンコードとデコードは期待どおりに機能します

必要に応じて、システム テストを使用して、輸送メカニズムの機能を個別にテストすることもできます。

于 2013-08-19T14:31:39.587 に答える