(self, which, encoding, lines)
| 67 | self._worker_threads[:] = [] |
| 68 | |
| 69 | def _output(self, which, encoding, lines): |
| 70 | try: |
| 71 | result = self._chunks[which] |
| 72 | except KeyError: |
| 73 | raise AssertionError( |
| 74 | f"{which} was not captured for {self.session.debuggee_id}" |
| 75 | ) |
| 76 | |
| 77 | with self._lock: |
| 78 | result = b"".join(result) |
| 79 | if encoding is not None: |
| 80 | result = result.decode(encoding) |
| 81 | |
| 82 | return result.splitlines() if lines else result |
| 83 | |
| 84 | def stdout(self, encoding=None): |
| 85 | """Returns stdout captured from the debugged process, as a single string. |
no test coverage detected