ChannelsLiveServerTestCase equivalent for pytest

Question

In pytest-django there is a builtin fixture live_server though it seems like this server (that is actually based on LiveServerTestCase) can't handle web-sockets or at least won't interact with my asgi.py module.

How can one mimic that fixture in order to use ChannelsLiveServerTestCase instead? Or anything else that will run a test-database and will be able to serve an ASGI application?

My goal eventually is to have as close to production environment as possible, for testing and being able to test interaction between different Consumers.

P.S: I know I can run manage.py testserver <Fixture> on another thread / process by overriding django_db_setup though I seek for a better solution.

Answer 1

@aaron's solution can't work, due to pytest-django conservative approach for database access.

another proccess wouldn't be aware that your test has database access permissions therefore you won't have database access. (here is a POC)

Using a session scoped fixture of daphne Server was enough for me.

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