import stat
import sys
import threading
+import time
import unittest
from unittest import mock
import warnings
+from test import support
from test.support import os_helper
from test.support import socket_helper
from test.support import wait_process
parent_handled = manager.Event()
def child_main():
- signal.signal(signal.SIGTERM, lambda *args: child_handled.set())
+ def on_sigterm(*args):
+ child_handled.set()
+ sys.exit()
+
+ signal.signal(signal.SIGTERM, on_sigterm)
child_started.set()
+ while True:
+ time.sleep(1)
async def main():
loop = asyncio.get_running_loop()
process.start()
child_started.wait()
os.kill(process.pid, signal.SIGTERM)
- process.join()
+ process.join(timeout=support.SHORT_TIMEOUT)
async def func():
await asyncio.sleep(0.1)
asyncio.run(main())
+ child_handled.wait(timeout=support.SHORT_TIMEOUT)
self.assertFalse(parent_handled.is_set())
self.assertTrue(child_handled.is_set())