Test interrupting a Server that is run explicitly inside asyncio
(
exception_signal: signal.Signals,
capture_signal: Callable[[signal.Signals], AbstractContextManager[None]],
unused_tcp_port: int,
)
| 67 | @pytest.mark.parametrize("exception_signal", signals) |
| 68 | @pytest.mark.parametrize("capture_signal", signal_captures) |
| 69 | async def test_server_interrupt( |
| 70 | exception_signal: signal.Signals, |
| 71 | capture_signal: Callable[[signal.Signals], AbstractContextManager[None]], |
| 72 | unused_tcp_port: int, |
| 73 | ): # pragma: py-win32 |
| 74 | """Test interrupting a Server that is run explicitly inside asyncio""" |
| 75 | |
| 76 | async def interrupt_running(srv: Server): |
| 77 | while not srv.started: |
| 78 | await asyncio.sleep(0.01) |
| 79 | signal.raise_signal(exception_signal) |
| 80 | |
| 81 | server = Server(Config(app=dummy_app, loop="asyncio", port=unused_tcp_port)) |
| 82 | asyncio.create_task(interrupt_running(server)) |
| 83 | with capture_signal(exception_signal) as witness: |
| 84 | await server.serve() |
| 85 | assert witness |
| 86 | # set by the server's graceful exit handler |
| 87 | assert server.should_exit |
| 88 | |
| 89 | |
| 90 | async def test_shutdown_on_early_exit_during_startup(unused_tcp_port: int): |
nothing calls this directly
no test coverage detected