[Answered ]-Testing celery task

by

2👍

I found the solution for this. My task was a periodic task, testing in django creates a new test db during test and I found that mailer queue used to get stored in that database and celery was not able to pick up from that test database.So, i had to explicitly call the task from my test case, it looks something like this.

@override_settings(task_eager_propagates=True,task_always_eager=True,broker_url='memory://',backend='memory')
def test_core_student_put_api(self):
        response = response = self.c.post('http://test.localhost:8000/login/',{'email':'admin@dummy.com','password':'dummy'})
        response_add_student=self.c.post('http://test.localhost:8000/student_admin/',{'email':'xxx@dummy.com','stu_number':'100','role':'STUDENT'},**{'HTTP_AUTHORIZATION':'JWT '+response.data['token']})

        self.assertTrue(project_tasks.delay())
        self.assertEqual(response_put_employee.status_code, 200)

I had to over ride celery settings as suggested in celery docs for test configuration.
I have explicitly called project_tasks.delay() with assertTrue. my task contains a simple send_mail() command. It is fetching right data from my signals as well.invite_stu signal calls send_html_mail() which adds to to the mail queue in test db and when i explicitly call my periodic task(which is a periodic task) it sends the mail.I had read somewhere that celery cant pick up periodic tasks for test cases because of temporary database issue, should do some more research on that, but for present this fixes my issue.

👤Anup Bengeri

Leave a comment