When using AsyncTestCase, and a test times out, then the stacktrace printed to the console does not help pinpoint where the timeout occurs. This makes debugging tests very hard...
In the _callMaybeCoro method of DeferrableTestCase, I'd like to see some kind of handling of timeouts and hopefully print a useful stacktrace.
.
Something like this can work:
@classmethod
def _callMaybeCoro(cls, coro: Coroutine[Any, Any, Any]) -> Generator:
async def withTimeout() -> None:
task = asyncio.create_task(coro)
_, pending = await asyncio.wait({task}, timeout=cls.timeout_ms / 1000, return_when=asyncio.FIRST_COMPLETED)
if task in pending:
print("\n=== BEGIN: COROUTINE STACK BEFORE CANCELLATION ===")
task.print_stack()
print("=== END: COROUTINE STACK BEFORE CANCELLATION ===")
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
raise TimeoutError
await task
future = cls.run_coroutine(withTimeout())
class Signal:
def __init__(self) -> None:
self.done = False
self.exception: BaseException | None = None
def check(self) -> bool:
if self.exception:
raise self.exception
return self.done
signal = Signal()
def onDone(future: FutureLike) -> None:
if ex := future.exception():
signal.exception = ex
elif future.done():
signal.done = True
future.add_done_callback(onDone)
yield {"condition": signal.check, "timeout": cls.timeout_ms}
When using
AsyncTestCase, and a test times out, then the stacktrace printed to the console does not help pinpoint where the timeout occurs. This makes debugging tests very hard...In the _callMaybeCoro method of DeferrableTestCase, I'd like to see some kind of handling of timeouts and hopefully print a useful stacktrace.
.
Something like this can work: