From 834bd5dd766cf212fb20d65d8a046c62a33006d4 Mon Sep 17 00:00:00 2001
From: Thomas Grainger <tagrain@gmail.com>
Date: Thu, 21 Jul 2022 12:35:20 +0100
Subject: [PATCH] gh-95010: Fix asyncio
GenericWatcherTests.test_create_subprocess_fails_with_inactive_watcher
(GH-95009)
The test was never run, because it was missing the TestCase class.
The test failed because the wrong attribute was patched.
---
Lib/test/test_asyncio/test_subprocess.py | 46 +++++++++++++-----------
1 file changed, 26 insertions(+), 20 deletions(-)
diff --git a/Lib/test/test_asyncio/test_subprocess.py b/Lib/test/test_asyncio/test_subprocess.py
index 305330a6918..961c463f8dc 100644
--- a/Lib/test/test_asyncio/test_subprocess.py
+++ b/Lib/test/test_asyncio/test_subprocess.py
@@ -700,34 +700,40 @@ class SubprocessPidfdWatcherTests(SubprocessWatcherMixin,
test_utils.TestCase):
Watcher = unix_events.PidfdChildWatcher
-else:
- # Windows
- class SubprocessProactorTests(SubprocessMixin, test_utils.TestCase):
-
- def setUp(self):
- super().setUp()
- self.loop = asyncio.ProactorEventLoop()
- self.set_event_loop(self.loop)
+ class GenericWatcherTests(test_utils.TestCase):
-class GenericWatcherTests:
+ def test_create_subprocess_fails_with_inactive_watcher(self):
+ watcher = mock.create_autospec(
+ asyncio.AbstractChildWatcher,
+ **{"__enter__.return_value.is_active.return_value": False}
+ )
- def test_create_subprocess_fails_with_inactive_watcher(self):
+ async def execute():
+ asyncio.set_child_watcher(watcher)
- async def execute():
- watcher = mock.create_authspec(asyncio.AbstractChildWatcher)
- watcher.is_active.return_value = False
- asyncio.set_child_watcher(watcher)
+ with self.assertRaises(RuntimeError):
+ await subprocess.create_subprocess_exec(
+ os_helper.FakePath(sys.executable), '-c', 'pass')
- with self.assertRaises(RuntimeError):
- await subprocess.create_subprocess_exec(
- os_helper.FakePath(sys.executable), '-c', 'pass')
+ watcher.add_child_handler.assert_not_called()
- watcher.add_child_handler.assert_not_called()
-
- self.assertIsNone(self.loop.run_until_complete(execute()))
+ with asyncio.Runner(loop_factory=asyncio.new_event_loop) as runner:
+ self.assertIsNone(runner.run(execute()))
+ self.assertListEqual(watcher.mock_calls, [
+ mock.call.__enter__(),
+ mock.call.__enter__().is_active(),
+ mock.call.__exit__(RuntimeError, mock.ANY, mock.ANY),
+ ])
+else:
+ # Windows
+ class SubprocessProactorTests(SubprocessMixin, test_utils.TestCase):
+ def setUp(self):
+ super().setUp()
+ self.loop = asyncio.ProactorEventLoop()
+ self.set_event_loop(self.loop)
if __name__ == '__main__':
--
GitLab