pytest/celery - 失败:不允许数据库访问

Question

我在让pytest测试django项目一部分的celery任务时遇到了一些麻烦。我能让它工作的唯一方法是在测试方法中导入任务。

# tasks.py
from __future__ import absolute_import
import logging
import django
django.setup() # flake8: noqa
from .celery import app
from .models import Run
from celery.exceptions import TaskError

@app.task(bind=True, throws=(TaskError,))
def get_runs_task(self, testrun_id=None):
    ...
    return True

---

工程：

# test_tasks.py
# flake8: noqa
from django.test import override_settings
import pytest

@override_settings(CELERY_ALWAYS_EAGER=True)
@pytest.mark.django_db
def test_get_runs_one():
    from ami.tasks import get_runs_task
    result = get_runs_task.delay(testrun_id='123456')
    assert result.successful()

---

显示Failed: Database access not allowed, use the "django_db" mark...错误：

# test_tasks.py
# flake8: noqa
from django.test import override_settings
import pytest
from ami.tasks import get_runs_task

@override_settings(CELERY_ALWAYS_EAGER=True)
@pytest.mark.django_db
def test_get_runs_one():
    result = get_runs_task.delay(testrun_id='123456')
    assert result.successful()

任何帮助将不胜感激，谢谢。

更新：（还有更多字段，但为了简洁起见，我省略了）

# models.py
from django.db import models

class Run(models.Model):
    run_id = models.CharField(
        max_length=15,
        verbose_name='Run ID')
# test_models.py
from ami.models import Run
from django.core.exceptions import ValidationError
import pytest
@pytest.fixture
def create_good_run():
    run = Run.objects.create(
        run_id='12345'
    return run
@pytest.mark.django_db
def test_inherit_values():
    run = create_good_run()
    assert run

我在tasks.py文件中import django和django.setup()，因为我正在django.core.exceptions.AppRegistryNotReady: Apps aren't loaded yet.我将它们移动到下面celery.py：

# celery.py
from __future__ import absolute_import
import os
import django
django.setup() # flake8: noqa
from celery import Celery
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'amiproj.settings')
from django.conf import settings  # flake8: noqa
app = Celery('ami')
app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
if __name__ == '__main__':
    app.start()

Answer 1

尝试改用pytest-django的settings装置：

import pytest
from ami.tasks import get_runs_task

@pytest.mark.django_db
def test_get_runs_one(settings):
    settings.CELERY_ALWAYS_EAGER = True
    result = get_runs_task.delay(testrun_id='123456')
    assert result.successful()

我敢打赌，关于 Django 的设置测试助手的某些内容无法与 pytest 的做事方式正常工作。

有关settings夹具的信息，请参阅pytest-django文档。