summaryrefslogtreecommitdiffstats
path: root/Lib/test/test_asyncio/test_subprocess.py
diff options
context:
space:
mode:
Diffstat (limited to 'Lib/test/test_asyncio/test_subprocess.py')
-rw-r--r--Lib/test/test_asyncio/test_subprocess.py46
1 files changed, 26 insertions, 20 deletions
diff --git a/Lib/test/test_asyncio/test_subprocess.py b/Lib/test/test_asyncio/test_subprocess.py
index 305330a..961c463 100644
--- a/Lib/test/test_asyncio/test_subprocess.py
+++ b/Lib/test/test_asyncio/test_subprocess.py
@@ -700,34 +700,40 @@ if sys.platform != 'win32':
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__':