jessey
03b2dd8363
D12 "build first" wrapper: `rigdoctor wrap %command%` (Steam launch option / Lutris/Heroic wrapper field) auto-brackets a focused diagnostic around a game — start a game-tagged capture on launch, clean stop on exit; a hard freeze leaves it unterminated → flagged as a crash next launch. - core/wrap.py: game name from SteamAppId, PATH-proof launch_option(), run() that doesn't disturb an existing capture and returns the game's exit code. - diagnostic.start() preserves an unanalyzed crash to diagnostic-crash.jsonl before clearing, so auto-relaunch can't wipe an unseen crash; pending_crash/ analyze_crash check the archive first. - GUI: "Auto-capture…" helper dialog (copyable launch-option string). - Tests for wrap (name resolution, exit-code passthrough, no-double-start). - docs: fix stale MODULES.md status column (M1/M3/M4/M5/M8/M10/M13 → done), update ROADMAP/MODULES for the wrapper + crash detection. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
112 lines
5.2 KiB
Python
112 lines
5.2 KiB
Python
"""Tests for the guided diagnostic orchestration (M3+M4 glue)."""
|
||
|
||
import tempfile
|
||
import time
|
||
import unittest
|
||
from pathlib import Path
|
||
from unittest import mock
|
||
|
||
from rigdoctor.core import diagnostic
|
||
from rigdoctor.core.crashlog import CrashLogWriter, summarize
|
||
from rigdoctor.core.health import Finding
|
||
from rigdoctor.core.sample import Reading, Sample
|
||
|
||
|
||
def _write_log(path: str, game: str) -> None:
|
||
w = CrashLogWriter(path)
|
||
w.write_event("session-start", "interval=1s")
|
||
w.write_event("game", game)
|
||
for temp in (60.0, 72.0, 81.0):
|
||
w.write_sample(Sample(ts=time.time(), readings=[Reading("gpu", "temp", temp, "°C", "")]))
|
||
w.write_event("gpu-lost", "nvidia-smi query timed out")
|
||
w.close()
|
||
|
||
|
||
class GameRecoveryTests(unittest.TestCase):
|
||
def test_game_recovered_from_log_event(self):
|
||
with tempfile.TemporaryDirectory() as d:
|
||
log = str(Path(d) / "capture.jsonl")
|
||
_write_log(log, "Path of Exile 2")
|
||
summary = summarize(log)
|
||
self.assertEqual(diagnostic._game_from_summary(summary), "Path of Exile 2")
|
||
|
||
def test_no_game_event_returns_none(self):
|
||
with tempfile.TemporaryDirectory() as d:
|
||
log = str(Path(d) / "capture.jsonl")
|
||
w = CrashLogWriter(log)
|
||
w.write_event("session-start")
|
||
w.close()
|
||
self.assertIsNone(diagnostic._game_from_summary(summarize(log)))
|
||
|
||
|
||
class FinishTests(unittest.TestCase):
|
||
def test_finish_combines_summary_and_findings(self):
|
||
with tempfile.TemporaryDirectory() as d:
|
||
log = Path(d) / "capture.jsonl"
|
||
_write_log(str(log), "Satisfactory")
|
||
fake = [Finding("warning", "GPU", "NVIDIA Xid 79 ×1", "fell off the bus")]
|
||
with mock.patch("rigdoctor.core.health.run_health_checks", return_value=fake), \
|
||
mock.patch.object(diagnostic.reccontrol, "stop_background", return_value=False), \
|
||
mock.patch.object(diagnostic.reccontrol, "running_pid", return_value=None):
|
||
result = diagnostic.finish(log_path=log)
|
||
self.assertEqual(result.game, "Satisfactory")
|
||
self.assertEqual(result.summary.samples, 3)
|
||
self.assertEqual(result.findings, fake)
|
||
# peak GPU temp captured in the window, GPU-lost event recorded
|
||
self.assertEqual(result.summary.maxima["gpu.temp"][0], 81.0)
|
||
self.assertTrue(any(kind == "gpu-lost" for _ts, kind, _d in result.summary.events))
|
||
|
||
|
||
class CrashDetectionTests(unittest.TestCase):
|
||
def _diag_log(self, d) -> Path:
|
||
return Path(d) / "diagnostic.jsonl"
|
||
|
||
def test_unterminated_session_is_a_pending_crash(self):
|
||
with tempfile.TemporaryDirectory() as d:
|
||
log = self._diag_log(d)
|
||
_write_log(str(log), "Tarkov") # has session-start + game, no session-stop
|
||
with mock.patch.object(diagnostic.config, "DIAG_LOG", log), \
|
||
mock.patch.object(diagnostic.config, "DIAG_CRASH", log.with_suffix(".crash")), \
|
||
mock.patch.object(diagnostic.reccontrol, "running_pid", return_value=None):
|
||
info = diagnostic.pending_crash()
|
||
self.assertIsNotNone(info)
|
||
self.assertEqual(info.game, "Tarkov")
|
||
self.assertTrue(info.gpu_lost) # _write_log writes a gpu-lost event
|
||
|
||
def test_clean_stop_is_not_a_crash(self):
|
||
with tempfile.TemporaryDirectory() as d:
|
||
log = self._diag_log(d)
|
||
w = CrashLogWriter(str(log))
|
||
w.write_event("session-start"); w.write_event("game", "X")
|
||
w.write_sample(Sample(time.time(), [Reading("gpu", "temp", 60.0, "°C", "")]))
|
||
w.write_event("session-stop", "samples=1")
|
||
w.close()
|
||
with mock.patch.object(diagnostic.config, "DIAG_LOG", log), \
|
||
mock.patch.object(diagnostic.config, "DIAG_CRASH", log.with_suffix(".crash")), \
|
||
mock.patch.object(diagnostic.reccontrol, "running_pid", return_value=None):
|
||
self.assertIsNone(diagnostic.pending_crash())
|
||
|
||
def test_acknowledge_clears_pending_crash(self):
|
||
with tempfile.TemporaryDirectory() as d:
|
||
log = self._diag_log(d)
|
||
_write_log(str(log), "Tarkov")
|
||
with mock.patch.object(diagnostic.config, "DIAG_LOG", log), \
|
||
mock.patch.object(diagnostic.config, "DIAG_CRASH", log.with_suffix(".crash")), \
|
||
mock.patch.object(diagnostic.reccontrol, "running_pid", return_value=None):
|
||
self.assertIsNotNone(diagnostic.pending_crash())
|
||
diagnostic.acknowledge_crash()
|
||
self.assertIsNone(diagnostic.pending_crash())
|
||
|
||
def test_running_capture_is_not_a_crash(self):
|
||
with tempfile.TemporaryDirectory() as d:
|
||
log = self._diag_log(d)
|
||
_write_log(str(log), "Tarkov")
|
||
with mock.patch.object(diagnostic.config, "DIAG_LOG", log), \
|
||
mock.patch.object(diagnostic.config, "DIAG_CRASH", log.with_suffix(".crash")), \
|
||
mock.patch.object(diagnostic.reccontrol, "running_pid", return_value=4321):
|
||
self.assertIsNone(diagnostic.pending_crash()) # it's in-progress, not crashed
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|