How can I patch a celery task

40 Views Asked by At

I'd like to know how I can write unit tests to test a celery task correctly. Unfortunately, the internet gives testing examples of very simple celery tasks.

My case looks as follows: I have a custom admin command which is a celery task that does 3 things:

  1. Makes a xlsx file from DB data,
  2. Uploads the xlsx file to s3,
  3. Send a link to the file to the request user (admin).

I'd like to test 2 assertions:

  1. to test the file itself (that the xlsx file has correct DB data). But I don't understand how I can read the file which is created in the celery task to compare what the file has against the data in the DB.
  2. to test that the link is sent to the admin correctly. (I can do it in my second attempt)

The code looks like this:

Admin site:

@admin.register(MyModal)
class MyModelAdmin(admin.ModelAdmin):
    def export_xlsx(self, request, queryset):
        my_celery_task.delay(ids, queryset, request.user)
    
    actions = [export_xlsx]

Celery task:

    @app.task(ignore_result=True)
    def my_celery_task(ids, queryset, user):
        # 1) makes a xlsx
        wb = openpyxl.Workbook()
        ws = wb.active
        # blablah
        wb.save(filename='/tmp/temp.xlsx')

        # 2) uploads to s3
        s3 = boto3.resource('s3', aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
                              aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
                              endpoint_url=settings.AWS_S3_ENDPOINT_URL)
        s3.Bucket(settings.AWS_STORAGE_BUCKET_NAME).upload_file('/tmp/temp.xlsx', 'file.xlsx')
    
        # 3) sends a link to the admin
        send_mail('title', 'message', settings.SERVER_EMAIL, [user.email], fail_silently=False)

Unit tests:

First attempt (with patch) - self.assertEqual(1, len(mail.outbox)) yields an error:

    @patch('my_celery_task.delay')
    def test_export_xlsx(self, mock_celery_task) -> None:
        data = {'action': 'export_xlsx', '_selected_action': [mydata]}
        url = '/admin/myurl'
        response = self.client.post(url, data, follow=True)

        self.assertEqual(200, response.status_code)
        self.assertEqual(1, len(mail.outbox))
        self.assertEqual(self.user.email, mail.outbox[0].to[0])
        self.assertEqual(settings.SENDER_EMAIL, mail.outbox[0].from_email)

Second attempt (with patch removed) - the tests run successfully:

    def test_export_xlsx(self) -> None:
        data = {'action': 'export_xlsx', '_selected_action': [mydata]}
        url = '/admin/myurl'
        response = self.client.post(url, data, follow=True)

        self.assertEqual(200, response.status_code)
        self.assertEqual(1, len(mail.outbox))
        self.assertEqual(self.user.email, mail.outbox[0].to[0])
        self.assertEqual(settings.SENDER_EMAIL, mail.outbox[0].from_email)

My questions are:

  1. How can I read the file which is created in the celery task to compare what the file has against the data in the DB?
  2. Why does my first attempt with patch fails on sending an email to the admin?
0

There are 0 best solutions below