import json import sys import tempfile import unittest from pathlib import Path from types import SimpleNamespace from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] SRC = ROOT / "src" if str(SRC) not in sys.path: sys.path.insert(0, str(SRC)) from config import Config from diagnostics import ( DiagnosticCheck, DiagnosticReport, run_doctor, run_self_check, ) class _FakeDesktop: def validate_hotkey(self, _hotkey: str) -> None: return class _Result: def __init__(self, *, returncode: int = 0, stdout: str = "", stderr: str = ""): self.returncode = returncode self.stdout = stdout self.stderr = stderr def _systemctl_side_effect(*results: _Result): iterator = iter(results) def _runner(_args): return next(iterator) return _runner class DiagnosticsTests(unittest.TestCase): def test_run_doctor_all_checks_pass(self): cfg = Config() with tempfile.TemporaryDirectory() as td: config_path = Path(td) / "config.json" config_path.write_text('{"config_version":1}\n', encoding="utf-8") with patch.dict("os.environ", {"DISPLAY": ":0"}, clear=False), patch( "diagnostics.load_existing", return_value=cfg ), patch("diagnostics.list_input_devices", return_value=[{"index": 1, "name": "Mic"}]), patch( "diagnostics.resolve_input_device", return_value=1 ), patch( "diagnostics.get_desktop_adapter", return_value=_FakeDesktop() ), patch( "diagnostics._run_systemctl_user", return_value=_Result(returncode=0, stdout="running\n"), ), patch("diagnostics.probe_managed_model") as probe_model: report = run_doctor(str(config_path)) self.assertEqual(report.status, "ok") self.assertTrue(report.ok) self.assertEqual( [check.id for check in report.checks], [ "config.load", "session.x11", "runtime.audio", "audio.input", "hotkey.parse", "injection.backend", "service.prereq", ], ) self.assertTrue(all(check.status == "ok" for check in report.checks)) probe_model.assert_not_called() def test_run_doctor_missing_config_warns_without_writing(self): with tempfile.TemporaryDirectory() as td: config_path = Path(td) / "config.json" with patch.dict("os.environ", {"DISPLAY": ":0"}, clear=False), patch( "diagnostics.list_input_devices", return_value=[] ), patch( "diagnostics._run_systemctl_user", return_value=_Result(returncode=0, stdout="running\n"), ): report = run_doctor(str(config_path)) self.assertEqual(report.status, "warn") results = {check.id: check for check in report.checks} self.assertEqual(results["config.load"].status, "warn") self.assertEqual(results["runtime.audio"].status, "warn") self.assertEqual(results["audio.input"].status, "warn") self.assertIn("open Settings", results["config.load"].next_step) self.assertFalse(config_path.exists()) def test_run_self_check_adds_deeper_readiness_checks(self): cfg = Config() model_path = Path("/tmp/model.gguf") with tempfile.TemporaryDirectory() as td: config_path = Path(td) / "config.json" config_path.write_text('{"config_version":1}\n', encoding="utf-8") with patch.dict("os.environ", {"DISPLAY": ":0"}, clear=False), patch( "diagnostics.load_existing", return_value=cfg ), patch("diagnostics.list_input_devices", return_value=[{"index": 1, "name": "Mic"}]), patch( "diagnostics.resolve_input_device", return_value=1 ), patch( "diagnostics.get_desktop_adapter", return_value=_FakeDesktop() ), patch( "diagnostics._run_systemctl_user", side_effect=_systemctl_side_effect( _Result(returncode=0, stdout="running\n"), _Result(returncode=0, stdout="/home/test/.config/systemd/user/aman.service\n"), _Result(returncode=0, stdout="enabled\n"), _Result(returncode=0, stdout="active\n"), ), ), patch( "diagnostics.probe_managed_model", return_value=SimpleNamespace( status="ready", path=model_path, message=f"managed editor model is ready at {model_path}", ), ), patch( "diagnostics.MODEL_DIR", model_path.parent ), patch( "diagnostics.os.access", return_value=True ), patch( "diagnostics._load_llama_bindings", return_value=(object(), object()) ), patch.dict( "sys.modules", {"faster_whisper": SimpleNamespace(WhisperModel=object())} ): report = run_self_check(str(config_path)) self.assertEqual(report.status, "ok") self.assertEqual( [check.id for check in report.checks[-5:]], [ "model.cache", "cache.writable", "service.unit", "service.state", "startup.readiness", ], ) self.assertTrue(all(check.status == "ok" for check in report.checks)) def test_run_self_check_missing_model_warns_without_downloading(self): cfg = Config() model_path = Path("/tmp/model.gguf") with tempfile.TemporaryDirectory() as td: config_path = Path(td) / "config.json" config_path.write_text('{"config_version":1}\n', encoding="utf-8") with patch.dict("os.environ", {"DISPLAY": ":0"}, clear=False), patch( "diagnostics.load_existing", return_value=cfg ), patch("diagnostics.list_input_devices", return_value=[{"index": 1, "name": "Mic"}]), patch( "diagnostics.resolve_input_device", return_value=1 ), patch( "diagnostics.get_desktop_adapter", return_value=_FakeDesktop() ), patch( "diagnostics._run_systemctl_user", side_effect=_systemctl_side_effect( _Result(returncode=0, stdout="running\n"), _Result(returncode=0, stdout="/home/test/.config/systemd/user/aman.service\n"), _Result(returncode=0, stdout="enabled\n"), _Result(returncode=0, stdout="active\n"), ), ), patch( "diagnostics.probe_managed_model", return_value=SimpleNamespace( status="missing", path=model_path, message=f"managed editor model is not cached at {model_path}", ), ) as probe_model, patch( "diagnostics.MODEL_DIR", model_path.parent ), patch( "diagnostics.os.access", return_value=True ), patch( "diagnostics._load_llama_bindings", return_value=(object(), object()) ), patch.dict( "sys.modules", {"faster_whisper": SimpleNamespace(WhisperModel=object())} ): report = run_self_check(str(config_path)) self.assertEqual(report.status, "warn") results = {check.id: check for check in report.checks} self.assertEqual(results["model.cache"].status, "warn") self.assertEqual(results["startup.readiness"].status, "warn") self.assertIn("networked connection", results["model.cache"].next_step) probe_model.assert_called_once() def test_report_json_schema_includes_status_and_next_step(self): report = DiagnosticReport( checks=[ DiagnosticCheck(id="config.load", status="warn", message="missing", next_step="open settings"), DiagnosticCheck(id="service.prereq", status="fail", message="broken", next_step="fix systemd"), ] ) payload = json.loads(report.to_json()) self.assertEqual(payload["status"], "fail") self.assertFalse(payload["ok"]) self.assertEqual(payload["checks"][0]["status"], "warn") self.assertEqual(payload["checks"][0]["next_step"], "open settings") self.assertEqual(payload["checks"][1]["hint"], "fix systemd") if __name__ == "__main__": unittest.main()