From 2f6cab72c40f3367d32a0761b57e3628d5652c8e Mon Sep 17 00:00:00 2001 From: Jessey van Offeren Date: Thu, 21 May 2026 20:16:22 +0200 Subject: [PATCH] feat: shared PTY terminal (M12 Tier 3) + readable form controls MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - feat(share): host-consented interactive terminal over the relay. The host shares a real PTY shell (core/pty_session.py); the guest renders it with pyte and sends keystrokes (gui/terminal_widget.py) — vim/top/tab-completion/Ctrl-C work. Runs as the host's user (never root). The host reads along live and can type too, e.g. a sudo password, which stays local and is never sent to the guest. Off by default. Guest also pulls inventory on join (req_full). - fix(gui): style all form controls (QLineEdit/QPlainTextEdit/spin boxes/combo/ terminals) dark-on-light-text — Fusion defaulted them to unreadable light-on-light. - replaces the command/response shell with the full PTY; adds pyte to the gui extra. Verified end-to-end against the deployed relay (guest keystroke ran on host PTY). Co-Authored-By: Claude Opus 4.7 (1M context) --- CHANGELOG.md | 11 ++ docs/ROADMAP.md | 4 +- pyproject.toml | 4 +- src/rigdoctor/__init__.py | 2 +- src/rigdoctor/core/pty_session.py | 59 +++++++ src/rigdoctor/gui/share_page.py | 220 +++++++++++++++++++++------ src/rigdoctor/gui/terminal_widget.py | 84 ++++++++++ src/rigdoctor/gui/theme.py | 12 ++ tests/test_pty_session.py | 27 ++++ 9 files changed, 368 insertions(+), 55 deletions(-) create mode 100644 src/rigdoctor/core/pty_session.py create mode 100644 src/rigdoctor/gui/terminal_widget.py create mode 100644 tests/test_pty_session.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d7d4bbf..4b76170 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,17 @@ All notable changes to RigDoctor are recorded here. Format follows (`MAJOR.MINOR.PATCH`, pre-1.0). `__version__` and `pyproject.toml` must match the git release tag (so the auto-updater, D18, can compare versions). +## [0.7.0] - 2026-05-21 +### Added +- **Shared terminal (M12, Tier 3)**: when the host enables it, the session shares a real **PTY** + shell — the guest gets an interactive terminal (vim, top, tab-completion, Ctrl-C) running on + the host as the host's user. The host **reads along** live and can type too, e.g. a `sudo` + password — which stays local and is never sent to the guest. Off by default, host-consented. + The guest also pulls the host's inventory on join. +### Fixed +- **Input contrast**: all form controls (text fields, spin boxes, combo boxes, terminals) now + use the dark theme with readable text (Fusion defaulted them to light-on-light). + ## [0.6.0] - 2026-05-21 ### Added - **Session sharing over the relay (M12)**: a **Share** tab — *Start shared session* (host) diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 68abee7..f4511a5 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -60,8 +60,8 @@ Escalating ladder, built in order: it in RigDoctor. One-way, safest. - [x] Tier 2: live read-only view — `rigdoctor share serve` (stdlib HTTP, token-gated: sensors + health + inventory). Remote = user-chosen tunnel; GUI controls still to add. -- [ ] Tier 3: gated interactive terminal (wrap tmate/sshx; read-only default, read-write on - explicit consent), with session audit log. +- [x] Tier 3: host-consented interactive terminal — a real PTY shell shared over the relay + (own `pty`, pyte-rendered guest), off by default; host reads along + can type (sudo). > **Out of scope:** stress/repro module (D7); multi-distro support and packaging beyond > Ubuntu/apt + `.deb` (D15) — a thin seam is kept but not built out. diff --git a/pyproject.toml b/pyproject.toml index fd80948..22419d0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "rigdoctor" -version = "0.6.0" +version = "0.7.0" description = "Modular hardware monitoring & crash diagnostics for Linux gamers." readme = "README.md" requires-python = ">=3.11" @@ -13,7 +13,7 @@ requires-python = ">=3.11" dependencies = [] [project.optional-dependencies] -gui = ["PySide6"] +gui = ["PySide6", "pyte"] [project.scripts] rigdoctor = "rigdoctor.cli:main" diff --git a/src/rigdoctor/__init__.py b/src/rigdoctor/__init__.py index 31154b3..6bc8e0b 100644 --- a/src/rigdoctor/__init__.py +++ b/src/rigdoctor/__init__.py @@ -1,3 +1,3 @@ """RigDoctor — modular hardware monitoring & crash diagnostics for Linux gamers.""" -__version__ = "0.6.0" +__version__ = "0.7.0" diff --git a/src/rigdoctor/core/pty_session.py b/src/rigdoctor/core/pty_session.py new file mode 100644 index 0000000..548cfff --- /dev/null +++ b/src/rigdoctor/core/pty_session.py @@ -0,0 +1,59 @@ +"""A pseudo-terminal running the host's shell (M12, Tier 3 — host side). + +Spawns the user's login shell in a real PTY so interactive programs work over a shared +session: vim, top, tab-completion, colours, Ctrl-C, and `sudo` (which prompts inside the +PTY — the host types that password locally, so it's never sent to the guest). Runs as the +host's own user — never elevated. Linux-only (uses `pty`/`termios`). +""" + +from __future__ import annotations + +import fcntl +import os +import pty +import signal +import struct +import termios + + +class PtySession: + def __init__(self, rows: int = 24, cols: int = 80): + self.pid, self.master_fd = pty.fork() + if self.pid == 0: # child: become the shell + os.environ["TERM"] = "xterm-256color" + shell = os.environ.get("SHELL", "/bin/bash") + try: + os.execvp(shell, [shell]) + finally: + os._exit(1) + os.set_blocking(self.master_fd, False) + self.set_size(rows, cols) + + def set_size(self, rows: int, cols: int) -> None: + try: + fcntl.ioctl(self.master_fd, termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols, 0, 0)) + except OSError: + pass + + def write(self, data: bytes) -> None: + try: + os.write(self.master_fd, data) + except OSError: + pass + + def read(self, size: int = 65536) -> bytes: + try: + return os.read(self.master_fd, size) + except (BlockingIOError, OSError): + return b"" + + def close(self) -> None: + try: + os.close(self.master_fd) + except OSError: + pass + try: + os.kill(self.pid, signal.SIGHUP) + os.waitpid(self.pid, os.WNOHANG) + except (OSError, ChildProcessError, ProcessLookupError): + pass diff --git a/src/rigdoctor/gui/share_page.py b/src/rigdoctor/gui/share_page.py index f6cc218..5c76a67 100644 --- a/src/rigdoctor/gui/share_page.py +++ b/src/rigdoctor/gui/share_page.py @@ -1,12 +1,20 @@ -"""Share page (M12, Tier 2 over the relay): host or join a read-only shared session.""" +"""Share page (M12): host or join a shared session over the relay. + +Guest sees the host's live sensors + health + inventory (read-only). If the host enables it, +a full **PTY terminal** is shared: the guest types and the commands run on the host (as the +host's user), the host reads along, and the host can type too — e.g. a sudo password, which +stays local and is never sent to the guest. +""" from __future__ import annotations +import base64 import json -from PySide6.QtCore import Qt, QTimer, QUrl +from PySide6.QtCore import Qt, QSocketNotifier, QTimer, QUrl from PySide6.QtWebSockets import QWebSocket from PySide6.QtWidgets import ( + QCheckBox, QFrame, QHBoxLayout, QLabel, @@ -19,14 +27,20 @@ from PySide6.QtWidgets import ( from ..config import load_config, load_token from ..core import share +from ..core.pty_session import PtySession from ..core.sampler import Sampler from ..core.sources import available_sources +from .terminal_widget import TerminalView def _relay_url() -> str: return load_config().get("relay_url", "wss://rigdoctor.jesseyvanofferen.com").rstrip("/") +def _b64(data: bytes) -> str: + return base64.b64encode(data).decode("ascii") + + def _card(title: str) -> tuple[QFrame, QVBoxLayout]: card = QFrame() card.setObjectName("Card") @@ -46,6 +60,8 @@ class SharePage(QWidget): self._sampler = Sampler(available_sources()) self._host_ws: QWebSocket | None = None self._guest_ws: QWebSocket | None = None + self._pty: PtySession | None = None + self._pty_notifier: QSocketNotifier | None = None self._last_report = None self._last_inv = None self._timer = QTimer(self) @@ -54,18 +70,22 @@ class SharePage(QWidget): root = QVBoxLayout(self) root.setContentsMargins(20, 18, 20, 18) - root.setSpacing(16) + root.setSpacing(14) title = QLabel("Share") title.setObjectName("PageTitle") root.addWidget(title) + root.addWidget(self._build_host()) + root.addWidget(self._build_guest(), 1) - # Host - host_card, hv = _card("Start a shared session") + # ------------------------------------------------------------------ host + def _build_host(self) -> QFrame: + card, v = _card("Start a shared session") self._host_status = QLabel("Let someone with an account view your machine, read-only.") self._host_status.setObjectName("Muted") self._host_status.setWordWrap(True) - hv.addWidget(self._host_status) - hrow = QHBoxLayout() + v.addWidget(self._host_status) + + row = QHBoxLayout() self._start_btn = QPushButton("Start shared session") self._start_btn.setObjectName("PrimaryButton") self._start_btn.clicked.connect(self._start_host) @@ -75,45 +95,25 @@ class SharePage(QWidget): self._code_label = QLabel("") self._code_label.setStyleSheet("font-weight:700; font-size:18px; color:#38bdf8; background:transparent;") self._code_label.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse) - hrow.addWidget(self._start_btn) - hrow.addWidget(self._stop_btn) - hrow.addSpacing(12) - hrow.addWidget(self._code_label) - hrow.addStretch(1) - hv.addLayout(hrow) - root.addWidget(host_card) + row.addWidget(self._start_btn) + row.addWidget(self._stop_btn) + row.addSpacing(12) + row.addWidget(self._code_label) + row.addStretch(1) + v.addLayout(row) - # Guest - guest_card, gv = _card("Join a shared session") - grow = QHBoxLayout() - self._code_input = QLineEdit() - self._code_input.setPlaceholderText("Enter share code") - self._code_input.setMaxLength(6) - self._code_input.setFixedWidth(160) - self._join_btn = QPushButton("Join") - self._join_btn.setObjectName("PrimaryButton") - self._join_btn.clicked.connect(self._join) - self._leave_btn = QPushButton("Leave") - self._leave_btn.setEnabled(False) - self._leave_btn.clicked.connect(self._leave) - grow.addWidget(self._code_input) - grow.addWidget(self._join_btn) - grow.addWidget(self._leave_btn) - grow.addStretch(1) - gv.addLayout(grow) - self._guest_status = QLabel("") - self._guest_status.setObjectName("Muted") - gv.addWidget(self._guest_status) - root.addWidget(guest_card) + self._allow_term = QCheckBox("Allow remote terminal — the guest runs commands as your user (you read along; you can type too, e.g. a sudo password)") + self._allow_term.setStyleSheet("color:#fb923c; background:transparent;") + self._allow_term.toggled.connect(self._toggle_terminal) + v.addWidget(self._allow_term) - self._view = QTextEdit() - self._view.setObjectName("Report") - self._view.setReadOnly(True) - self._view.setVisible(False) - root.addWidget(self._view, 1) - root.addStretch(0) + self._host_term = TerminalView() + self._host_term.keys.connect(lambda b: self._pty.write(b) if self._pty else None) + self._host_term.resized.connect(lambda r, c: self._pty.set_size(r, c) if self._pty else None) + self._host_term.setVisible(False) + v.addWidget(self._host_term) + return card - # --- host --------------------------------------------------------------- def _start_host(self) -> None: if not load_token(): self._host_status.setText("Set a Gitea access token in Setup → Account access first.") @@ -135,13 +135,66 @@ class SharePage(QWidget): if data.get("error"): self._host_status.setText(f"Rejected: {data['error']}") return - if "code" in data: + if "code" in data: # relay handshake self._code_label.setText(data["code"]) self._host_status.setText(f"Sharing as {data.get('user', '?')} — give this code to whoever should view your machine.") self._stop_btn.setEnabled(True) self._host_ws.sendTextMessage(share.host_full_frame(self._sampler)) + self._send_terminal_state() + if self._allow_term.isChecked(): + self._start_pty() self._timer.start() - # guest input also arrives here; ignored (read-only session) + return + kind = data.get("type") # frames forwarded from a guest + if kind == "req_full": + self._host_ws.sendTextMessage(share.host_full_frame(self._sampler)) + elif kind == "pty_in" and self._pty: + self._pty.write(base64.b64decode(data["data"])) + elif kind == "pty_resize" and self._pty: + self._pty.set_size(int(data["rows"]), int(data["cols"])) + + def _toggle_terminal(self, on: bool) -> None: + if on and self._host_ws and self._code_label.text(): + self._start_pty() + elif not on: + self._stop_pty() + self._send_terminal_state() + + def _send_terminal_state(self) -> None: + if self._host_ws and self._code_label.text(): + self._host_ws.sendTextMessage(json.dumps({"type": "terminal", "enabled": self._allow_term.isChecked()})) + + def _start_pty(self) -> None: + if self._pty: + return + rows, cols = self._host_term.grid() + self._pty = PtySession(rows=rows, cols=cols) + self._pty_notifier = QSocketNotifier(self._pty.master_fd, QSocketNotifier.Type.Read, self) + self._pty_notifier.activated.connect(self._on_pty_output) + self._host_term.reset() + self._host_term.setVisible(True) + + def _on_pty_output(self) -> None: + if not self._pty: + return + data = self._pty.read() + if not data: # shell exited / EOF + self._stop_pty() + self._send_terminal_state() + self._allow_term.setChecked(False) + return + self._host_term.feed(data) + if self._host_ws: + self._host_ws.sendTextMessage(json.dumps({"type": "pty", "data": _b64(data)})) + + def _stop_pty(self) -> None: + if self._pty_notifier: + self._pty_notifier.setEnabled(False) + self._pty_notifier = None + if self._pty: + self._pty.close() + self._pty = None + self._host_term.setVisible(False) def _stream(self) -> None: if self._host_ws: @@ -149,6 +202,7 @@ class SharePage(QWidget): def _stop_host(self) -> None: self._timer.stop() + self._stop_pty() if self._host_ws: self._host_ws.close() self._host_ws = None @@ -159,13 +213,54 @@ class SharePage(QWidget): def _host_closed(self) -> None: self._timer.stop() + self._stop_pty() self._start_btn.setEnabled(True) self._stop_btn.setEnabled(False) if self._code_label.text(): self._code_label.setText("") self._host_status.setText("Disconnected from the relay.") - # --- guest -------------------------------------------------------------- + # ----------------------------------------------------------------- guest + def _build_guest(self) -> QFrame: + card, v = _card("Join a shared session") + row = QHBoxLayout() + self._code_input = QLineEdit() + self._code_input.setPlaceholderText("Enter share code") + self._code_input.setMaxLength(6) + self._code_input.setFixedWidth(160) + self._join_btn = QPushButton("Join") + self._join_btn.setObjectName("PrimaryButton") + self._join_btn.clicked.connect(self._join) + self._leave_btn = QPushButton("Leave") + self._leave_btn.setEnabled(False) + self._leave_btn.clicked.connect(self._leave) + row.addWidget(self._code_input) + row.addWidget(self._join_btn) + row.addWidget(self._leave_btn) + row.addStretch(1) + v.addLayout(row) + self._guest_status = QLabel("") + self._guest_status.setObjectName("Muted") + v.addWidget(self._guest_status) + + self._view = QTextEdit() + self._view.setObjectName("Report") + self._view.setReadOnly(True) + self._view.setVisible(False) + self._view.setMinimumHeight(200) + v.addWidget(self._view) + + self._term_label = QLabel("") + self._term_label.setObjectName("Muted") + self._term_label.setVisible(False) + v.addWidget(self._term_label) + self._guest_term = TerminalView() + self._guest_term.keys.connect(self._guest_key) + self._guest_term.resized.connect(self._guest_resize) + self._guest_term.setVisible(False) + v.addWidget(self._guest_term) + return card + def _join(self) -> None: code = self._code_input.text().strip().upper() if not load_token(): @@ -195,19 +290,43 @@ class SharePage(QWidget): self._guest_status.setText(f"Viewing {data.get('host', '?')}'s machine — read-only.") self._leave_btn.setEnabled(True) self._view.setVisible(True) + self._guest_ws.sendTextMessage(json.dumps({"type": "req_full"})) return kind = data.get("type") - if kind == "full": - self._last_report = data.get("report") - self._last_inv = data.get("inventory") if kind in ("full", "snapshot"): + if kind == "full": + self._last_report = data.get("report") + self._last_inv = data.get("inventory") self._view.setHtml(share.guest_html(data.get("snapshot"), self._last_report, self._last_inv)) + elif kind == "terminal": + self._set_terminal_visible(bool(data.get("enabled"))) + elif kind == "pty": + self._guest_term.feed(base64.b64decode(data["data"])) + + def _set_terminal_visible(self, enabled: bool) -> None: + self._term_label.setVisible(True) + self._term_label.setText("Terminal enabled by host — your keystrokes run on their machine. Click here and type." + if enabled else "Terminal not enabled by the host.") + self._guest_term.setVisible(enabled) + if enabled: + self._guest_term.reset() + self._guest_resize(*self._guest_term.grid()) + self._guest_term.setFocus() + + def _guest_key(self, data: bytes) -> None: + if self._guest_ws: + self._guest_ws.sendTextMessage(json.dumps({"type": "pty_in", "data": _b64(data)})) + + def _guest_resize(self, rows: int, cols: int) -> None: + if self._guest_ws: + self._guest_ws.sendTextMessage(json.dumps({"type": "pty_resize", "rows": rows, "cols": cols})) def _leave(self) -> None: if self._guest_ws: self._guest_ws.close() self._guest_ws = None - self._view.setVisible(False) + for w in (self._view, self._term_label, self._guest_term): + w.setVisible(False) self._leave_btn.setEnabled(False) self._join_btn.setEnabled(True) self._guest_status.setText("Left the session.") @@ -220,6 +339,7 @@ class SharePage(QWidget): def shutdown(self) -> None: self._timer.stop() + self._stop_pty() for ws in (self._host_ws, self._guest_ws): if ws: ws.close() diff --git a/src/rigdoctor/gui/terminal_widget.py b/src/rigdoctor/gui/terminal_widget.py new file mode 100644 index 0000000..72e5d92 --- /dev/null +++ b/src/rigdoctor/gui/terminal_widget.py @@ -0,0 +1,84 @@ +"""A minimal terminal view: renders PTY output via pyte and emits keystrokes (M12, Tier 3). + +Used by both sides of a shared session — the host (mirrors its local PTY, can also type, e.g. +a sudo password) and the guest (renders the streamed PTY, sends keystrokes). Monochrome for +now; cursor addressing / layout (vim, top) work via pyte. +""" + +from __future__ import annotations + +import pyte +from PySide6.QtCore import Qt, Signal +from PySide6.QtGui import QFontDatabase, QFontMetrics +from PySide6.QtWidgets import QPlainTextEdit + + +class TerminalView(QPlainTextEdit): + keys = Signal(bytes) # user keystrokes -> bytes for the PTY + resized = Signal(int, int) # rows, cols + + def __init__(self, rows: int = 24, cols: int = 80): + super().__init__() + self.setLineWrapMode(QPlainTextEdit.LineWrapMode.NoWrap) + self.setFont(QFontDatabase.systemFont(QFontDatabase.SystemFont.FixedFont)) + self.setUndoRedoEnabled(False) + self.setMinimumHeight(260) + self._rows, self._cols = rows, cols + self._screen = pyte.Screen(cols, rows) + self._stream = pyte.ByteStream(self._screen) + + def grid(self) -> tuple[int, int]: + return self._rows, self._cols + + def feed(self, data: bytes) -> None: + self._stream.feed(data) + self._render() + + def reset(self) -> None: + self._screen.reset() + self._render() + + def _render(self) -> None: + bar = self.verticalScrollBar().value() + self.setPlainText("\n".join(self._screen.display)) + self.verticalScrollBar().setValue(bar) + + def resizeEvent(self, event): # noqa: N802 (Qt override) + super().resizeEvent(event) + fm = QFontMetrics(self.font()) + cw = max(1, fm.horizontalAdvance("M")) + ch = max(1, fm.height()) + cols = max(20, self.viewport().width() // cw) + rows = max(6, self.viewport().height() // ch) + if (rows, cols) != (self._rows, self._cols): + self._rows, self._cols = rows, cols + self._screen.resize(rows, cols) + self._render() + self.resized.emit(rows, cols) + + def keyPressEvent(self, event): # noqa: N802 (Qt override) + data = self._translate(event) + if data: + self.keys.emit(data) + event.accept() # display comes from PTY output, not local editing + + @staticmethod + def _translate(event) -> bytes: + key = event.key() + mod = event.modifiers() + k = Qt.Key + if mod & Qt.KeyboardModifier.ControlModifier and k.Key_A.value <= key <= k.Key_Z.value: + return bytes([key - k.Key_A.value + 1]) # Ctrl-A..Ctrl-Z + special = { + k.Key_Return.value: b"\r", k.Key_Enter.value: b"\r", + k.Key_Backspace.value: b"\x7f", k.Key_Tab.value: b"\t", + k.Key_Escape.value: b"\x1b", + k.Key_Up.value: b"\x1b[A", k.Key_Down.value: b"\x1b[B", + k.Key_Right.value: b"\x1b[C", k.Key_Left.value: b"\x1b[D", + k.Key_Home.value: b"\x1b[H", k.Key_End.value: b"\x1b[F", + k.Key_Delete.value: b"\x1b[3~", k.Key_PageUp.value: b"\x1b[5~", k.Key_PageDown.value: b"\x1b[6~", + } + if key in special: + return special[key] + text = event.text() + return text.encode("utf-8") if text else b"" diff --git a/src/rigdoctor/gui/theme.py b/src/rigdoctor/gui/theme.py index 7e2d0cb..d90ed49 100644 --- a/src/rigdoctor/gui/theme.py +++ b/src/rigdoctor/gui/theme.py @@ -14,6 +14,7 @@ CARD_BORDER = "#2a2f39" TRACK = "#2a2f39" TEXT = "#e6e8eb" MUTED = "#8b929c" +INPUT_BG = "#0d0f13" # form-control background (must stay dark — see contrast rule) ACCENT = "#38bdf8" COLD = "#7dd3fc" # icey-blue @@ -138,4 +139,15 @@ QCheckBox::indicator:checked {{ QDialog {{ background: {BG}; }} QMessageBox {{ background: {CARD}; }} QDialog QLabel, QMessageBox QLabel {{ color: {TEXT}; background: transparent; }} + +/* Form controls: keep dark bg + light text (Fusion defaults to light-on-light here). */ +QLineEdit, QPlainTextEdit, QAbstractSpinBox, QComboBox {{ + background: {INPUT_BG}; color: {TEXT}; + border: 1px solid {CARD_BORDER}; border-radius: 6px; padding: 5px 8px; + selection-background-color: {ACCENT}; selection-color: #06222e; +}} +QLineEdit:focus, QPlainTextEdit:focus, QAbstractSpinBox:focus, QComboBox:focus {{ + border: 1px solid {ACCENT}; +}} +QLineEdit:disabled, QPlainTextEdit:disabled, QAbstractSpinBox:disabled {{ color: {MUTED}; }} """ diff --git a/tests/test_pty_session.py b/tests/test_pty_session.py new file mode 100644 index 0000000..46d5a77 --- /dev/null +++ b/tests/test_pty_session.py @@ -0,0 +1,27 @@ +"""Tests for the host PTY session (M12 Tier 3).""" + +import time +import unittest + +from rigdoctor.core.pty_session import PtySession + + +class PtySessionTests(unittest.TestCase): + def test_runs_command_and_reads_output(self): + pty = PtySession(rows=24, cols=80) + try: + time.sleep(0.4) + pty.read() # drain the shell prompt + pty.write(b"echo PTY_MARKER_42\n") + deadline = time.time() + 3 + buf = "" + while time.time() < deadline and "PTY_MARKER_42" not in buf: + time.sleep(0.1) + buf += pty.read().decode(errors="replace") + self.assertIn("PTY_MARKER_42", buf) + finally: + pty.close() + + +if __name__ == "__main__": + unittest.main()