V2

test_router.py

@@ -0,0 +1,146 @@
+import unittest
+from types import SimpleNamespace
+
+import jack
+
+import router
+
+
+class FakeJackClient:
+    def __init__(self):
+        self.available_ports = set()
+        self.connections = set()
+        self.connect_calls = []
+        self.disconnect_calls = []
+        self.failed_disconnects = set()
+
+    def get_all_connections(self, destination):
+        if destination not in self.available_ports:
+            raise jack.JackError(f"missing destination: {destination}")
+        return [
+            SimpleNamespace(name=source)
+            for source, target in self.connections
+            if target == destination
+        ]
+
+    def connect(self, source, destination):
+        if source not in self.available_ports or destination not in self.available_ports:
+            raise jack.JackError("missing port")
+        self.connect_calls.append((source, destination))
+        self.connections.add((source, destination))
+
+    def disconnect(self, source, destination):
+        self.disconnect_calls.append((source, destination))
+        if (source, destination) in self.failed_disconnects:
+            raise jack.JackError("injected disconnect failure")
+        try:
+            self.connections.remove((source, destination))
+        except KeyError as error:
+            raise jack.JackError("not connected") from error
+
+
+class AudioRouterTests(unittest.TestCase):
+    def setUp(self):
+        self.jack_client = FakeJackClient()
+        self.router = router.JackRouter(jack_client=self.jack_client)
+        for port_group in self.router.cfg["ports"].values():
+            self.jack_client.available_ports.update(port_group.values())
+
+    def route_pairs(self, route_name):
+        route = self.router.cfg["routes"][route_name]
+        source = self.router.cfg["ports"][route["source"]]
+        output = self.router.cfg["ports"][route["output"]]
+        return {
+            (source["left"], output["left"]),
+            (source["right"], output["right"]),
+        }
+
+    def test_ytplayer_preset_policy(self):
+        expected = {
+            "mute_all": set(),
+            "service": set(),
+            "announcements": self.route_pairs("ytplayer_to_analog"),
+            "aux_only": self.route_pairs("ytplayer_to_analog"),
+            "video": self.route_pairs("ytplayer_to_hdmi"),
+        }
+
+        all_ytplayer_pairs = (
+            self.route_pairs("ytplayer_to_hdmi")
+            | self.route_pairs("ytplayer_to_analog")
+        )
+        for mode, expected_pairs in expected.items():
+            with self.subTest(mode=mode):
+                actual = set(self.router.routes_for_mode(mode)) & all_ytplayer_pairs
+                self.assertEqual(actual, expected_pairs)
+
+    def test_service_contains_only_mixer_to_hdmi(self):
+        self.assertEqual(
+            set(self.router.routes_for_mode("service")),
+            self.route_pairs("mixer_to_hdmi"),
+        )
+
+    def test_exclusive_source_cannot_target_both_outputs(self):
+        self.router.cfg["presets"]["video"]["ytplayer_to_analog"] = True
+        with self.assertRaisesRegex(ValueError, "exclusive source 'ytplayer'"):
+            self.router._validate_config()
+
+    def test_reconcile_connects_virtual_bus_when_ports_appear_late(self):
+        ytplayer_ports = set(self.router.cfg["ports"]["ytplayer"].values())
+        self.jack_client.available_ports -= ytplayer_ports
+
+        self.router.set_mode("video")
+        self.assertFalse(
+            self.jack_client.connections & self.route_pairs("ytplayer_to_hdmi")
+        )
+
+        self.jack_client.available_ports |= ytplayer_ports
+        self.router.reconcile()
+        self.assertEqual(
+            self.jack_client.connections & self.route_pairs("ytplayer_to_hdmi"),
+            self.route_pairs("ytplayer_to_hdmi"),
+        )
+
+    def test_switch_is_break_before_make(self):
+        self.router.set_mode("announcements")
+        analog_pairs = self.route_pairs("ytplayer_to_analog")
+        hdmi_pairs = self.route_pairs("ytplayer_to_hdmi")
+        self.assertEqual(self.jack_client.connections & analog_pairs, analog_pairs)
+
+        self.router.set_mode("video")
+        self.assertFalse(self.jack_client.connections & analog_pairs)
+        self.assertEqual(self.jack_client.connections & hdmi_pairs, hdmi_pairs)
+
+    def test_disconnect_failure_cannot_bridge_exclusive_outputs(self):
+        self.router.set_mode("announcements")
+        analog_pairs = self.route_pairs("ytplayer_to_analog")
+        hdmi_pairs = self.route_pairs("ytplayer_to_hdmi")
+        self.jack_client.failed_disconnects = set(analog_pairs)
+
+        self.router.set_mode("video")
+
+        self.assertEqual(self.jack_client.connections & analog_pairs, analog_pairs)
+        self.assertFalse(self.jack_client.connections & hdmi_pairs)
+
+    def test_repeated_same_mode_does_not_break_working_links(self):
+        self.router.set_mode("video")
+        desired = set(self.router.routes_for_mode("video"))
+        disconnect_count = len(self.jack_client.disconnect_calls)
+        connect_count = len(self.jack_client.connect_calls)
+
+        self.router.set_mode("video")
+
+        self.assertTrue(desired <= self.jack_client.connections)
+        self.assertEqual(len(self.jack_client.disconnect_calls), disconnect_count + 8)
+        self.assertEqual(len(self.jack_client.connect_calls), connect_count)
+
+    def test_connect_uses_port_names_to_avoid_duplicate_link(self):
+        source, destination = next(iter(self.route_pairs("ytplayer_to_hdmi")))
+        self.jack_client.connections.add((source, destination))
+
+        self.router.connect(source, destination)
+
+        self.assertEqual(self.jack_client.connect_calls, [])
+
+
+if __name__ == "__main__":
+    unittest.main()