import json import queue import signal import threading import unittest from types import SimpleNamespace from unittest import mock import homeassistant_discovery import player_service as player class FakePublishClient: def __init__(self): self.published = [] self.subscriptions = [] self.acks = [] def publish(self, topic, payload, qos=0, retain=False): self.published.append((topic, payload, qos, retain)) return SimpleNamespace(rc=0) def subscribe(self, topic, qos=0): self.subscriptions.append((topic, qos)) return (0, 1) def ack(self, message_id, qos): self.acks.append((message_id, qos)) return player.mqtt.MQTT_ERR_SUCCESS class FakeProcess: def __init__(self, returncode=None): self.returncode = returncode self.terminate_calls = 0 self.kill_calls = 0 self.wait_calls = [] def poll(self): return self.returncode def terminate(self): self.terminate_calls += 1 self.returncode = 0 def kill(self): self.kill_calls += 1 self.returncode = -9 def wait(self, timeout=None): self.wait_calls.append(timeout) return self.returncode def playback(generation=1, returncode=None, path="/media/video.mp4", slot=1): return player.PlaybackContext( generation=generation, process=FakeProcess(returncode), path=path, slot=slot, stop_event=threading.Event(), ) class PlayerServiceTests(unittest.TestCase): def setUp(self): self.original_client = player.client self.original_playback = player.current_playback self.original_queue = player.command_queue self.original_mqtt_generation = player.mqtt_connection_generation player.current_playback = None player.command_queue = queue.Queue() player.mqtt_connection_generation = 0 player.shutdown_event.clear() def tearDown(self): player.client = self.original_client player.current_playback = self.original_playback player.command_queue = self.original_queue player.mqtt_connection_generation = self.original_mqtt_generation player.shutdown_event.clear() def test_decode_command_rejects_non_objects_and_bad_envelopes(self): bad_payloads = ("", "not json", "null", "[]", '"text"', "1", "{}") with mock.patch.object(player, "log"): for payload in bad_payloads: with self.subTest(payload=payload): self.assertIsNone(player.decode_command(payload)) self.assertEqual( player.decode_command('{"action":"stop"}'), {"action": "stop"}, ) def test_decode_command_contains_parser_resource_errors(self): huge_integer = '{"action":"stop","value":' + ("9" * 5000) + "}" deeply_nested = ("[" * 1200) + ("0") + ("]" * 1200) with mock.patch.object(player, "log"): self.assertIsNone(player.decode_command(huge_integer)) self.assertIsNone(player.decode_command(deeply_nested)) def test_malformed_play_fields_do_not_escape(self): fake_client = FakePublishClient() player.client = fake_client malformed = ( {"action": "play", "path": 123}, {"action": "play", "path": {"bad": True}}, {"action": "play", "slot": float("inf")}, ) with mock.patch.object(player, "log"): for command in malformed: with self.subTest(command=command): player.handle_cmd(command) status_payloads = [ json.loads(payload) for topic, payload, _, _ in fake_client.published if topic == player.TOPIC_STATUS ] self.assertEqual(len(status_payloads), len(malformed)) self.assertTrue(all(item["state"] == "error" for item in status_payloads)) def test_on_message_rejects_bad_payload_then_queues_valid_command(self): fake_client = FakePublishClient() player.client = fake_client with mock.patch.object(player, "log"): player.on_message( None, None, SimpleNamespace( topic=player.TOPIC_CMD, payload=b"[]", mid=11, qos=1, ), ) player.on_message( None, None, SimpleNamespace(topic=player.TOPIC_CMD, payload=b'{"action":"stop"}'), ) self.assertEqual(fake_client.acks, [(11, 1)]) self.assertEqual(player.command_queue.qsize(), 1) self.assertEqual( player.command_queue.get_nowait().data, {"action": "stop"}, ) def test_on_message_contains_unexpected_callback_exception(self): message = SimpleNamespace(topic=player.TOPIC_CMD, payload=b'{"action":"stop"}') with mock.patch.object(player, "decode_command", side_effect=RuntimeError("boom")): with mock.patch.object(player, "log_exception") as logged: player.on_message(None, None, message) logged.assert_called_once() def test_repeated_commands_are_preserved_in_order(self): command = {"action": "stop"} for _ in range(3): self.assertTrue(player.enqueue_command(command)) self.assertEqual( [player.command_queue.get_nowait().data for _ in range(3)], [command, command, command], ) def test_qos1_command_is_acked_only_after_main_thread_execution(self): fake_client = FakePublishClient() player.client = fake_client message = SimpleNamespace( topic=player.TOPIC_CMD, payload=b'{"action":"stop"}', mid=42, qos=1, ) player.on_message(None, None, message) self.assertEqual(fake_client.acks, []) with mock.patch.object(player, "handle_cmd") as handled: player.process_pending_commands() handled.assert_called_once_with({"action": "stop"}) self.assertEqual(fake_client.acks, [(42, 1)]) def test_queued_command_cannot_ack_a_reused_mid_after_reconnect(self): fake_client = FakePublishClient() player.client = fake_client player.mqtt_connection_generation = 3 player.enqueue_command( {"action": "stop"}, message_id=42, qos=1, connection_generation=3, ) player.mqtt_connection_generation = 4 with mock.patch.object(player, "handle_cmd") as handled: with mock.patch.object(player, "log"): player.process_pending_commands() handled.assert_called_once_with({"action": "stop"}) self.assertEqual(fake_client.acks, []) def test_status_topics_qos_retain_and_payload_are_unchanged(self): fake_client = FakePublishClient() player.client = fake_client active = playback(path="/media/example.mp4", slot=4) player.publish_status("playing", playback=active, label="Example") self.assertEqual(fake_client.published[0][0], "youtube/player/status") self.assertEqual(fake_client.published[0][2:], (1, False)) self.assertEqual( json.loads(fake_client.published[0][1]), { "state": "playing", "path": "/media/example.mp4", "slot": 4, "label": "Example", }, ) self.assertEqual( fake_client.published[1], ("youtube/player/state", "playing", 0, True), ) def test_stale_generation_cannot_stop_current_playback(self): old = playback(generation=1) current = playback(generation=2) player.current_playback = current with mock.patch.object(player, "_remove_mpv_socket"): self.assertFalse(player.stop_mpv(expected=old)) self.assertIs(player.current_playback, current) self.assertEqual(current.process.terminate_calls, 0) self.assertFalse(current.stop_event.is_set()) def test_stale_generation_cannot_publish_telemetry(self): fake_client = FakePublishClient() player.client = fake_client old = playback(generation=1) player.current_playback = playback(generation=2) player.publish_telemetry( old, pos=1, dur=2, paused=False, title=None, filename=None, playlist_pos=0, playlist_count=1, ) self.assertEqual(fake_client.published, []) def test_stop_failure_preserves_error_clear_idle_publish_order(self): class FailingTerminateProcess(FakeProcess): def terminate(self): raise OSError("cannot terminate") fake_client = FakePublishClient() player.client = fake_client active = playback() active.process = FailingTerminateProcess(returncode=None) player.current_playback = active with mock.patch.object(player, "_remove_mpv_socket"): with mock.patch.object(player, "log"): player.stop_mpv("stop") self.assertEqual( [topic for topic, _, _, _ in fake_client.published], [ player.TOPIC_STATUS, player.TOPIC_STATE, player.TELEMETRY_TOPIC, player.TOPIC_STATUS, player.TOPIC_STATE, ], ) self.assertEqual(json.loads(fake_client.published[0][1])["state"], "error") self.assertIsNone(json.loads(fake_client.published[2][1])["slot"]) self.assertEqual(json.loads(fake_client.published[3][1])["state"], "idle") def test_unstoppable_process_remains_owned_and_blocks_duplicates(self): class UnstoppableProcess(FakeProcess): def terminate(self): raise OSError("cannot terminate") def kill(self): raise OSError("cannot kill") fake_client = FakePublishClient() player.client = fake_client active = playback() active.process = UnstoppableProcess(returncode=None) player.current_playback = active with mock.patch.object(player, "_remove_mpv_socket") as remove_socket: with mock.patch.object(player, "log"): self.assertFalse(player.stop_mpv("stop")) self.assertIs(player.current_playback, active) remove_socket.assert_not_called() self.assertEqual( [json.loads(payload)["state"] for topic, payload, _, _ in fake_client.published if topic == player.TOPIC_STATUS], ["error"], ) def test_mpv_is_pinned_to_virtual_pipewire_bus(self): fake_process = FakeProcess(returncode=None) fake_thread = SimpleNamespace( name="fake-listener", start=mock.Mock(), join=mock.Mock(), is_alive=mock.Mock(return_value=False), ) with mock.patch.object(player.os.path, "exists", return_value=True): with mock.patch.object(player.subprocess, "Popen", return_value=fake_process) as popen: with mock.patch.object(player.threading, "Thread", return_value=fake_thread): with mock.patch.object(player, "_remove_mpv_socket"): player.handle_play({"action": "play", "path": "/media/video.mp4"}) args = popen.call_args.args[0] self.assertIn("--audio-device=pipewire/ytplayer_bus", args) self.assertIn("--audio-client-name=YTPlayer", args) self.assertIn("--audio-channels=stereo", args) self.assertIn("--audio-fallback-to-null=no", args) def test_worker_construction_failure_reaps_started_mpv(self): fake_client = FakePublishClient() player.client = fake_client fake_process = FakeProcess(returncode=None) with mock.patch.object(player.os.path, "exists", return_value=True): with mock.patch.object(player.subprocess, "Popen", return_value=fake_process): with mock.patch.object( player.threading, "Thread", side_effect=RuntimeError("cannot construct worker"), ): with mock.patch.object(player, "_remove_mpv_socket"): with mock.patch.object(player, "log"): player.handle_play( {"action": "play", "path": "/media/video.mp4"}, ) self.assertEqual(fake_process.terminate_calls, 1) self.assertIsNone(player.current_playback) status_payloads = [ json.loads(payload) for topic, payload, _, _ in fake_client.published if topic == player.TOPIC_STATUS ] self.assertEqual(status_payloads[-1]["state"], "error") def test_mpv_error_preserves_historical_error_then_idle_sequence(self): fake_client = FakePublishClient() player.client = fake_client failed = playback(returncode=4) player.current_playback = failed with mock.patch.object(player, "_remove_mpv_socket"): with mock.patch.object(player, "log"): player.poll_current_playback() state_messages = [ (topic, json.loads(payload) if topic == player.TOPIC_STATUS else payload) for topic, payload, _, _ in fake_client.published if topic in (player.TOPIC_STATUS, player.TOPIC_STATE) ] self.assertEqual( [ item[1]["state"] if item[0] == player.TOPIC_STATUS else item[1] for item in state_messages ], ["error", "error", "idle", "idle"], ) self.assertEqual(state_messages[-2][1]["reason"], "mpv_error") def test_shutdown_stops_command_drain_without_losing_pending_items(self): player.command_queue.put_nowait({"action": "first"}) player.command_queue.put_nowait({"action": "second"}) def handle_one(command): player.shutdown_event.set() with mock.patch.object(player, "handle_cmd", side_effect=handle_one) as handled: player.process_pending_commands() handled.assert_called_once_with({"action": "first"}) self.assertEqual(player.command_queue.get_nowait(), {"action": "second"}) def test_main_uses_nonblocking_mqtt_connect(self): class FakeMqttClient(FakePublishClient): def __init__(self): super().__init__() self.calls = [] self.suppress_exceptions = False def username_pw_set(self, *args): self.calls.append(("username_pw_set", args)) def reconnect_delay_set(self, **kwargs): self.calls.append(("reconnect_delay_set", kwargs)) def manual_ack_set(self, enabled): self.calls.append(("manual_ack_set", enabled)) def max_queued_messages_set(self, count): self.calls.append(("max_queued_messages_set", count)) def will_set(self, *args, **kwargs): self.calls.append(("will_set", args, kwargs)) def connect_async(self, *args, **kwargs): self.calls.append(("connect_async", args, kwargs)) def loop_start(self): self.calls.append(("loop_start",)) player.shutdown_event.set() return player.mqtt.MQTT_ERR_SUCCESS def disconnect(self): self.calls.append(("disconnect",)) def loop_stop(self): self.calls.append(("loop_stop",)) fake_client = FakeMqttClient() player.client = fake_client with mock.patch.object(signal, "signal"): with mock.patch.object(player, "stop_mpv"): player.main() call_names = [call[0] for call in fake_client.calls] self.assertIn("connect_async", call_names) self.assertIn("manual_ack_set", call_names) self.assertNotIn("connect", call_names) self.assertNotIn("disconnect", call_names) self.assertLess(call_names.index("connect_async"), call_names.index("loop_start")) def test_discovery_failure_cannot_escape_on_connect(self): fake_client = FakePublishClient() player.client = fake_client with mock.patch.object(player, "publish_discovery", side_effect=TypeError("bad discovery")): with mock.patch.object(player, "log_exception") as logged: player.on_connect(fake_client, None, None, 0) self.assertEqual(fake_client.subscriptions, [(player.TOPIC_CMD, 1)]) logged.assert_called_once() self.assertEqual(fake_client.published[0][0], player.TOPIC_STATUS) def test_reconnect_preserves_historical_idle_payload_while_playing(self): fake_client = FakePublishClient() player.client = fake_client player.current_playback = playback(path="/media/live.mp4", slot=7) with mock.patch.object(player, "publish_discovery"): with mock.patch.object(player, "log"): player.on_connect(fake_client, None, None, 0) self.assertEqual(fake_client.subscriptions, [(player.TOPIC_CMD, 1)]) self.assertEqual( json.loads(fake_client.published[0][1]), { "state": "idle", "path": "/media/live.mp4", "slot": 7, "message": "player online", }, ) self.assertEqual(fake_client.published[1], (player.TOPIC_STATE, "idle", 0, True)) def test_pause_status_contract_is_preserved_when_ipc_fails(self): fake_client = FakePublishClient() player.client = fake_client player.current_playback = playback() with mock.patch.object(player, "mpv_ipc", return_value=False): with mock.patch.object(player, "log"): player.handle_cmd({"action": "pause"}) self.assertEqual(json.loads(fake_client.published[0][1])["state"], "paused") self.assertEqual(fake_client.published[1][1], "paused") class HomeAssistantDiscoveryTests(unittest.TestCase): def test_discovery_contract_is_unchanged(self): client = FakePublishClient() homeassistant_discovery.publish_discovery( client, discovery_prefix="homeassistant", device_id="ytplayer", device_name="YT Player", telemetry_topic="youtube/player/telemetry", status_topic="youtube/player/status", ) self.assertEqual(len(client.published), 9) self.assertTrue( all(qos == 1 and retain is True for _, _, qos, retain in client.published) ) self.assertEqual( [topic for topic, _, _, _ in client.published], [ f"homeassistant/sensor/ytplayer/{entity}/config" for entity in ( "position", "duration", "remaining", "position_fmt", "duration_fmt", "remaining_fmt", "title", "paused", "slot", ) ], ) first_payload = json.loads(client.published[0][1]) self.assertEqual(first_payload["state_topic"], "youtube/player/telemetry") self.assertEqual(first_payload["availability_topic"], "youtube/player/status") self.assertEqual(first_payload["unit_of_measurement"], "s") if __name__ == "__main__": unittest.main()