A Dante device Companion's module never listed before gets the Dante Controller connection restarted once

The module only reads a device's channels when it starts, so a receiver added later is missing
from the crosspoint lists until the connection restarts (the two AD4Q receivers today). The
watchdog now remembers every Dante device it has seen in state.json and, when a newcomer appears,
restarts that connection once: never during a live service, at most every ten minutes, and never
for a device that merely dropped out and came back.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
2026-10-18 06:38:27 -04:00
co-authored by Claude Fable 5.1
parent efd9b973eb
commit 87bfe7c162
2 changed files with 92 additions and 0 deletions
+49
View File
@@ -2255,3 +2255,52 @@ class TestCoreMIDI(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
class TestDanteNewcomers(unittest.TestCase):
"""A Dante device the module never listed before gets the Dante Controller connection restarted once."""
def setUp(self):
self.snap = healthy_snapshot(time.time())
self.engine = sp.Engine(CONFIG, None)
self.restarts = []
me = self
class Comp:
def restart_connection(self, cid):
me.restarts.append(cid); return True
self.comp = Comp()
self.label = sp.dante_label(CONFIG)
self.snap.conns[self.label] = {"id": "dante-1", "enabled": True, "category": "good"}
def devices(self, names):
self.snap.vars[(self.label, "devices")] = json.dumps(names)
def test_first_list_is_remembered_then_a_newcomer_restarts_once(self):
now = time.time()
self.devices(["A", "B"])
self.assertFalse(sp.watch_dante_newcomers(self.snap, CONFIG, self.comp, self.engine, now))
self.assertEqual(self.engine.state.get("dante_known"), ["A", "B"])
self.assertFalse(sp.watch_dante_newcomers(self.snap, CONFIG, self.comp, self.engine, now + 60)) # nothing new
self.devices(["A", "B", "BOH-AD4Q-A"])
self.assertTrue(sp.watch_dante_newcomers(self.snap, CONFIG, self.comp, self.engine, now + 120))
self.assertEqual(self.restarts, ["dante-1"])
self.assertEqual(self.engine.state.get("dante_known"), ["A", "B", "BOH-AD4Q-A"])
self.devices(["A", "B", "BOH-AD4Q-A", "X"])
self.assertFalse(sp.watch_dante_newcomers(self.snap, CONFIG, self.comp, self.engine, now + 180)) # ten-minute cooldown
self.assertTrue(sp.watch_dante_newcomers(self.snap, CONFIG, self.comp, self.engine, now + 800))
self.assertEqual(len(self.restarts), 2)
self.devices(["A", "B"]) # devices going away never restart
self.assertFalse(sp.watch_dante_newcomers(self.snap, CONFIG, self.comp, self.engine, now + 2000))
self.devices(["A", "B", "BOH-AD4Q-A"]) # nor coming back
self.assertFalse(sp.watch_dante_newcomers(self.snap, CONFIG, self.comp, self.engine, now + 2600))
def test_waits_for_the_service_to_end(self):
now = time.time()
self.devices(["A"])
sp.watch_dante_newcomers(self.snap, CONFIG, self.comp, self.engine, now)
self.devices(["A", "NEW"])
self.engine.any_recording = True
self.assertFalse(sp.watch_dante_newcomers(self.snap, CONFIG, self.comp, self.engine, now + 100))
self.engine.any_recording = False
self.assertTrue(sp.watch_dante_newcomers(self.snap, CONFIG, self.comp, self.engine, now + 200))