Open johnthagen opened 11 months ago
Hi @johnthagen, thanks!
I'm not anti this. I wonder if @andrewgodwin would accept the same on the base ApplicationCommunicator
class? 🤔
@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)
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:
pytest-xdist
InMemoryChannelLayer
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.
Currently
WebsocketCommunicator
takestimeout
parameters for each of its methods: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 thistimeout
globally, so that it could be set in a single place and then applied to all instances.This could look like:
As an example, HTTPX provides an API for this:
This is also related to the single step debugability of unit tests, as mentioned in
2019