Интеграционные тесты django-channels с помощью pytest?

Я пытаюсь создать простой тестовый фикстур в стиле live-сервера для pytest для тестирования с django-каналами. Я пришел к самому простому решению:

import pytest


@pytest.fixture(scope="function")
def channels_live_server(request):
    from channels.testing import ChannelsLiveServerTestCase

    class MyChannelsLiveServerTestCase(ChannelsLiveServerTestCase):
        @property
        def url(self):
            return self.live_server_url

    server = MyChannelsLiveServerTestCase()
    server._pre_setup()

    yield server

    server._post_teardown()

При использовании веб-сервер запускается в фоновом режиме и обслуживает веб-страницы просто отлично... за исключением того, что он использует "производственные" настройки, что означает, что используемая база данных НЕ является тестовой базой данных.

Есть идеи, как это исправить? Я понимаю, что многопроцессорность на каком-то более позднем этапе использует Popen и запускает сервер, используя контекст spawn. Я на macOS, если это имеет значение.

В Django-chnanels 4.1 проблема с классами по умолчанию на моей установке заключалась в том, что имя тестовой базы данных не задавалось должным образом. Поэтому мне пришлось использовать этот патч, чтобы channels_live_server исправить это в моем коде.

Мое решение - простой и быстрый патч, потому что вся проблема выглядит как часть более серьезной проблемы, когда окружение родительского процесса на самом деле не копируется в окружение DaphneProcess через мультипроцессинг. Или, возможно, некоторые настройки Django опускаются в процессе.

Единственное, что делает этот код, это устанавливает правильное имя базы данных для тестирования, в том числе при использовании с pytest-xdist.

from functools import partial

import pytest
from channels.testing import ChannelsLiveServerTestCase
from daphne.testing import DaphneProcess
from django.db.backends.base.creation import TEST_DATABASE_PREFIX


class PatchedDaphneProcess(DaphneProcess):

    def __init__(self, *args, **kwargs):
        self.test_db_name = kwargs.pop("test_db_name")
        super().__init__(*args, **kwargs)

    def run(self):
        from django.conf import settings

        settings.DATABASES["default"]["NAME"] = self.test_db_name
        return super().run()


class PatchedLiveServerTestCase(ChannelsLiveServerTestCase):
    # ProtocolServerProcess = PatchedDaphneProcess

    def __init__(self, *args, **kw):
        self.test_db_name = kw.pop("test_db_name")
        super().__init__(*args, **kw)

    @property
    def ProtocolServerProcess(self):
        return partial(PatchedDaphneProcess, test_db_name=self.test_db_name)

    @property
    def url(self):
        return self.live_server_url


@pytest.fixture(scope="function")
def channels_live_server(request, transactional_db):
    def test_db_name():
        from django.conf import settings

        n = settings.DATABASES["default"].get("TEST", {}).get("NAME", None)
        if n:
            return n
        if settings.DATABASES["default"]["NAME"].startswith(TEST_DATABASE_PREFIX):
            return settings.DATABASES["default"]["NAME"]

        return TEST_DATABASE_PREFIX + settings.DATABASES["default"]["NAME"]

    server = PatchedLiveServerTestCase(test_db_name=test_db_name())
    server._pre_setup()

    yield server

    server._post_teardown()
Вернуться на верх