()
| 594 | def test_close_kill_running(self): |
| 595 | |
| 596 | async def kill_running(): |
| 597 | create = self.loop.subprocess_exec(asyncio.SubprocessProtocol, |
| 598 | *PROGRAM_BLOCKED) |
| 599 | transport, protocol = await create |
| 600 | |
| 601 | kill_called = False |
| 602 | def kill(): |
| 603 | nonlocal kill_called |
| 604 | kill_called = True |
| 605 | orig_kill() |
| 606 | |
| 607 | proc = transport.get_extra_info('subprocess') |
| 608 | orig_kill = proc.kill |
| 609 | proc.kill = kill |
| 610 | returncode = transport.get_returncode() |
| 611 | transport.close() |
| 612 | await asyncio.wait_for(transport._wait(), 5) |
| 613 | return (returncode, kill_called) |
| 614 | |
| 615 | # Ignore "Close running child process: kill ..." log |
| 616 | with test_utils.disable_logger(): |
nothing calls this directly
no test coverage detected