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
46 changes: 26 additions & 20 deletions Lib/test/test_asyncio/test_subprocess.py
Original file line number Original file line Diff line number Diff line change
Expand Up @@ -700,34 +700,40 @@ class SubprocessPidfdWatcherTests(SubprocessWatcherMixin,
test_utils.TestCase): test_utils.TestCase):
Watcher = unix_events.PidfdChildWatcher 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(): with self.assertRaises(RuntimeError):
watcher = mock.create_authspec(asyncio.AbstractChildWatcher) await subprocess.create_subprocess_exec(
watcher.is_active.return_value = False os_helper.FakePath(sys.executable), '-c', 'pass')
asyncio.set_child_watcher(watcher)


with self.assertRaises(RuntimeError): watcher.add_child_handler.assert_not_called()
await subprocess.create_subprocess_exec(
os_helper.FakePath(sys.executable), '-c', 'pass')


watcher.add_child_handler.assert_not_called() with asyncio.Runner(loop_factory=asyncio.new_event_loop) as runner:

self.assertIsNone(runner.run(execute()))
self.assertIsNone(self.loop.run_until_complete(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__': if __name__ == '__main__':
Expand Down