(self)
| 65 | self._pump_task = asyncio.create_task(self._pump_stdout()) |
| 66 | |
| 67 | async def close(self): |
| 68 | # Cancel the pump task if it exists |
| 69 | if self._pump_task: |
| 70 | self._pump_task.cancel() |
| 71 | try: |
| 72 | await self._pump_task |
| 73 | except asyncio.CancelledError: |
| 74 | pass |
| 75 | except Exception: |
| 76 | pass |
| 77 | |
| 78 | # Terminate the process if it exists |
| 79 | if self._proc: |
| 80 | try: |
| 81 | if getattr(self._proc, "returncode", None) is None: |
| 82 | self._proc.terminate() |
| 83 | except ProcessLookupError: |
| 84 | pass |
| 85 | except Exception: |
| 86 | pass |
| 87 | try: |
| 88 | await self._proc.wait() |
| 89 | except Exception: |
| 90 | pass |
| 91 | |
| 92 | self._release_pty_master() |
| 93 | self._proc = None |
| 94 | self._pump_task = None |
| 95 | |
| 96 | def _release_pty_master(self): |
| 97 | """Release the POSIX PTY master exactly once. |
no test coverage detected