Skip to content

gh-129874: improve test_events to use correct task implementation #129891

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Feb 9, 2025
Merged

gh-129874: improve test_events to use correct task implementation #129891

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 14 additions & 1 deletion Lib/test/test_asyncio/test_events.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2930,11 +2930,16 @@ class GetEventLoopTestsMixin:
get_running_loop_impl = None
get_event_loop_impl = None

Task = None
Future = None

def setUp(self):
self._get_running_loop_saved = events._get_running_loop
self._set_running_loop_saved = events._set_running_loop
self.get_running_loop_saved = events.get_running_loop
self.get_event_loop_saved = events.get_event_loop
self._Task_saved = asyncio.Task
self._Future_saved = asyncio.Future

events._get_running_loop = type(self)._get_running_loop_impl
events._set_running_loop = type(self)._set_running_loop_impl
Expand All @@ -2946,6 +2951,8 @@ def setUp(self):
asyncio.get_running_loop = type(self).get_running_loop_impl
asyncio.get_event_loop = type(self).get_event_loop_impl

asyncio.Task = asyncio.tasks.Task = type(self).Task
asyncio.Future = asyncio.futures.Future = type(self).Future
super().setUp()

self.loop = asyncio.new_event_loop()
Expand All @@ -2968,8 +2975,10 @@ def tearDown(self):
asyncio.get_running_loop = self.get_running_loop_saved
asyncio.get_event_loop = self.get_event_loop_saved

if sys.platform != 'win32':
asyncio.Task = asyncio.tasks.Task = self._Task_saved
asyncio.Future = asyncio.futures.Future = self._Future_saved

if sys.platform != 'win32':
def test_get_event_loop_new_process(self):
# bpo-32126: The multiprocessing module used by
# ProcessPoolExecutor is not functional when the
Expand Down Expand Up @@ -3088,6 +3097,8 @@ class TestPyGetEventLoop(GetEventLoopTestsMixin, unittest.TestCase):
get_running_loop_impl = events._py_get_running_loop
get_event_loop_impl = events._py_get_event_loop

Task = asyncio.tasks._PyTask
Future = asyncio.futures._PyFuture

try:
import _asyncio # NoQA
Expand All @@ -3102,6 +3113,8 @@ class TestCGetEventLoop(GetEventLoopTestsMixin, unittest.TestCase):
get_running_loop_impl = events._c_get_running_loop
get_event_loop_impl = events._c_get_event_loop

Task = asyncio.tasks._CTask
Future = asyncio.futures._CFuture

class TestServer(unittest.TestCase):

Expand Down
Loading