DjangoテストケースデータベースをCeleryに公開

Question

Djangoテストケースを実行すると、テストが完了するたびにデータベース書き込みがロールバックされるように、独立したテストデータベースが作成されます。私はCeleryとの統合テストを作成しようとしていますが、この一時的なテスト・データベースにCeleryをどのように接続するかを理解することはできません。素朴な設定では、Djangoに保存されたオブジェクトはCeleryには表示されず、Celeryに保存されたオブジェクトは無期限に保持されます。ここで

は、例えば、テストケースである：セロリは、そのDjangoのテストケースオブジェクトを見ることができるようにそれを作るにはどうすればよい

1. ：

   import json 
   from rest_framework.test import APITestCase 
   from myapp.models import MyModel 
   from myapp.util import get_result_from_response 
   
   class MyTestCase(APITestCase): 
       @classmethod 
       def setUpTestData(cls): 
        # This object is not visible to Celery 
        MyModel(id='test_object').save() 
   
       def test_celery_integration(self): 
        # This view spawns a Celery task 
        # Task should see MyModel.objects.get(id='test_object'), but can't 
        http_response = self.client.post('/', 'test_data', format='json') 
   
        result = get_result_from_response(http_response) 
        result.get() # Wait for task to finish before ending test case 
        # Objects saved by Celery task should be deleted, but persist 
   

   は、私は2つの質問がありますか？

2. セロリによって保存されたすべてのオブジェクトが、テストが完了したら自動的にロールバックされるようにするにはどうすればよいですか？

は、私がこれを自動的に行うことが不可能である場合は、手動でオブジェクトをクリーンアップするために喜んだが、それでも APISimpleTestCaseで tearDown内のオブジェクトの削除はロールバックされているようです。私はセロリの依存関係をスキップ推薦するあなたのユニットテストのために

Answer 1

これは、Djangoテストケース内でCeleryワーカーを開始することで可能です。

背景

Djangoのインメモリデータベースはsqlite3です。 the description page for Sqlite in-memory databasesに記載されているように、「インメモリデータベースを共有する[A] 11個のデータベース接続は同じプロセス内にある必要があります。これは、Djangoがメモリ内のテストデータベースを使用し、セロリが別のプロセスで開始されている限り、CeleryとDjangoにテストデータベースを共有させることは基本的に不可能であることを意味します。

ただし、celery.contrib.testing.worker.start_workerを使用すると、同じプロセス内の別のスレッドでCeleryワーカーを開始することができます。この作業者は、メモリ内のデータベースにアクセスできます。

これは、セロリが既にDjangoプロジェクトでthe usual wayに設定されていることを前提としています。

ソリューション

ジャンゴ - セロリは、いくつかのクロススレッド間通信を必要とするので、孤立したトランザクションで実行されていないだけでテストケースが動作します。テストケースは、SimpleTestCaseまたはその残りの同等物APISimpleTestCaseから直接継承し、クラス属性allow_database_queriesをTrueに設定する必要があります。

キーは、setUpClassメソッドTestCaseでセロリの作業を開始し、tearDownClassメソッドで閉じます。キーの機能は、おそらくmysite.celery.appから得られた、それぞれsetUpClass及びtearDownClassに呼び出されなければならない__enter__と__exit__方法を有するパイソンContextManagerを返し、現在のセロリアプリケーションのインスタンスを必要とする、celery.contrib.testing.worker.start_worker(app)あります。おそらく、手動で入力してデコレータなどでContextManagerを使用しないようにする方法がありますが、わかりませんでした。ここでは一例tests.pyファイルです：

from celery.contrib.testing.worker import start_worker 
from django.test import SimpleTestCase 

from mysite.celery import app 

class BatchSimulationTestCase(SimpleTestCase): 
    allow_database_queries = True 

    @classmethod 
    def setUpClass(cls): 
     super().setUpClass() 

     # Start up celery worker 
     cls.celery_worker = start_worker(app) 
     cls.celery_worker.__enter__() 

    @classmethod 
    def tearDownClass(cls): 
     super().tearDownClass() 

     # Close worker 
     cls.celery_worker.__exit__(None, None, None) 

    def test_my_function(self): 
     # my_task.delay() or something 

が何らかの理由で、検査員は、労働者に障害が発生した場合に、より良いエラーメッセージを提供するために、おそらく、'celery.ping'と呼ばれるタスクを使用しようとします。さらに、キーワード引数として〜Falseを設定しても、その存在の有無はまだ確認されています（start_worker）。探しているタスクはcelery.contrib.testing.tasks.pingです。ただし、このタスクはデフォルトではインストールされません。このタスクをcelery.contrib.testingにINSTALLED_APPSに追加してsettings.pyに追加することで可能です。しかし、これは作業者に見えるようにするだけです。作業者を生成するコードではありません。ワーカーを生成するコードはassert 'celery.ping' in app.tasksで失敗します。これをコメントアウトするとすべてが機能しますが、インストールされたライブラリを変更するのは良い解決策ではありません。私はおそらく何か間違ったことをやっているが、どこかでそのようなcelery.pyとして、app.autodiscover_tasks()によってピックアップすることができ、私は上の和解問題を回避するには、単純な関数をコピーしている：

今

@app.task(name='celery.ping') 
def ping(): 
    # type:() -> str 
    """Simple task that just returns 'pong'.""" 
    return 'pong' 

、テストが実行されたときに、ノーあり別のセロリプロセスを開始する必要があります。 CeleryワーカーはDjangoテストプロセスで別のスレッドとして開始されます。この作業者は、デフォルトのメモリ内テストデータベースを含む、メモリ内のデータベースをすべて見ることができます。労働者の数を制御するには、start_workerで利用可能なオプションがありますが、デフォルトは1人の労働者です。

Answer 2

、2次のリンクは、あなたのユニットテストを開始するにはnecesarryに関する情報を提供します：

• http://docs.celeryproject.org/projects/django-celery/en/2.4/cookbook/unit-testing.html
• http://docs.celeryproject.org/en/latest/userguide/testing.html

あなたの場合実際には、サーバー、ワーカー、キューの組み合わせでdockercomposeを適切に設定し、django-celeryドキュメントからカスタムCeleryTestRunnerを拡張するキューを含むセロリ関数呼び出しをテストしたいと考えています。しかし、テストシステムがプロダクションから代表的なものに遠く離れているため、私はそれから利益を得ることはできません。