147 lines
5.5 KiB
Python
147 lines
5.5 KiB
Python
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()
|