diff --git a/CHANGELOG.md b/CHANGELOG.md index 16f1afd..d7d4bbf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,22 @@ 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.6.0] - 2026-05-21 +### Added +- **Session sharing over the relay (M12)**: a **Share** tab — *Start shared session* (host) + hands you a short code and streams a read-only live view; *Enter share code* (guest) joins + someone else's session and views their sensors/health/inventory. Both connect outbound over + WebSocket to the relay (`relay_url`, default `wss://rigdoctor.jesseyvanofferen.com`), gated + by your Gitea access token — no port forwarding. Read-only. + +## [0.5.0] - 2026-05-21 +### Added +- **Session sharing (M12, Tier 2)**: `rigdoctor share serve` starts a **read-only** live view + (sensors auto-refresh + health report + inventory) over a local HTTP server, gated by a + random share token. Bind to localhost for local testing, or to all interfaces behind a + user-chosen tunnel (Tailscale/cloudflared/SSH) for remote help. No actions, no terminal. + (Tier 1 export and Tier 3 gated terminal still to come — D16.) + ## [0.4.1] - 2026-05-21 ### Fixed - Checkbox contrast: a checked checkbox is now a clear accent-filled box with a checkmark diff --git a/docs/MODULES.md b/docs/MODULES.md index 65ad78f..c218373 100644 --- a/docs/MODULES.md +++ b/docs/MODULES.md @@ -18,7 +18,7 @@ Status: ⬜ not started · 🟦 designing · 🟨 in progress · ✅ done | M10 | Desktop GUI | Desktop UI | **python3-pyside6** | all | P2 | 🟨 | | M11 | Tray / menu-bar applet | Desktop UI | **python3-pyside6** (+ AppIndicator on GNOME) | all | P2 | ⬜ | | M9 | Installer | (meta) | none | all | P1 | 🟨 | -| M12 | Session sharing / remote assist | Sharing | none (Tier 3: tmate/sshx) | all | P3 | ⬜ | +| M12 | Session sharing / remote assist | Sharing | none (Tier 3: tmate/sshx) | all | P3 | 🟨 | | M13 | Auto-update | (core) | none (stdlib; user-local file swap) | all | P3 | 🟨 | | ~~M7~~ | ~~Stress / repro~~ | — | — | — | — | ❌ dropped (D7) | diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 4131fee..68abee7 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -58,8 +58,8 @@ Ubuntu + NVIDIA first; `.deb` distribution (see `DECISIONS.md`). Escalating ladder, built in order: - [ ] Tier 1: `share export` — diagnostic bundle (inventory + recent log + report); B opens it in RigDoctor. One-way, safest. -- [ ] Tier 2: live read-only view (local server + user-chosen tunnel: Tailscale/cloudflared/ - SSH; no hosted relay), token-gated, A approves, revocable. +- [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. diff --git a/pyproject.toml b/pyproject.toml index c2cf622..fd80948 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "rigdoctor" -version = "0.4.1" +version = "0.6.0" description = "Modular hardware monitoring & crash diagnostics for Linux gamers." readme = "README.md" requires-python = ">=3.11" diff --git a/src/rigdoctor/__init__.py b/src/rigdoctor/__init__.py index af2685f..31154b3 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.4.1" +__version__ = "0.6.0" diff --git a/src/rigdoctor/cli.py b/src/rigdoctor/cli.py index 15bd2e9..3842113 100644 --- a/src/rigdoctor/cli.py +++ b/src/rigdoctor/cli.py @@ -295,6 +295,12 @@ def cmd_uninstall(args) -> int: return 0 +def cmd_share_serve(args) -> int: + from .core import share + + return share.serve(host=args.host, port=args.port) + + def cmd_collect_priv(args) -> int: """Internal: emit root-only data (SMART + dmidecode) as JSON, run via pkexec at launch.""" from dataclasses import asdict @@ -405,6 +411,13 @@ def build_parser() -> argparse.ArgumentParser: cp = sub.add_parser("collect-priv", help=argparse.SUPPRESS) # internal: run via pkexec cp.set_defaults(func=cmd_collect_priv) + share_p = sub.add_parser("share", help="session sharing (M12)") + share_sub = share_p.add_subparsers(dest="share_cmd", required=True) + serve_p = share_sub.add_parser("serve", help="serve a read-only live view (token-gated)") + serve_p.add_argument("--host", default="127.0.0.1", help="bind address (use 0.0.0.0 + a tunnel for remote)") + serve_p.add_argument("--port", type=int, default=8765, help="port") + serve_p.set_defaults(func=cmd_share_serve) + inv = sub.add_parser("inventory", help="system inventory (M5): export hardware/OS details") inv.add_argument("--json", action="store_true", help="output JSON") inv.add_argument("--markdown", action="store_true", help="output Markdown (for forum/bug reports)") diff --git a/src/rigdoctor/config.py b/src/rigdoctor/config.py index 1b3d731..12dfa1a 100644 --- a/src/rigdoctor/config.py +++ b/src/rigdoctor/config.py @@ -142,6 +142,7 @@ DEFAULTS: dict = { "alerts_enabled": True, # desktop notifications on overheat / GPU-lost / new version "gpu_temp_alert": 90.0, # °C — alert when GPU reaches this "cpu_temp_alert": 95.0, # °C — alert when CPU reaches this + "relay_url": "wss://rigdoctor.jesseyvanofferen.com", # session-sharing relay (M12) } diff --git a/src/rigdoctor/core/share.py b/src/rigdoctor/core/share.py new file mode 100644 index 0000000..9df990a --- /dev/null +++ b/src/rigdoctor/core/share.py @@ -0,0 +1,194 @@ +"""Session sharing (M12, Tier 2): a read-only live view over a local HTTP server. + +Serves the live sensor snapshot + health report + inventory, **read-only**, gated by a +random share token. Bind to localhost for local testing, or to all interfaces behind a +user-chosen tunnel (Tailscale / cloudflared / SSH) for remote help. No actions, no terminal. +""" + +from __future__ import annotations + +import json +import secrets +from dataclasses import asdict +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import parse_qs, urlparse + +from .sampler import Sampler +from .sources import available_sources + +_PAGE = """ +RigDoctor — shared + +

RigDoctor read-only share

+

A live view shared by the machine's owner. You can look, not change anything.

+
loading…
+

Health

loading…
+

Inventory

loading…
+""" + + +def _snapshot(sampler: Sampler) -> dict: + sample = sampler.sample() + groups: dict[str, list] = {} + for r in sample.readings: + if r.metric == "name": + item = {"name": "device", "value": r.label, "unit": ""} + else: + item = {"name": (r.label + " " + r.metric).strip() if r.label else r.metric, + "value": r.value, "unit": r.unit} + groups.setdefault(r.source, []).append(item) + return {"ts": sample.ts, "groups": groups} + + +def _report() -> list: + from .health import run_health_checks + return [asdict(f) for f in run_health_checks()] + + +def _inventory() -> dict: + from .inventory import collect, to_dict + return to_dict(collect()) + + +# --- Relay (M12) frames: a host streams these; a guest renders them. ----------------- + +def host_full_frame(sampler: Sampler) -> str: + """Initial frame: live snapshot + health report + inventory.""" + return json.dumps({"type": "full", "snapshot": _snapshot(sampler), + "report": _report(), "inventory": _inventory()}) + + +def host_snapshot_frame(sampler: Sampler) -> str: + """Recurring frame: just the live snapshot.""" + return json.dumps({"type": "snapshot", "snapshot": _snapshot(sampler)}) + + +def _fmt(value, unit: str) -> str: + if value is None: + return "N/A" + if unit == "°C": + try: + return f"{float(value):.1f} °C" + except (TypeError, ValueError): + return str(value) + return f"{value} {unit}".strip() + + +def guest_html(snapshot: dict | None, report: list | None, inventory: dict | None) -> str: + """Render a received frame as read-only dark HTML for the guest's view.""" + import html as _html + + def esc(x) -> str: + return _html.escape(str(x)) + + out = ['
'] + if snapshot: + for group, items in snapshot.get("groups", {}).items(): + out.append(f'

{esc(group).upper()}

') + for it in items: + out.append(f'' + f'') + out.append("
{esc(it.get("name"))}{esc(_fmt(it.get("value"), it.get("unit", "")))}
") + if report: + out.append('

HEALTH

') + colors = {"critical": "#f87171", "warning": "#fb923c", "ok": "#4ade80"} + for f in report: + sev = f.get("severity", "info") + out.append(f'
[{esc(sev).upper()}] ' + f'{esc(f.get("category"))}: {esc(f.get("title"))}
') + if inventory: + out.append('

INVENTORY

') + for section, kv in inventory.items(): + out.append(f'

{esc(section)}

') + for k, v in kv.items(): + out.append(f'') + out.append("
{esc(k)}{esc(v)}
") + out.append("
") + return "".join(out) + + +class _Handler(BaseHTTPRequestHandler): + def log_message(self, *args): # quiet + pass + + def _authed(self, query: dict) -> bool: + return secrets.compare_digest(query.get("t", [""])[0], self.server.token) + + def _send(self, code: int, ctype: str, body: bytes) -> None: + self.send_response(code) + self.send_header("Content-Type", ctype) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self) -> None: # noqa: N802 + parsed = urlparse(self.path) + if not self._authed(parse_qs(parsed.query)): + self._send(403, "text/plain", b"Forbidden: missing or invalid share token") + return + if parsed.path == "/": + self._send(200, "text/html; charset=utf-8", _PAGE.encode()) + elif parsed.path == "/api/snapshot": + self._send(200, "application/json", json.dumps(_snapshot(self.server.sampler)).encode()) + elif parsed.path == "/api/report": + self._send(200, "application/json", json.dumps(_report()).encode()) + elif parsed.path == "/api/inventory": + self._send(200, "application/json", json.dumps(_inventory()).encode()) + else: + self._send(404, "text/plain", b"Not found") + + +class _Server(ThreadingHTTPServer): + daemon_threads = True + + def __init__(self, addr, token: str): + super().__init__(addr, _Handler) + self.token = token + self.sampler = Sampler(available_sources()) + + +def make_server(host: str = "127.0.0.1", port: int = 0, token: str | None = None) -> tuple[_Server, str]: + token = token or secrets.token_urlsafe(16) + return _Server((host, port), token), token + + +def serve(host: str = "127.0.0.1", port: int = 8765) -> int: + srv, token = make_server(host, port) + url = f"http://{host}:{srv.server_address[1]}/?t={token}" + print( + f"Sharing a read-only live view at:\n {url}\n\n" + "Anyone with this URL (and network access to this host) can VIEW your sensors,\n" + "health report, and inventory — read-only. For remote help, expose it via a tunnel\n" + "(Tailscale / cloudflared / `ssh -R`). Press Ctrl-C to stop sharing.", + flush=True, + ) + try: + srv.serve_forever() + except KeyboardInterrupt: + print("\nStopped sharing.") + finally: + srv.shutdown() + return 0 diff --git a/src/rigdoctor/gui/main_window.py b/src/rigdoctor/gui/main_window.py index ceb3b2c..6cc0fe9 100644 --- a/src/rigdoctor/gui/main_window.py +++ b/src/rigdoctor/gui/main_window.py @@ -33,10 +33,11 @@ from .inventory_page import InventoryPage from .notifications_page import NotificationsPage from .recorder_page import RecorderPage from .setup_page import SetupPage +from .share_page import SharePage from .theme import ACCENT, GOOD, MUTED from .worker import SamplerWorker -_NAV_ITEMS = ["Dashboard", "Logs", "Health", "Setup", "Inventory", "Notifications"] +_NAV_ITEMS = ["Dashboard", "Logs", "Health", "Setup", "Inventory", "Notifications", "Share"] class MainWindow(QMainWindow): @@ -70,12 +71,14 @@ class MainWindow(QMainWindow): self.inventory_page = InventoryPage() self.notifications_page = NotificationsPage() self.notifications_page.changed.connect(self._apply_alert_settings) + self.share_page = SharePage() self._stack.addWidget(self.dashboard) # 0 Dashboard self._stack.addWidget(self.recorder_page) # 1 Logs self._stack.addWidget(self.health_page) # 2 Health self._stack.addWidget(self.setup_page) # 3 Setup self._stack.addWidget(self.inventory_page) # 4 Inventory self._stack.addWidget(self.notifications_page) # 5 Notifications + self._stack.addWidget(self.share_page) # 6 Share content_layout.addWidget(self._stack) layout.addWidget(self._build_sidebar()) @@ -306,4 +309,5 @@ class MainWindow(QMainWindow): def closeEvent(self, event) -> None: # noqa: N802 (Qt override) self._worker.stop() + self.share_page.shutdown() super().closeEvent(event) diff --git a/src/rigdoctor/gui/setup_page.py b/src/rigdoctor/gui/setup_page.py index 1ece46a..dcfb254 100644 --- a/src/rigdoctor/gui/setup_page.py +++ b/src/rigdoctor/gui/setup_page.py @@ -86,8 +86,13 @@ class SetupPage(QWidget): comp_layout.addLayout(controls) root.addWidget(comp_card) - # Update access (M13): token gating updates to Gitea account holders. - upd_card, upd_layout = _panel("Update access") + # Account access (M13/M12): one Gitea token gates updates and session sharing. + upd_card, upd_layout = _panel("Account access") + hint = QLabel("A Gitea access token unlocks updates and session sharing. " + "Create it with scopes read:user and read:repository.") + hint.setObjectName("Muted") + hint.setWordWrap(True) + upd_layout.addWidget(hint) self._upd_status = QLabel("") self._upd_status.setObjectName("Muted") self._upd_status.setWordWrap(True) @@ -95,7 +100,7 @@ class SetupPage(QWidget): token_row = QHBoxLayout() self._token_input = QLineEdit() self._token_input.setEchoMode(QLineEdit.EchoMode.Password) - self._token_input.setPlaceholderText("Paste a Gitea token (scope: read:repository)") + self._token_input.setPlaceholderText("Paste a Gitea token (read:user + read:repository)") save_btn = QPushButton("Save token") save_btn.setObjectName("PrimaryButton") save_btn.clicked.connect(self._save_token) diff --git a/src/rigdoctor/gui/share_page.py b/src/rigdoctor/gui/share_page.py new file mode 100644 index 0000000..f6cc218 --- /dev/null +++ b/src/rigdoctor/gui/share_page.py @@ -0,0 +1,225 @@ +"""Share page (M12, Tier 2 over the relay): host or join a read-only shared session.""" + +from __future__ import annotations + +import json + +from PySide6.QtCore import Qt, QTimer, QUrl +from PySide6.QtWebSockets import QWebSocket +from PySide6.QtWidgets import ( + QFrame, + QHBoxLayout, + QLabel, + QLineEdit, + QPushButton, + QTextEdit, + QVBoxLayout, + QWidget, +) + +from ..config import load_config, load_token +from ..core import share +from ..core.sampler import Sampler +from ..core.sources import available_sources + + +def _relay_url() -> str: + return load_config().get("relay_url", "wss://rigdoctor.jesseyvanofferen.com").rstrip("/") + + +def _card(title: str) -> tuple[QFrame, QVBoxLayout]: + card = QFrame() + card.setObjectName("Card") + v = QVBoxLayout(card) + v.setContentsMargins(16, 14, 16, 14) + v.setSpacing(10) + head = QLabel(title) + head.setStyleSheet("font-weight: 700; background: transparent;") + v.addWidget(head) + return card, v + + +class SharePage(QWidget): + def __init__(self) -> None: + super().__init__() + self.setObjectName("Page") + self._sampler = Sampler(available_sources()) + self._host_ws: QWebSocket | None = None + self._guest_ws: QWebSocket | None = None + self._last_report = None + self._last_inv = None + self._timer = QTimer(self) + self._timer.setInterval(2000) + self._timer.timeout.connect(self._stream) + + root = QVBoxLayout(self) + root.setContentsMargins(20, 18, 20, 18) + root.setSpacing(16) + title = QLabel("Share") + title.setObjectName("PageTitle") + root.addWidget(title) + + # Host + host_card, hv = _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() + self._start_btn = QPushButton("Start shared session") + self._start_btn.setObjectName("PrimaryButton") + self._start_btn.clicked.connect(self._start_host) + self._stop_btn = QPushButton("Stop") + self._stop_btn.setEnabled(False) + self._stop_btn.clicked.connect(self._stop_host) + 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) + + # 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._view = QTextEdit() + self._view.setObjectName("Report") + self._view.setReadOnly(True) + self._view.setVisible(False) + root.addWidget(self._view, 1) + root.addStretch(0) + + # --- host --------------------------------------------------------------- + def _start_host(self) -> None: + if not load_token(): + self._host_status.setText("Set a Gitea access token in Setup → Account access first.") + return + self._host_status.setText("Connecting to the relay…") + self._start_btn.setEnabled(False) + self._host_ws = QWebSocket() + self._host_ws.connected.connect(lambda: self._host_ws.sendTextMessage(json.dumps({"token": load_token()}))) + self._host_ws.textMessageReceived.connect(self._host_msg) + self._host_ws.disconnected.connect(self._host_closed) + self._host_ws.errorOccurred.connect(lambda *_: self._host_status.setText(f"Relay error: {self._host_ws.errorString()}")) + self._host_ws.open(QUrl(_relay_url() + "/ws/host")) + + def _host_msg(self, text: str) -> None: + try: + data = json.loads(text) + except ValueError: + return + if data.get("error"): + self._host_status.setText(f"Rejected: {data['error']}") + return + if "code" in data: + 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._timer.start() + # guest input also arrives here; ignored (read-only session) + + def _stream(self) -> None: + if self._host_ws: + self._host_ws.sendTextMessage(share.host_snapshot_frame(self._sampler)) + + def _stop_host(self) -> None: + self._timer.stop() + if self._host_ws: + self._host_ws.close() + self._host_ws = None + self._code_label.setText("") + self._stop_btn.setEnabled(False) + self._start_btn.setEnabled(True) + self._host_status.setText("Stopped sharing.") + + def _host_closed(self) -> None: + self._timer.stop() + 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 -------------------------------------------------------------- + def _join(self) -> None: + code = self._code_input.text().strip().upper() + if not load_token(): + self._guest_status.setText("Set a Gitea access token in Setup → Account access first.") + return + if not code: + self._guest_status.setText("Enter a share code.") + return + self._guest_status.setText("Connecting…") + self._join_btn.setEnabled(False) + self._guest_ws = QWebSocket() + self._guest_ws.connected.connect(lambda: self._guest_ws.sendTextMessage(json.dumps({"token": load_token()}))) + self._guest_ws.textMessageReceived.connect(self._guest_msg) + self._guest_ws.disconnected.connect(self._guest_closed) + self._guest_ws.errorOccurred.connect(lambda *_: self._guest_status.setText(f"Relay error: {self._guest_ws.errorString()}")) + self._guest_ws.open(QUrl(_relay_url() + "/ws/guest/" + code)) + + def _guest_msg(self, text: str) -> None: + try: + data = json.loads(text) + except ValueError: + return + if data.get("error"): + self._guest_status.setText(data["error"]) + return + if "joined" in data: + self._guest_status.setText(f"Viewing {data.get('host', '?')}'s machine — read-only.") + self._leave_btn.setEnabled(True) + self._view.setVisible(True) + 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"): + self._view.setHtml(share.guest_html(data.get("snapshot"), self._last_report, self._last_inv)) + + def _leave(self) -> None: + if self._guest_ws: + self._guest_ws.close() + self._guest_ws = None + self._view.setVisible(False) + self._leave_btn.setEnabled(False) + self._join_btn.setEnabled(True) + self._guest_status.setText("Left the session.") + + def _guest_closed(self) -> None: + self._join_btn.setEnabled(True) + self._leave_btn.setEnabled(False) + if self._view.isVisible(): + self._guest_status.setText("Session ended (host disconnected).") + + def shutdown(self) -> None: + self._timer.stop() + for ws in (self._host_ws, self._guest_ws): + if ws: + ws.close() diff --git a/tests/test_relay_frames.py b/tests/test_relay_frames.py new file mode 100644 index 0000000..0e4440e --- /dev/null +++ b/tests/test_relay_frames.py @@ -0,0 +1,38 @@ +"""Tests for M12 relay frames + guest HTML rendering (host/guest data shapes).""" + +import json +import unittest + +from rigdoctor.core import share +from rigdoctor.core.sampler import Sampler +from rigdoctor.core.sources import available_sources + + +class RelayFrameTests(unittest.TestCase): + def setUp(self): + self.sampler = Sampler(available_sources()) + + def test_full_frame_shape(self): + frame = json.loads(share.host_full_frame(self.sampler)) + self.assertEqual(frame["type"], "full") + self.assertIn("groups", frame["snapshot"]) + self.assertIsInstance(frame["report"], list) + self.assertIsInstance(frame["inventory"], dict) + + def test_snapshot_frame_shape(self): + frame = json.loads(share.host_snapshot_frame(self.sampler)) + self.assertEqual(frame["type"], "snapshot") + self.assertIn("groups", frame["snapshot"]) + + def test_guest_html_renders(self): + snap = {"groups": {"gpu": [{"name": "temp", "value": 51.0, "unit": "°C"}]}} + report = [{"severity": "ok", "category": "Logs", "title": "No errors"}] + inv = {"System": {"Kernel": "7.0.0"}} + html = share.guest_html(snap, report, inv) + self.assertIn("51.0 °C", html) + self.assertIn("No errors", html) + self.assertIn("Kernel", html) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_share.py b/tests/test_share.py new file mode 100644 index 0000000..3f3e59c --- /dev/null +++ b/tests/test_share.py @@ -0,0 +1,46 @@ +"""Tests for M12 Tier 2 share server: token gating + endpoints.""" + +import json +import threading +import unittest +import urllib.error +import urllib.request + +from rigdoctor.core import share + + +class ShareServerTests(unittest.TestCase): + def setUp(self): + self.srv, self.token = share.make_server("127.0.0.1", 0) + self.port = self.srv.server_address[1] + self.thread = threading.Thread(target=self.srv.serve_forever, daemon=True) + self.thread.start() + + def tearDown(self): + self.srv.shutdown() + + def _url(self, path, token=None): + q = f"?t={token}" if token else "" + return f"http://127.0.0.1:{self.port}{path}{q}" + + def test_requires_token(self): + with self.assertRaises(urllib.error.HTTPError) as cm: + urllib.request.urlopen(self._url("/api/snapshot"), timeout=10) + self.assertEqual(cm.exception.code, 403) + + def test_bad_token_rejected(self): + with self.assertRaises(urllib.error.HTTPError) as cm: + urllib.request.urlopen(self._url("/api/snapshot", "wrong"), timeout=10) + self.assertEqual(cm.exception.code, 403) + + def test_snapshot_with_token(self): + data = json.load(urllib.request.urlopen(self._url("/api/snapshot", self.token), timeout=10)) + self.assertIn("groups", data) + + def test_page_served(self): + body = urllib.request.urlopen(self._url("/", self.token), timeout=10).read() + self.assertIn(b"read-only share", body) + + +if __name__ == "__main__": + unittest.main()