2013-08-13 20 views
18

モデルがdjangoアプリケーションに保存されたときに実行されるセロリのタスクを修正するためにPythonモックライブラリを使用して、正しく呼び出されているかどうかを確認しようとしています。Djangoの単体テストでモックを使ってセロリのタスクを修正する

基本的には、タスクがmyapp.tasks内で定義されており、そのような私のmodels.pyファイルの先頭にインポートされます:

from .tasks import mytask

...そして使用してモデル内save()上で実行されますmytask.delay(foo, bar)。これまでのところ、Celerydなどを実際に実行しているときにはうまくいきます。

私はタスクを疑似する単体テストを構築したいのですが、正しい引数で呼び出されることを確認するだけです。実際にはこれまでセロリの仕事をしてみてください。

ので、テストファイルで、私は、標準的なテストケースのこの内部のようなものを持っている:

from mock import patch # at the top of the file 

# ...then later 
def test_celery_task(self): 
    with patch('myapp.models.mytask.delay') as mock_task: 
     # ...create an instance of the model and save it etc 
     self.assertTrue(mock_task.called) 

を...それが呼ばれることは決してありません/常にfalseです。私はいろいろな化身を試しました(代わりにmyapp.models.mytaskをパッチして、代わりにmock_task.delayが呼び出されたかどうかを確認しています)。私は偽のドキュメントからインポート経路が重要であると主張しました。 (私はそれを正しく理解すれば、myapp.models.mytask.delayいうよりmyapp.tasks.mytask.delayだろう)テスト下のモジュール。

私はここで間違っているつもりです

は?セロリのタスクをパッチでいくつかの特定の困難は、私のように使用されているcelery.taskを(パッチを適用できますか?ありますデコレータをmytaskに)代わりに?

+1

は、あなたが「=真CELERY_ALWAYS_EAGER」を設定する代わりに、それを嘲笑試してみましたか? – clsung

答えて

21

あなたが持っている問題は、これがセロリの仕事であるという事実とは関係ありません。 )

具体的には、あなたが該当する行は次のようになりますので、「MYTASK」をインポートして、そこの上にパッチを適用されて表示したり、他のファイルを見つける必要があります。

with patch('myapp.myview.mytask.delay') as mock_task: 

をするいくつかのより多くの味がありますここでは、この:

http://www.voidspace.org.uk/python/mock/patch.html#where-to-patch

+0

乾杯!私はまだ試してみましたが(プロジェクトは今休止中ですが)、すぐに試してみて、これに答えてマークします。あなたが提案しているテーマについてさまざまなバリエーションを試したことを思い出しているようですが、私の血糖値は当時低かったのです...-) – Emil

+1

実際には、 、質問コードに例示されているように...それを動作させることはできません。しかたがない。 – Emil

+0

質問はモデルにパッチを当てることです。モデルで "遅延"を使用していないと思われるので間違っていますが、おそらくビューであるため、私のパッチコード(上記)は少し異なります。 –

18

@taskデコレータはTaskオブジェクト(documentationを参照)機能を置き換えます。あなたが自分自身を模擬するならば、のオブジェクトをMagicMockに置き換え、タスクをまったくスケジュールしません。その代わり、Taskオブジェクトのrun()メソッドを模擬そうのような:

@override_settings(CELERY_ALWAYS_EAGER=True) 
@patch('monitor.tasks.monitor_user.run') 
def test_monitor_all(self, monitor_user): 
    """ 
    Test monitor.all task 
    """ 

    user = ApiUserFactory() 
    tasks.monitor_all.delay() 
    monitor_user.assert_called_once_with(user.key) 
+1

あなたは2つの質問で、この答え、言葉のための単語を投稿した理由はありますか? –

+0

これは、この質問だけでなく、他の質問についても有益な情報です。彼らはまったく同じ質問ではありません。 –

+0

BTW回答が2つの質問に関連して有用であると考えるなら、そのうちの1つを重複して報告することができ、誰かがそれを世話するでしょう。同じ答えを投稿することは間違ったことです。 –

関連する問題