Add a global timeout parameter to Communicators

[johnthagen]

Currently WebsocketCommunicator takes timeout parameters for each of its methods:

    async def connect(self, timeout=1):
        ...

Since the Communicator classes are designed for testing, and as such there may be many repeated usages, it would be nice if you could also control this timeout globally, so that it could be set in a single place and then applied to all instances.

This could look like:

communicator = WebsocketCommunicator(application, path, timeout=5.0)

As an example, HTTPX provides an API for this:

• https://www.python-httpx.org/advanced/#setting-a-default-timeout-on-a-client

This is also related to the single step debugability of unit tests, as mentioned in

• Control async timeout when unit testing and using interactive debugging #2019

[carltongibson]

Hi @johnthagen, thanks!

I'm not anti this. I wonder if @andrewgodwin would accept the same on the base ApplicationCommunicator class? 🤔

[johnthagen]

@carltongibson Oh, I didn't even realize that ApplicationCommunicator was all the way up in asgiref. 👀

In case this is useful to anyone else, I wrote a workaround using a subclass to use in the short term

class DefaultTimeoutWebsocketCommunicator(WebsocketCommunicator):
    """WebsocketCommunicator that provides a configurable default timeout."""

    timeout = 3.0
    """Default timeout to use when one is not supplied."""

    async def connect(self, timeout: float | None = None) -> tuple[bool, int | None]:
        match timeout:
            case float():
                return await super().connect(timeout=timeout)
            case None:
                return await super().connect(timeout=self.timeout)
            case _ as unreachable:
                assert_never(unreachable)

    async def receive_from(self, timeout: float | None = None) -> str | bytes:
        match timeout:
            case float():
                return await super().receive_from(timeout=timeout)
            case None:
                return await super().receive_from(timeout=self.timeout)
            case _ as unreachable:
                assert_never(unreachable)

    async def receive_json_from(self, timeout: float | None = None) -> dict[str, Any]:
        match timeout:
            case float():
                return await super().receive_json_from(timeout=timeout)
            case None:
                return await super().receive_json_from(timeout=self.timeout)
            case _ as unreachable:
                assert_never(unreachable)

    async def disconnect(self, code: int = 1000, timeout: float | None = None) -> None:
        match timeout:
            case float():
                await super().disconnect(code=code, timeout=timeout)
            case None:
                await super().disconnect(code=code, timeout=self.timeout)
            case _ as unreachable:
                assert_never(unreachable)

[johnthagen]

I also forgot to mention the motivation behind this request. We have observed in a production codebase that uses Django Channels that the default 1 second timeout can cause flaky timeout errors in our test suite. These are affected by the following real-world concerns, that increase the load/latency of the test suite and can cause operations to take longer than the timeout to complete:

• Running the test suite in parallel using pytest-xdist
• Running the test suite against an actual Redis backend for full end-to-end testing of the channels, rather than InMemoryChannelLayer
• Test suite running on shared CI servers that are loaded with other concurrent CI jobs
• Developers running the test suite on laptops with modest hardware

Taking these in combination, we've observed that being able to globally control the default timeout can create a more robust test suite and avoid a lot of DRY issues with seeing the timeout parameter dozens of places across many tests.