Skip to content
Merged
Changes from all commits
Commits
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
gh-95010: fix asyncio GenericWatcherTests.test_create_subprocess_fail…
…s_with_inactive_watcher

the test was never run, because it was missing the TestCase class
the test failed because the wrong attribute was patched
  • Loading branch information
graingert committed Jul 19, 2022
commit 20b9bd0c1a38d9d243eefdd947a3dfac6ef6f7c0
46 changes: 26 additions & 20 deletions Lib/test/test_asyncio/test_subprocess.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__':
Expand Down