Hold-to-fix for the Lighting PC and the LiveStream (OBS) PC
- OBS: start OBS on the LiveStream PC if it is not running, then reconnect Companion. If the PC does not answer, send Wake-on-LAN and show WAKING PC. - ARENA: start Resolume Arena on the Lighting PC if it is not running, then reconnect Companion. VISTA SILENT: start Vista if it is not running. A running OBS, Arena or Vista is never closed. - MIDI (LIGHT PC): restart the rtpMIDI service; it reconnects to its saved peers (NA Ingest, NA ProP, NA Tracks). - Fixes may set "reconnect" (restart the check's Companion connection after the command succeeds); hosts may set "mac"/"wake_broadcast" for Wake-on-LAN. - windows-setup.ps1 finds OBS on any drive and Vista, and leaves Wake-on-LAN alone when it is already on (changing it can reset the adapter). - Tests stub the fix executors so a hold in a test never reaches a machine. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -465,6 +465,8 @@ class TestFixes(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.snap = healthy_snapshot(SUNDAY)
|
||||
self.engine = sp.Engine(CONFIG)
|
||||
# Safety net: a hold in these tests must never reach a real machine.
|
||||
self.engine.executor = self.engine.poe_executor = lambda *a: True
|
||||
|
||||
def plan(self, tile):
|
||||
run_steps(self.engine, self.snap, SUNDAY, 4)
|
||||
@@ -492,6 +494,59 @@ class TestFixes(unittest.TestCase):
|
||||
self.assertEqual(CONFIG["hosts"][fix["host"]]["ssh"], "media@10.6.7.13")
|
||||
self.assertIn('schtasks /Run /TN "\\StatusPanel\\Start BibleApp"', fix["run"]) # cmd.exe, not sh
|
||||
|
||||
def test_lighting_and_livestream_pcs(self):
|
||||
self.snap.conns["OBS-Youtube"]["category"] = "error"
|
||||
self.snap.conns["arena"]["category"] = "error"
|
||||
self.snap.artnet_since, self.snap.artnet_last = SUNDAY - 100, {"10.1.10.206": SUNDAY - 100}
|
||||
self.snap.midi["NA Ingest"]["peers"] = [p for p in self.snap.midi["NA Ingest"]["peers"]
|
||||
if p["name"] != "Lighting PC"]
|
||||
self.snap.mdns = self.snap.mdns - {"Lighting PC"}
|
||||
self.assertEqual(self.plan("obs"), [("fix", "obs")])
|
||||
self.assertEqual(self.plan("arena"), [("fix", "arena")])
|
||||
self.assertEqual(self.plan("apps"), [("fix", "vista")])
|
||||
self.assertEqual(self.plan("midi"), [("fix", "rtpmidi")]) # two checks, one restart
|
||||
for fix_id in ("obs", "arena", "vista"): # never closes a running app
|
||||
self.assertTrue(CONFIG["fixes"][fix_id]["run"].startswith("tasklist "), fix_id)
|
||||
|
||||
def test_an_off_pc_is_woken(self):
|
||||
from unittest import mock
|
||||
woke = []
|
||||
done = mock.Mock(returncode=255, stdout="", stderr="ssh: connect to host 10.1.9.126 port 22: timed out")
|
||||
with mock.patch.object(sp.subprocess, "run", return_value=done), \
|
||||
mock.patch.object(sp, "wake_on_lan", side_effect=lambda mac, bc=None: woke.append((mac, bc))):
|
||||
self.assertEqual(sp.execute_fix(CONFIG, "obs", {"conn": "OBS-Youtube"}), sp.WAKING)
|
||||
self.assertEqual(woke, [("6c:24:08:d7:fb:57", "10.1.255.255")])
|
||||
self.assertFalse(sp.execute_fix(CONFIG, "arena", {"conn": "arena"})) # no MAC: just fails
|
||||
self.assertEqual(len(woke), 1)
|
||||
# the button says so
|
||||
self.snap.conns["OBS-Youtube"]["category"] = "error"
|
||||
self.engine.executor = lambda cfg, fix_id, chk: sp.WAKING
|
||||
_, t = run_steps(self.engine, self.snap, SUNDAY, 4)
|
||||
self.engine.run_fixes("obs", self.snap, None, t)
|
||||
time.sleep(0.2)
|
||||
self.assertEqual(self.engine.step(self.snap, time.time())["obs"]["x"], "OBS\nWAKING PC\nOBS")
|
||||
|
||||
def test_magic_packet(self):
|
||||
from unittest import mock
|
||||
with mock.patch.object(sp.socket, "socket") as sock:
|
||||
packet = sp.wake_on_lan("6c:24:08:d7:fb:57", "10.1.255.255")
|
||||
self.assertEqual(packet, b"\xff" * 6 + bytes.fromhex("6c2408d7fb57") * 16)
|
||||
sent = [c.args[1] for c in sock.return_value.__enter__.return_value.sendto.call_args_list]
|
||||
self.assertEqual(sent, [("10.1.255.255", 9), ("255.255.255.255", 9)])
|
||||
|
||||
def test_companion_reconnects_after_the_app_starts(self):
|
||||
from unittest import mock
|
||||
comp = mock.Mock()
|
||||
comp.connections.return_value = [{"label": "arena", "id": "abc"}]
|
||||
ok = mock.Mock(returncode=0, stdout="", stderr="")
|
||||
with mock.patch.object(sp.subprocess, "run", return_value=ok), \
|
||||
mock.patch.object(sp, "Companion", return_value=comp):
|
||||
self.assertTrue(sp.execute_fix(CONFIG, "arena", {"conn": "arena"}))
|
||||
comp.restart_connection.assert_called_once_with("abc")
|
||||
comp.reset_mock()
|
||||
self.assertTrue(sp.execute_fix(CONFIG, "vista", {"type": "artnet"})) # no "reconnect"
|
||||
comp.restart_connection.assert_not_called()
|
||||
|
||||
def test_unreachable_mac_is_not_a_verdict(self):
|
||||
self.snap.remote["tracks\0pgrep -xq Live"] = None
|
||||
self.assertEqual(self.plan("apps"), [])
|
||||
@@ -632,7 +687,8 @@ class TestFixes(unittest.TestCase):
|
||||
self.snap.vars[("audinate-dantecontroller", "devices")] = json.dumps(devices)
|
||||
self.assertEqual(self.plan("dante"), [("fix", "dvs_ingest")])
|
||||
# A failure with no fix must not claim there is nothing wrong.
|
||||
self.snap.midi["NA Ingest"]["peers"] = []
|
||||
self.snap.midi["NA Ingest"]["peers"] = [p for p in self.snap.midi["NA Ingest"]["peers"]
|
||||
if p["name"] != "SuperRack-LiveBox"]
|
||||
_, t = run_steps(self.engine, self.snap, SUNDAY, 4)
|
||||
self.engine.run_fixes("midi", self.snap, None, t)
|
||||
self.assertEqual(self.engine.notes["midi"]["text"], "NO AUTO\nFIX")
|
||||
|
||||
Reference in New Issue
Block a user