feat(m9): systemd --user trigger modes + game-launch watcher — 0.23.0
D6 trigger modes, no root: - core/service.py: write/enable `systemd --user` units; apply_mode(manual/ always-on/game-launch) reconciles the recorder + watcher services; status(). - core/watcher.py + `rigdoctor watch`: poll Steam RunningAppID, auto-bracket a focused capture (D12 zero-config fallback; wrapper stays primary). - CLI `rigdoctor service status|mode`; config `trigger_mode`. - GUI Settings: "Recording trigger" dropdown (Apply runs apply_mode off-thread). - Tests for unit generation, mode reconciliation, watcher transitions/parse. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,58 @@
|
||||
"""Tests for the M9 systemd --user trigger-mode service manager."""
|
||||
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
from rigdoctor.core import service
|
||||
|
||||
|
||||
class UnitTextTests(unittest.TestCase):
|
||||
def test_unit_text_has_required_sections(self):
|
||||
txt = service.unit_text("RigDoctor recorder", ["record", "run"])
|
||||
self.assertIn("[Unit]", txt)
|
||||
self.assertIn("[Service]", txt)
|
||||
self.assertIn("ExecStart=", txt)
|
||||
self.assertIn("record run", txt)
|
||||
self.assertIn("WantedBy=default.target", txt)
|
||||
|
||||
|
||||
class ApplyModeTests(unittest.TestCase):
|
||||
def test_unknown_mode_rejected(self):
|
||||
ok, msg = service.apply_mode("turbo")
|
||||
self.assertFalse(ok)
|
||||
self.assertIn("Unknown", msg)
|
||||
|
||||
def test_no_systemd_saves_mode_but_reports(self):
|
||||
with mock.patch.object(service, "available", return_value=False), \
|
||||
mock.patch.object(service.config, "update_config") as update:
|
||||
ok, msg = service.apply_mode("always-on")
|
||||
self.assertFalse(ok)
|
||||
self.assertIn("available", msg.lower())
|
||||
update.assert_called_once_with(trigger_mode="always-on")
|
||||
|
||||
def test_always_on_enables_recorder_disables_watch(self):
|
||||
calls = []
|
||||
with mock.patch.object(service, "available", return_value=True), \
|
||||
mock.patch.object(service, "install_units"), \
|
||||
mock.patch.object(service, "_enable", side_effect=lambda n: calls.append(("enable", n)) or (0, "")), \
|
||||
mock.patch.object(service, "_disable", side_effect=lambda n: calls.append(("disable", n)) or (0, "")), \
|
||||
mock.patch.object(service.config, "update_config"):
|
||||
ok, _ = service.apply_mode("always-on")
|
||||
self.assertTrue(ok)
|
||||
self.assertIn(("enable", service.RECORDER_UNIT), calls)
|
||||
self.assertIn(("disable", service.WATCH_UNIT), calls)
|
||||
|
||||
def test_manual_disables_both(self):
|
||||
disabled = []
|
||||
with mock.patch.object(service, "available", return_value=True), \
|
||||
mock.patch.object(service, "install_units"), \
|
||||
mock.patch.object(service, "_enable", return_value=(0, "")), \
|
||||
mock.patch.object(service, "_disable", side_effect=lambda n: disabled.append(n) or (0, "")), \
|
||||
mock.patch.object(service.config, "update_config"):
|
||||
ok, _ = service.apply_mode("manual")
|
||||
self.assertTrue(ok)
|
||||
self.assertEqual(set(disabled), {service.RECORDER_UNIT, service.WATCH_UNIT})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,69 @@
|
||||
"""Tests for the M9/D12 game-launch watcher (RunningAppID parse + transitions)."""
|
||||
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
from rigdoctor.core import watcher
|
||||
|
||||
_REGISTRY = """"Registry"
|
||||
{
|
||||
\t"HKCU"
|
||||
\t{
|
||||
\t\t"Software"
|
||||
\t\t{
|
||||
\t\t\t"Valve"
|
||||
\t\t\t{
|
||||
\t\t\t\t"Steam"
|
||||
\t\t\t\t{
|
||||
\t\t\t\t\t"RunningAppID"\t\t"%s"
|
||||
\t\t\t\t}
|
||||
\t\t\t}
|
||||
\t\t}
|
||||
\t}
|
||||
}
|
||||
"""
|
||||
|
||||
|
||||
class TransitionTests(unittest.TestCase):
|
||||
def test_transitions(self):
|
||||
self.assertEqual(watcher.transition(0, 570), "start")
|
||||
self.assertEqual(watcher.transition(570, 0), "stop")
|
||||
self.assertIsNone(watcher.transition(570, 570))
|
||||
self.assertIsNone(watcher.transition(0, 0))
|
||||
|
||||
|
||||
class FindKeyTests(unittest.TestCase):
|
||||
def test_case_insensitive_nested(self):
|
||||
data = {"Registry": {"HKCU": {"steam": {"runningappid": "42"}}}}
|
||||
self.assertEqual(watcher._find_key(data, "RunningAppID"), "42")
|
||||
|
||||
def test_missing(self):
|
||||
self.assertIsNone(watcher._find_key({"a": {"b": "c"}}, "RunningAppID"))
|
||||
|
||||
|
||||
class RunningAppIdTests(unittest.TestCase):
|
||||
def _with_registry(self, content):
|
||||
d = tempfile.mkdtemp()
|
||||
path = Path(d) / "registry.vdf"
|
||||
path.write_text(content)
|
||||
return path
|
||||
|
||||
def test_reads_running_appid(self):
|
||||
path = self._with_registry(_REGISTRY % "570")
|
||||
with mock.patch.object(watcher, "_registry_path", return_value=path):
|
||||
self.assertEqual(watcher.running_appid(), 570)
|
||||
|
||||
def test_zero_when_idle(self):
|
||||
path = self._with_registry(_REGISTRY % "0")
|
||||
with mock.patch.object(watcher, "_registry_path", return_value=path):
|
||||
self.assertEqual(watcher.running_appid(), 0)
|
||||
|
||||
def test_zero_when_no_registry(self):
|
||||
with mock.patch.object(watcher, "_registry_path", return_value=None):
|
||||
self.assertEqual(watcher.running_appid(), 0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user