| 16 | |
| 17 | |
| 18 | class SysCommandWorker: |
| 19 | def __init__( |
| 20 | self, |
| 21 | cmd: str | list[str], |
| 22 | peek_output: bool | None = False, |
| 23 | environment_vars: dict[str, str] | None = None, |
| 24 | working_directory: str = './', |
| 25 | remove_vt100_escape_codes_from_lines: bool = True, |
| 26 | ): |
| 27 | if isinstance(cmd, str): |
| 28 | cmd = shlex.split(cmd) |
| 29 | |
| 30 | if cmd and not cmd[0].startswith(('/', './')): # Path() does not work well |
| 31 | cmd[0] = locate_binary(cmd[0]) |
| 32 | |
| 33 | self.cmd = cmd |
| 34 | self.peek_output = peek_output |
| 35 | # define the standard locale for command outputs. For now the C ascii one. Can be overridden |
| 36 | self.environment_vars = {'LC_ALL': 'C'} |
| 37 | if environment_vars: |
| 38 | self.environment_vars.update(environment_vars) |
| 39 | |
| 40 | self.working_directory = working_directory |
| 41 | |
| 42 | self.exit_code: int | None = None |
| 43 | self._trace_log = b'' |
| 44 | self._trace_log_pos = 0 |
| 45 | self.poll_object = epoll() |
| 46 | self.child_fd: int | None = None |
| 47 | self.started = False |
| 48 | self.ended = False |
| 49 | self.remove_vt100_escape_codes_from_lines: bool = remove_vt100_escape_codes_from_lines |
| 50 | |
| 51 | def __contains__(self, key: bytes) -> bool: |
| 52 | """ |
| 53 | Contains will also move the current buffert position forward. |
| 54 | This is to avoid re-checking the same data when looking for output. |
| 55 | """ |
| 56 | assert isinstance(key, bytes) |
| 57 | |
| 58 | index = self._trace_log.find(key, self._trace_log_pos) |
| 59 | if index >= 0: |
| 60 | self._trace_log_pos += index + len(key) |
| 61 | return True |
| 62 | |
| 63 | return False |
| 64 | |
| 65 | def __iter__(self, *args: str, **kwargs: dict[str, Any]) -> Iterator[bytes]: |
| 66 | last_line = self._trace_log.rfind(b'\n') |
| 67 | lines = filter(None, self._trace_log[self._trace_log_pos : last_line].splitlines()) |
| 68 | for line in lines: |
| 69 | if self.remove_vt100_escape_codes_from_lines: |
| 70 | line = clear_vt100_escape_codes(line) |
| 71 | |
| 72 | yield line + b'\n' |
| 73 | |
| 74 | self._trace_log_pos = last_line |
| 75 |
no outgoing calls
no test coverage detected