(task) add test for task API and confirm unkown task behaviour

According to Celery's author here:
https://github.com/celery/celery/issues/3596#issuecomment-262102185

There is no way to check if a task ID exists. It will just return
"PENDING" indefinitely. Add tests for this accordingly.

Closes #91
This commit is contained in:
Sylvain Zimmer
2025-05-13 23:14:57 +02:00
parent 3a3c79c645
commit dce87f864f
2 changed files with 24 additions and 7 deletions
+5 -5
View File
@@ -2,14 +2,13 @@
import logging
from celery.result import AsyncResult
from celery import states as celery_states
from celery.result import AsyncResult
from drf_spectacular.utils import (
OpenApiExample,
extend_schema,
inline_serializer,
)
from rest_framework import exceptions as drf_exceptions
from rest_framework import permissions
from rest_framework import serializers as drf_serializers
from rest_framework.response import Response
@@ -66,10 +65,11 @@ class TaskDetailView(APIView):
def get(self, request, task_id):
"""Get the status of a Celery task."""
# Check if the task exists
task_result = AsyncResult(task_id, app=celery_app)
if not task_result.id:
raise drf_exceptions.NotFound("Task not found")
# By default unknown tasks will be in PENDING. There is no reliable
# way to check if a task exists or not with Celery.
# https://github.com/celery/celery/issues/3596#issuecomment-262102185
# Prepare the response data
result_data = {
@@ -144,9 +144,26 @@ class TestApiDraftAndSendMessage:
assert send_response.status_code == status.HTTP_200_OK
mock_send_outbound_message.assert_called()
# Checks on the returned task_id
task_id = send_response.data["task_id"]
assert task_id is not None
# TODO: checks on returned task_id
# Check with an unknown task_id
task_response = client.get(
reverse("task-detail", kwargs={"task_id": "unknown-task-id"})
)
assert task_response.status_code == status.HTTP_200_OK
assert task_response.data["status"] == "PENDING"
# Call the task API
task_response = client.get(reverse("task-detail", kwargs={"task_id": task_id}))
assert task_response.status_code == status.HTTP_200_OK
assert task_response.data["status"] == "SUCCESS"
assert task_response.data["result"] is not None
assert task_response.data["result"]["message_id"] == draft_message_id
assert task_response.data["error"] is None
mock_send_outbound_message.assert_called()
sent_message = models.Message.objects.get(id=draft_message_id)
assert sent_message.raw_mime