django - 如何使用 Django 对一个简单的预定 celerybeat 健康检查任务进行单元测试？

Question

今天是个好日子。我希望你一切都好。我是 Django 新手，正在尝试学习 RESTful 开发和其他协同工作的工具（如 Celery）的基础知识。由于其非常完整的文档，我能够创建需要在后台运行的任务。然而，考虑到一些限制和大量数据，其中一些任务已经远远落后，工人往往会死亡。出于这个原因，我已经开始了一个故障排除过程，看看我的员工在他们停止响应之前走了多远。我创建了一个非常简单的任务，它只是试图每 2 分钟检查一次芹菜工人的健康状况，并将这些结果发送到我可以在我的网络服务器中看到的 django 模型：

from celery import app, shared_task
from ..models.recommendation import HealthCheck
import logging
from django.utils import timezone


@shared_task(bind=True)
def health_check_monitor(self):
    celery_worker = self.request.hostname
    try:
        logging.info('Celery is alive')
        HealthCheck.objects.update_or_create(
            alive=True, timestamp=timezone.now(),
            defaults={'worker_id': celery_worker})
    except TimeoutError:
        logging.info('Celery is dead')
        HealthCheck.objects.update_or_create(
            alive=False, timestamp=timezone.now(),
            defaults={'worker_id': celery_worker})

该HealthCheck模型将仅由一行组成，该行标识工作人员的姓名 ([e-mail])、运行的时间戳及其状态（死亡或活着）。该任务的时间表是在我的全局 django 设置中设置的，如下所示：

CELERY_BEAT_SCHEDULE = {
    'health_check_monitor': {
        'task': 'project.app.tasks.health_check_monitor.health_check_monitor',
        'schedule': crontab(),
         .
         .
         .

不管我的芹菜工人会发生什么，现在是时候对这些任务进行一些单元测试了。而且我不太擅长单元测试，所以我一直在拖延。这个决定让我很生气。即使为我的health_check_monitor任务进行单元测试，我也遇到了问题。棺材上的最后一颗钉子是，网络上确实没有太多关于如何对 SCHEDULED 任务进行单元测试的信息。我的测试如下所示：

class TestCeleryBeat(APITestCase):

    def test_celery_healthcheck(self):
        """Test that celery workers are functioning correctly for the given schedule"""
        from .tasks import health_check_monitor
        task = health_check_monitor.s().delay()
        result = task.get()
        self.assertEqual(task.status, 'SUCCESS')

但是由于某种原因，测试操作一直卡在上面两个小时。我不知道为什么会这样。对芹菜有更多专业知识的人可以帮我设计这个简单的测试吗？我什至不会展示我的其他任务，因为我确信在我掌握了这个窍门之后，我可以自己完成其他任务。

标签： djangounit-testingcelerydjango-celerycelerybeat