Django 테스트 DB를 사용하기 위해 샐러리 가져 오기

Question

장고 앱의 일부 셀러리 작업에 대한 단위 테스트를 작성하려고합니다. 이러한 작업은 모델 ID를 인수로 사용하고, 일부 작업을 수행하고, 모델을 업데이트합니다. devserver 및 셀러리 작업자를 실행할 때 모든 것이 잘 작동하지만 테스트를 실행할 때 샐러리 작업은 테스트 실행의 일부로 생성되고 파괴되는 django 테스트 db를 사용하지 않는다는 것이 분명 해졌다. 질문 : 셀러리가 나머지 테스트와 동일한 임시 DB를 사용하려면 어떻게해야합니까?

위에서 볼 수 있듯이 비슷한 문제에 대한 모든 대답에서 제안 된 설정 우선 적용을 사용하고 있습니다.

업데이트 : 개체 id를 작업에 전달하는 대신 작업을 db에 전달하는 대신 개체 자체를 작업에 전달하면 테스트가 제대로 작동 함을 나타냅니다. 작업. 그래서 적어도 지금은 그것이 나의 수정 일 것입니다. 내 테스트에서

:

class JobTest(TestCase): 

    @override_settings(CELERY_ALWAYS_EAGER=True, 
         CELERY_EAGER_PROPAGATES_EXCEPTIONS=True, 
         BROKER_BACKEND='memory') 
    def test_Job_Complete(self): 
     job = models.Job() 
     job.save() 
     tasks.do_a_thing(job.id) 
     self.assertTrue(job.complete) 

내 작업에서는 :

는

@celery.task 
def do_a_thing(job_id): 
    job = models.Job.objects.get(pk=job_id) 
    bunch_of_things(job) 
    job.complete = True 
    job.save() 

Answer 1

코드에 명백한 문제가 없습니다. 셀러리 작업자는 실행할 필요가 없습니다. 이 설정을 사용하면 celery가 작업을 동 기적으로 실행하고 실제로 메시지 대기열에 아무 것도 보내지 않습니다.

어쨌든 라이브 셀러리 작업자와 테스트를 쉽게 실행할 수 없습니다. 각 테스트는 트랜잭션으로 래핑되므로 동일한 데이터베이스에 연결되어 있어도 트랜잭션이 항상 테스트에 의해 롤백됩니다 근로자가 결코 이용할 수 없게된다.

정말로 이것을 수행해야하는 경우 this stackoverflow answer을 확인하십시오.

Answer 2

비슷한 문제가 발생했습니다. 다음 해결책은 은 깨끗하지 않습니다하지만 작동합니다.

1. 메인 에서 상속 된 별도의 장고 설정 파일을 만듭니다. integration_testing.py이라고합시다.

2. 파일은 다음과 같아야합니다

   #!/usr/bin/env bash

   :
   from .settings import *

   DATABASES = { 'default': { 'ENGINE': '<your engine>', 'NAME': 'test_<your database name>', 'USER': <your db user>, 'PASSWORD': <your db password>, 'HOST': <your hostname>, 'PORT': <your port number>, }

3. 사용자 환경을 설정하고 에게 셀러리 노동자를 시작하는 쉘 스크립트를 작성 export DJANGO_SETTINGS_MODULE="YOURPROJECTNAME.settings.integration_testing"

   celery purge -A YOURPROJECTNAME -f && celery worker -A YOURPROJECTNAME -l debug

4. 이 방식으로 셀러리를 구성한 경우에는, 위의 작품 :

   app = Celery('YOURPROJECTNAME')

   app.config_from_object('django.conf:settings', namespace='CELERY')

5. 실행 백그라운드에서 스크립트를 실행합니다.

6. 은 (장고 - 나머지 프레임 워크 또는 APITransactionTestCase) 셀러리 TransactionTestCase에서 상속을 포함 모든 테스트

7. 셀러리를 사용하는 단위 테스트를 실행하십시오. 셀러리 작업은 이제 테스트 DB를 사용하게됩니다. 그리고 최고의 희망.

Answer 3

한 가지 방법은 셀러리 노동자가 테스트 spawn the Celery worker inside the test itself에 같이 동일한 테스트 데이터베이스를 사용하도록 구성되어 있음을 보장합니다. 이는 TestCase의 setUpClass 방법에서 celery.contrib.testing.worker.start_worker을 사용하여 수행 할 수 있습니다.

또한 TestCase 대신 Django에서 SimpleTestCase 또는 나머지에서 APISimpleTestCase을 사용해야하므로 Celery 스레드와 테스트 스레드가 서로 테스트 데이터베이스를 변경하는 것을 볼 수 있습니다. 변경 사항은 테스트가 끝날 때까지 계속 파괴되지만, tearDown 메서드에서 수동으로 파괴하지 않으면 테스트간에 파괴되지 않습니다.