testing - 测试 django celery，与数据库交互？

Question

通常，当使用纯 Django 时，我得到了一个清晰的测试数据库，当我在测试函数中创建一个对象时，我在项目 dode 中也得到了相同的对象。但是，当我使用 celery 时，项目中的数据库与生产中的数据库相同，而不是获取测试数据库。

• 1. 当我进入管理页面并创建一个新的时PeriodicTask，我在运行测试而不是清除数据库进行测试时也得到了该任务。
• 注意：所有型号都会发生这种情况，不仅PeriodicTask在我测试芹菜时

@app.task
def create_alert(beat_id, perams):
    return PeriodicTask.objects.count()

    def test_celery_beats_data_base(self):
        schedule, created = IntervalSchedule.objects.get_or_create(
            every=10,
            period=IntervalSchedule.SECONDS)

        PeriodicTask.objects.create(
            interval=schedule,
            name='Importing contacts',
            task='proj.tasks.import_contacts',
            one_off=True
            # args=[]
            # kwargs={}
        )
        beat_id = PeriodicTask.objects.first().id
        result = create_alert.delay(beat_id, 'perams')
        ic(result.get())  #1.  this return 0 # the probleme is this should return 1 as wel
        ic(PeriodicTask.objects.count())  #2. this return 1

标签： testingcelerypytestcelerybeat