| 25 | |
| 26 | |
| 27 | def _test_signals_async( |
| 28 | browser_name: str, launch_arguments: Dict, wait_queue: "multiprocessing.Queue[str]" |
| 29 | ) -> None: |
| 30 | # On Windows, hint to mypy and pyright that they shouldn't check this function |
| 31 | if sys.platform == "win32": |
| 32 | return |
| 33 | |
| 34 | os.setpgrp() |
| 35 | sigint_received = False |
| 36 | |
| 37 | def my_sig_handler(signum: int, frame: Any) -> None: |
| 38 | nonlocal sigint_received |
| 39 | sigint_received = True |
| 40 | |
| 41 | signal.signal(signal.SIGINT, my_sig_handler) |
| 42 | |
| 43 | async def main() -> None: |
| 44 | playwright = await async_playwright().start() |
| 45 | browser = await playwright[browser_name].launch( |
| 46 | **launch_arguments, |
| 47 | handle_sigint=False, |
| 48 | ) |
| 49 | context = await browser.new_context() |
| 50 | page = await context.new_page() |
| 51 | notified = False |
| 52 | try: |
| 53 | while not sigint_received: |
| 54 | if not notified: |
| 55 | wait_queue.put("ready") |
| 56 | notified = True |
| 57 | await page.wait_for_timeout(100) |
| 58 | finally: |
| 59 | wait_queue.put("close context") |
| 60 | await context.close() |
| 61 | wait_queue.put("close browser") |
| 62 | await browser.close() |
| 63 | wait_queue.put("close playwright") |
| 64 | await playwright.stop() |
| 65 | wait_queue.put("all done") |
| 66 | |
| 67 | asyncio.run(main()) |
| 68 | |
| 69 | |
| 70 | def _test_signals_sync( |