Skip to content

Commit

Permalink
Add distinction between error cases in TimerContext
Browse files Browse the repository at this point in the history
  • Loading branch information
GDYendell committed Dec 12, 2024
1 parent 0070045 commit 5ad9083
Show file tree
Hide file tree
Showing 3 changed files with 36 additions and 2 deletions.
1 change: 1 addition & 0 deletions CONTRIBUTORS.txt
Original file line number Diff line number Diff line change
Expand Up @@ -140,6 +140,7 @@ Frederik Peter Aalund
Gabriel Tremblay
Gang Ji
Gary Wilson Jr.
Gary Yendell
Gennady Andreyev
Georges Dubus
Greg Holt
Expand Down
11 changes: 10 additions & 1 deletion aiohttp/helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -658,7 +658,16 @@ def assert_timeout(self) -> None:
def __enter__(self) -> BaseTimerContext:
task = asyncio.current_task(loop=self._loop)
if task is None:
raise RuntimeError("Timeout context manager should be used inside a task")
try:
asyncio.current_task()
except RuntimeError:
raise RuntimeError(
"Timeout context manager should be used inside a task"
)
else:
raise RuntimeError(
"Timeout context manager used in a task on a different event loop"
)

if sys.version_info >= (3, 11):
# Remember if the task was already cancelling
Expand Down
26 changes: 25 additions & 1 deletion tests/test_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import weakref
from math import ceil, modf
from pathlib import Path
from threading import Thread
from typing import Dict, Iterator, Optional, Union
from unittest import mock
from urllib.request import getproxies_environment
Expand Down Expand Up @@ -403,11 +404,34 @@ async def task_with_timeout() -> None:


def test_timer_context_no_task(loop: asyncio.AbstractEventLoop) -> None:
with pytest.raises(RuntimeError):
with pytest.raises(
RuntimeError, match="Timeout context manager should be used inside a task"
):
with helpers.TimerContext(loop):
pass


async def test_timer_context_wrong_event_loop() -> None:
loop = asyncio.get_running_loop()

async def use_timer(loop: asyncio.AbstractEventLoop = loop) -> None:
with pytest.raises(
RuntimeError,
match="Timeout context manager used in a task on a different event loop",
):
with helpers.TimerContext(loop):
pass

other_loop = asyncio.new_event_loop()
thread = Thread(target=other_loop.run_forever)
thread.start()

asyncio.run_coroutine_threadsafe(use_timer(), other_loop).result(timeout=5)

other_loop.stop()
thread.join()


async def test_weakref_handle(loop: asyncio.AbstractEventLoop) -> None:
cb = mock.Mock()
helpers.weakref_handle(cb, "test", 0.01, loop)
Expand Down

0 comments on commit 5ad9083

Please sign in to comment.