使用数据库交互测试 django celery？

问题描述

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

• 在 1. ? 中，当我转到管理页面并创建新的 PeriodicTask 时，我在运行测试时也完成了该任务，而不是清除数据库进行测试。
• 注意：不仅在我测试 celery 时使用 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

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）