Close process by closing stdin and waiting for it to exit.
| 39 | |
| 40 | @dataclasses.dataclass(slots=True) |
| 41 | class Pclose(contextlib.AbstractContextManager): |
| 42 | """Close process by closing stdin and waiting for it to exit.""" |
| 43 | |
| 44 | proc: subprocess.Popen[str] |
| 45 | |
| 46 | # Set a long timeout to give time for any cache exports to pack layers up |
| 47 | # which currently has to happen synchronously with the session. |
| 48 | timeout: int = 300 |
| 49 | |
| 50 | def __exit__(self, exc_type, exc_value, traceback): |
| 51 | # Kill the child process by closing stdin, not via SIGKILL, so it has |
| 52 | # a chance to drain logs. |
| 53 | try: |
| 54 | if self.proc.stdin: |
| 55 | self.proc.stdin.close() |
| 56 | except AttributeError: |
| 57 | # FakeProcess doesn't have a stdin attribute (tests) |
| 58 | self.proc.terminate() |
| 59 | |
| 60 | try: |
| 61 | self._wait() |
| 62 | except Exception: # Including KeyboardInterrupt, wait handled that. |
| 63 | self.proc.kill() |
| 64 | # We don't call proc.wait() again as proc.__exit__ does that for us. |
| 65 | raise |
| 66 | |
| 67 | def _wait(self): |
| 68 | # avoids raise-within-try (TRY301) |
| 69 | if self.proc.wait(self.timeout): |
| 70 | # non-zero exit code |
| 71 | msg = make_process_error_msg(self.proc, None, None) |
| 72 | raise SessionError(msg) |
| 73 | |
| 74 | |
| 75 | @contextlib.contextmanager |
no outgoing calls
no test coverage detected