ChannelsLiveServerTestCase 相当于 pytest

Question

在 pytest-django 中，有一个内置装置

live_server

，尽管这个服务器（实际上基于

LiveServerTestCase

）无法处理网络套接字，或者至少不会与我的

 交互asgi.py

模块。

如何模仿该装置才能使用

ChannelsLiveServerTestCase

来代替？或者其他任何可以运行测试数据库并且能够为 ASGI 应用程序提供服务的东西？

我的最终目标是拥有尽可能接近生产环境，用于测试并能够测试不同消费者之间的交互。

P.S：我知道我可以通过覆盖

manage.py testserver <Fixture>

 在另一个线程/进程上运行 

django_db_setup，尽管我寻求更好的解决方案。

Answer 1

您可以基于以下实现来实现

channels_live_server

夹具：

• live_server

  夹具，实例化

  • LiveServer

    助手，启动

    LiveServerThread

    和

• ChannelsLiveServerTestCase

  ，开始

  DaphneProcess

  。

@medihack在https://github.com/pytest-dev/pytest-django/issues/1027:

实现了它

from functools import partial
from channels.routing import get_default_application
from daphne.testing import DaphneProcess
from django.contrib.staticfiles.handlers import ASGIStaticFilesHandler
from django.core.exceptions import ImproperlyConfigured
from django.db import connections
from django.test.utils import modify_settings


def make_application(*, static_wrapper):
    # Module-level function for pickle-ability
    application = get_default_application()
    if static_wrapper is not None:
        application = static_wrapper(application)
    return application


class ChannelsLiveServer:
    host = "localhost"
    ProtocolServerProcess = DaphneProcess
    static_wrapper = ASGIStaticFilesHandler
    serve_static = True

    def __init__(self) -> None:
        for connection in connections.all():
            if connection.vendor == "sqlite" and connection.is_in_memory_db():
                raise ImproperlyConfigured(
                    "ChannelsLiveServer can not be used with in memory databases"
                )

        self._live_server_modified_settings = modify_settings(ALLOWED_HOSTS={"append": self.host})
        self._live_server_modified_settings.enable()

        get_application = partial(
            make_application,
            static_wrapper=self.static_wrapper if self.serve_static else None,
        )

        self._server_process = self.ProtocolServerProcess(self.host, get_application)
        self._server_process.start()
        self._server_process.ready.wait()
        self._port = self._server_process.port.value

    def stop(self) -> None:
        self._server_process.terminate()
        self._server_process.join()
        self._live_server_modified_settings.disable()

    @property
    def url(self) -> str:
        return f"http://{self.host}:{self._port}"


@pytest.fixture
def channels_live_server(request):
    server = ChannelsLiveServer()
    request.addfinalizer(server.stop)
    return server

Answer 2

@aaron 的解决方案无法工作，因为 pytest-django 保守的数据库访问方法。

另一个进程不会知道您的测试具有数据库访问权限，因此您将没有数据库访问权限。 （这是一个 POC）

使用 daphne 的会话范围固定装置

Server

对我来说就足够了。

import threading
import time

from functools import partial

from django.contrib.staticfiles.handlers import ASGIStaticFilesHandler
from django.core.exceptions import ImproperlyConfigured
from django.db import connections
from django.test.utils import modify_settings

from daphne.server import Server as DaphneServer
from daphne.endpoints import build_endpoint_description_strings


def get_open_port() -> int:
    import socket
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.bind(("", 0))
    s.listen(1)
    port = s.getsockname()[1]
    s.close()
    return port


def make_application(*, static_wrapper):
    # Module-level function for pickle-ability
    if static_wrapper is not None:
        application = static_wrapper(your_asgi_app)
    return application


class ChannelsLiveServer:
    port = get_open_port()
    host = "localhost"
    static_wrapper = ASGIStaticFilesHandler
    serve_static = True

    def __init__(self) -> None:
        for connection in connections.all():
            if connection.vendor == "sqlite" and connection.is_in_memory_db():
                raise ImproperlyConfigured(
                    "ChannelsLiveServer can not be used with in memory databases"
                )

        self._live_server_modified_settings = modify_settings(ALLOWED_HOSTS={"append": self.host})
        self._live_server_modified_settings.enable()

        get_application = partial(
            make_application,
            static_wrapper=self.static_wrapper if self.serve_static else None,
        )
        endpoints = build_endpoint_description_strings(
            host=self.host, port=self.port
        )

        self._server = DaphneServer(
            application=get_application(),
            endpoints=endpoints
        )
        t = threading.Thread(target=self._server.run)
        t.start()
        for i in range(10):
            time.sleep(0.10)
            if self._server.listening_addresses:
                break
        assert self._server.listening_addresses[0]

    def stop(self) -> None:
        self._server.stop()
        self._live_server_modified_settings.disable()

    @property
    def url(self) -> str:
        return f"ws://{self.host}:{self.port}"

    @property
    def http_url(self):
        return f"http://{self.host}:{self.port}"

@pytest.fixture(scope='session')
def channels_live_server(request, live_server):
    server = ChannelsLiveServer()
    request.addfinalizer(server.stop)
    return server