(self)
| 89 | return "".join(list(self._buf)[-int(lines) :]) |
| 90 | |
| 91 | def close(self) -> None: |
| 92 | self._stop.set() |
| 93 | try: |
| 94 | if self.proc.isalive(): |
| 95 | try: |
| 96 | self.proc.sendline("exit") |
| 97 | except Exception: |
| 98 | pass |
| 99 | self.proc.close(force=True) |
| 100 | except Exception: |
| 101 | pass |
| 102 | try: |
| 103 | if self._reader and self._reader.is_alive(): |
| 104 | self._reader.join(timeout=1.0) |
| 105 | except Exception: |
| 106 | pass |
| 107 | |
| 108 | |
| 109 | @dataclass |
no outgoing calls
no test coverage detected