Compare commits

...

1 Commits

Author SHA1 Message Date
jessey 2f6cab72c4 feat: shared PTY terminal (M12 Tier 3) + readable form controls
release / release (push) Successful in 14s
- 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) <noreply@anthropic.com>
2026-05-21 20:16:22 +02:00
9 changed files with 368 additions and 55 deletions
+11
View File
@@ -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 (`MAJOR.MINOR.PATCH`, pre-1.0). `__version__` and `pyproject.toml` must match the git
release tag (so the auto-updater, D18, can compare versions). 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 ## [0.6.0] - 2026-05-21
### Added ### Added
- **Session sharing over the relay (M12)**: a **Share** tab — *Start shared session* (host) - **Session sharing over the relay (M12)**: a **Share** tab — *Start shared session* (host)
+2 -2
View File
@@ -60,8 +60,8 @@ Escalating ladder, built in order:
it in RigDoctor. One-way, safest. it in RigDoctor. One-way, safest.
- [x] Tier 2: live read-only view — `rigdoctor share serve` (stdlib HTTP, token-gated: - [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. 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 - [x] Tier 3: host-consented interactive terminal — a real PTY shell shared over the relay
explicit consent), with session audit log. (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 > **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. > Ubuntu/apt + `.deb` (D15) — a thin seam is kept but not built out.
+2 -2
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "rigdoctor" name = "rigdoctor"
version = "0.6.0" version = "0.7.0"
description = "Modular hardware monitoring & crash diagnostics for Linux gamers." description = "Modular hardware monitoring & crash diagnostics for Linux gamers."
readme = "README.md" readme = "README.md"
requires-python = ">=3.11" requires-python = ">=3.11"
@@ -13,7 +13,7 @@ requires-python = ">=3.11"
dependencies = [] dependencies = []
[project.optional-dependencies] [project.optional-dependencies]
gui = ["PySide6"] gui = ["PySide6", "pyte"]
[project.scripts] [project.scripts]
rigdoctor = "rigdoctor.cli:main" rigdoctor = "rigdoctor.cli:main"
+1 -1
View File
@@ -1,3 +1,3 @@
"""RigDoctor — modular hardware monitoring & crash diagnostics for Linux gamers.""" """RigDoctor — modular hardware monitoring & crash diagnostics for Linux gamers."""
__version__ = "0.6.0" __version__ = "0.7.0"
+59
View File
@@ -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
+170 -50
View File
@@ -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 from __future__ import annotations
import base64
import json import json
from PySide6.QtCore import Qt, QTimer, QUrl from PySide6.QtCore import Qt, QSocketNotifier, QTimer, QUrl
from PySide6.QtWebSockets import QWebSocket from PySide6.QtWebSockets import QWebSocket
from PySide6.QtWidgets import ( from PySide6.QtWidgets import (
QCheckBox,
QFrame, QFrame,
QHBoxLayout, QHBoxLayout,
QLabel, QLabel,
@@ -19,14 +27,20 @@ from PySide6.QtWidgets import (
from ..config import load_config, load_token from ..config import load_config, load_token
from ..core import share from ..core import share
from ..core.pty_session import PtySession
from ..core.sampler import Sampler from ..core.sampler import Sampler
from ..core.sources import available_sources from ..core.sources import available_sources
from .terminal_widget import TerminalView
def _relay_url() -> str: def _relay_url() -> str:
return load_config().get("relay_url", "wss://rigdoctor.jesseyvanofferen.com").rstrip("/") 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]: def _card(title: str) -> tuple[QFrame, QVBoxLayout]:
card = QFrame() card = QFrame()
card.setObjectName("Card") card.setObjectName("Card")
@@ -46,6 +60,8 @@ class SharePage(QWidget):
self._sampler = Sampler(available_sources()) self._sampler = Sampler(available_sources())
self._host_ws: QWebSocket | None = None self._host_ws: QWebSocket | None = None
self._guest_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_report = None
self._last_inv = None self._last_inv = None
self._timer = QTimer(self) self._timer = QTimer(self)
@@ -54,18 +70,22 @@ class SharePage(QWidget):
root = QVBoxLayout(self) root = QVBoxLayout(self)
root.setContentsMargins(20, 18, 20, 18) root.setContentsMargins(20, 18, 20, 18)
root.setSpacing(16) root.setSpacing(14)
title = QLabel("Share") title = QLabel("Share")
title.setObjectName("PageTitle") title.setObjectName("PageTitle")
root.addWidget(title) root.addWidget(title)
root.addWidget(self._build_host())
root.addWidget(self._build_guest(), 1)
# Host # ------------------------------------------------------------------ host
host_card, hv = _card("Start a shared session") 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 = QLabel("Let someone with an account view your machine, read-only.")
self._host_status.setObjectName("Muted") self._host_status.setObjectName("Muted")
self._host_status.setWordWrap(True) self._host_status.setWordWrap(True)
hv.addWidget(self._host_status) v.addWidget(self._host_status)
hrow = QHBoxLayout()
row = QHBoxLayout()
self._start_btn = QPushButton("Start shared session") self._start_btn = QPushButton("Start shared session")
self._start_btn.setObjectName("PrimaryButton") self._start_btn.setObjectName("PrimaryButton")
self._start_btn.clicked.connect(self._start_host) self._start_btn.clicked.connect(self._start_host)
@@ -75,45 +95,25 @@ class SharePage(QWidget):
self._code_label = QLabel("") self._code_label = QLabel("")
self._code_label.setStyleSheet("font-weight:700; font-size:18px; color:#38bdf8; background:transparent;") self._code_label.setStyleSheet("font-weight:700; font-size:18px; color:#38bdf8; background:transparent;")
self._code_label.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse) self._code_label.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
hrow.addWidget(self._start_btn) row.addWidget(self._start_btn)
hrow.addWidget(self._stop_btn) row.addWidget(self._stop_btn)
hrow.addSpacing(12) row.addSpacing(12)
hrow.addWidget(self._code_label) row.addWidget(self._code_label)
hrow.addStretch(1) row.addStretch(1)
hv.addLayout(hrow) v.addLayout(row)
root.addWidget(host_card)
# Guest 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)")
guest_card, gv = _card("Join a shared session") self._allow_term.setStyleSheet("color:#fb923c; background:transparent;")
grow = QHBoxLayout() self._allow_term.toggled.connect(self._toggle_terminal)
self._code_input = QLineEdit() v.addWidget(self._allow_term)
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._view = QTextEdit() self._host_term = TerminalView()
self._view.setObjectName("Report") self._host_term.keys.connect(lambda b: self._pty.write(b) if self._pty else None)
self._view.setReadOnly(True) self._host_term.resized.connect(lambda r, c: self._pty.set_size(r, c) if self._pty else None)
self._view.setVisible(False) self._host_term.setVisible(False)
root.addWidget(self._view, 1) v.addWidget(self._host_term)
root.addStretch(0) return card
# --- host ---------------------------------------------------------------
def _start_host(self) -> None: def _start_host(self) -> None:
if not load_token(): if not load_token():
self._host_status.setText("Set a Gitea access token in Setup → Account access first.") 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"): if data.get("error"):
self._host_status.setText(f"Rejected: {data['error']}") self._host_status.setText(f"Rejected: {data['error']}")
return return
if "code" in data: if "code" in data: # relay handshake
self._code_label.setText(data["code"]) 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._host_status.setText(f"Sharing as {data.get('user', '?')} — give this code to whoever should view your machine.")
self._stop_btn.setEnabled(True) self._stop_btn.setEnabled(True)
self._host_ws.sendTextMessage(share.host_full_frame(self._sampler)) 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() 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: def _stream(self) -> None:
if self._host_ws: if self._host_ws:
@@ -149,6 +202,7 @@ class SharePage(QWidget):
def _stop_host(self) -> None: def _stop_host(self) -> None:
self._timer.stop() self._timer.stop()
self._stop_pty()
if self._host_ws: if self._host_ws:
self._host_ws.close() self._host_ws.close()
self._host_ws = None self._host_ws = None
@@ -159,13 +213,54 @@ class SharePage(QWidget):
def _host_closed(self) -> None: def _host_closed(self) -> None:
self._timer.stop() self._timer.stop()
self._stop_pty()
self._start_btn.setEnabled(True) self._start_btn.setEnabled(True)
self._stop_btn.setEnabled(False) self._stop_btn.setEnabled(False)
if self._code_label.text(): if self._code_label.text():
self._code_label.setText("") self._code_label.setText("")
self._host_status.setText("Disconnected from the relay.") 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: def _join(self) -> None:
code = self._code_input.text().strip().upper() code = self._code_input.text().strip().upper()
if not load_token(): 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._guest_status.setText(f"Viewing {data.get('host', '?')}'s machine — read-only.")
self._leave_btn.setEnabled(True) self._leave_btn.setEnabled(True)
self._view.setVisible(True) self._view.setVisible(True)
self._guest_ws.sendTextMessage(json.dumps({"type": "req_full"}))
return return
kind = data.get("type") 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 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)) 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: def _leave(self) -> None:
if self._guest_ws: if self._guest_ws:
self._guest_ws.close() self._guest_ws.close()
self._guest_ws = None 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._leave_btn.setEnabled(False)
self._join_btn.setEnabled(True) self._join_btn.setEnabled(True)
self._guest_status.setText("Left the session.") self._guest_status.setText("Left the session.")
@@ -220,6 +339,7 @@ class SharePage(QWidget):
def shutdown(self) -> None: def shutdown(self) -> None:
self._timer.stop() self._timer.stop()
self._stop_pty()
for ws in (self._host_ws, self._guest_ws): for ws in (self._host_ws, self._guest_ws):
if ws: if ws:
ws.close() ws.close()
+84
View File
@@ -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""
+12
View File
@@ -14,6 +14,7 @@ CARD_BORDER = "#2a2f39"
TRACK = "#2a2f39" TRACK = "#2a2f39"
TEXT = "#e6e8eb" TEXT = "#e6e8eb"
MUTED = "#8b929c" MUTED = "#8b929c"
INPUT_BG = "#0d0f13" # form-control background (must stay dark — see contrast rule)
ACCENT = "#38bdf8" ACCENT = "#38bdf8"
COLD = "#7dd3fc" # icey-blue COLD = "#7dd3fc" # icey-blue
@@ -138,4 +139,15 @@ QCheckBox::indicator:checked {{
QDialog {{ background: {BG}; }} QDialog {{ background: {BG}; }}
QMessageBox {{ background: {CARD}; }} QMessageBox {{ background: {CARD}; }}
QDialog QLabel, QMessageBox QLabel {{ color: {TEXT}; background: transparent; }} 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}; }}
""" """
+27
View File
@@ -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()