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:
- Makes a xlsx file from DB data,
- Uploads the xlsx file to s3,
- Send a link to the file to the request user (admin).
I'd like to test 2 assertions:
- 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.
- 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:
- 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?
- Why does my first attempt with patch fails on sending an email to the admin?