jessey
305c88ba09
A focused capture that ends without a clean stop (no session-stop, no live recorder) is treated as a likely hard freeze. - core/diagnostic.py: pending_crash() detects the unterminated session; acknowledge_crash() dismisses it; analyze_crash() combines the captured window (final readings + GPU-lost) with a focused scan of the PREVIOUS (crashed) boot + SMART/driver/persistence/temps. - health.check_previous_boot() scans `journalctl -k -b -1`; run_health_checks gained include_journal to avoid double-scanning for the crash path. - GUI: Games page shows a warning banner on launch for an interrupted diagnostic with Analyze crash / Dismiss → results dialog. - Tests for crash detection / clean-stop / acknowledge / in-progress. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
108 lines
4.8 KiB
Python
108 lines
4.8 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.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.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.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.reccontrol, "running_pid", return_value=4321):
|
||
self.assertIsNone(diagnostic.pending_crash()) # it's in-progress, not crashed
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|