import importlib import json import os import sys import tempfile import unittest from pathlib import Path from unittest import mock from controller.config import ConfigError, load_config from controller.contracts import ( PLAYER_COMMAND_QOS, PLAYER_COMMAND_RETAIN, SLOT_IDS, Topics, ) from controller.ha import HA_DEVICE, SlotAnnouncement, discovery_messages, slot_messages class ContractCharacterizationTests(unittest.TestCase): def test_player_and_legacy_topic_defaults_are_preserved(self): topics = Topics() self.assertEqual(topics.player_command, "youtube/player/cmd") self.assertEqual(topics.player_status, "youtube/player/status") self.assertEqual(topics.player_state, "youtube/player/state") self.assertEqual(topics.player_telemetry, "youtube/player/telemetry") self.assertEqual(topics.controller_command, "youtube/controller/cmd") self.assertEqual(topics.ingest_progress, "youtube/ingest/progress") self.assertEqual(topics.playback_base, "youtube/media/playback") self.assertEqual(topics.router_mode_set, "audio/player/mode/set") self.assertEqual(topics.router_mode_state, "audio/player/mode/state") self.assertEqual(topics.controller_result, "youtube/controller/v2/events") self.assertEqual(PLAYER_COMMAND_QOS, 1) self.assertIs(PLAYER_COMMAND_RETAIN, False) self.assertEqual(SLOT_IDS, tuple(range(1, 11))) def test_existing_ha_discovery_ids_payloads_and_retain_are_preserved(self): config = load_config(environ={}) messages = list(discovery_messages(config.topics, config.homeassistant)) self.assertEqual(len(messages), 30) topic, payload, qos, retain = messages[0] self.assertEqual(topic, "homeassistant/sensor/yt_slot_1_title/config") self.assertEqual(qos, 0) self.assertIs(retain, True) decoded = json.loads(payload) self.assertEqual(decoded["unique_id"], "yt_slot_1_title") self.assertEqual(decoded["state_topic"], "youtube/media/playback/1/title") self.assertEqual(decoded["device"], HA_DEVICE) self.assertEqual(messages[-1][0], "homeassistant/sensor/yt_slot_10_count/config") def test_slot_topics_remain_plain_retained_values_and_clear_item_tail(self): announcement = SlotAnnouncement( slot=4, title="Service", state="ready", count=2, items=("One", "Two"), ) messages = list(slot_messages(announcement, Topics(), clear_items_through=4)) self.assertEqual( messages, [ ("youtube/media/playback/4/title", "Service", 0, True), ("youtube/media/playback/4/state", "ready", 0, True), ("youtube/media/playback/4/count", "2", 0, True), ("youtube/media/playback/4/items/1", "One", 0, True), ("youtube/media/playback/4/items/2", "Two", 0, True), ("youtube/media/playback/4/items/3", "", 0, True), ("youtube/media/playback/4/items/4", "", 0, True), ], ) class ConfigurationTests(unittest.TestCase): def test_safe_router_defaults_are_finite_and_service_usable(self): config = load_config(environ={}) self.assertEqual(config.router.required_mode, "video") self.assertEqual(config.router.confirmation_timeout_seconds, 3.0) self.assertEqual(config.router.when_state_unknown, "attempt") self.assertIs(config.router.continue_on_confirmation_failure, True) self.assertIsNone(config.router.fallback_restore_mode) self.assertEqual(config.storage.uncertain_review_after_seconds, 900.0) self.assertEqual(config.mqtt.publish_ack_timeout_seconds, 2.0) self.assertEqual(config.mqtt.router_echo_grace_seconds, 2.0) def test_documented_environment_overrides_apply(self): env = { "MQTT_HOST": "broker", "MQTT_PORT": "2883", "MQTT_USERNAME": "user", "MQTT_PASSWORD": "pass", "YTPLAYER_MEDIA_BASE": "/media-test", "YTPLAYER_DATA_BASE": "/data-test", "YTPLAYER_COOKIE_PATH": "", "PORT": "9000", } config = load_config(environ=env) self.assertEqual(config.mqtt.host, "broker") self.assertEqual(config.mqtt.port, 2883) self.assertEqual(config.mqtt.username, "user") self.assertEqual(config.mqtt.password, "pass") self.assertEqual(config.storage.media_base, Path("/media-test")) self.assertEqual(config.storage.data_base, Path("/data-test")) self.assertIsNone(config.downloads.cookie_path) self.assertEqual(config.web.port, 9000) def test_blank_optional_mqtt_credentials_are_treated_as_unset(self): config = load_config( environ={ "MQTT_USERNAME": "", "MQTT_PASSWORD": " ", } ) self.assertIsNone(config.mqtt.username) self.assertIsNone(config.mqtt.password) def test_blank_compose_credentials_do_not_mask_yaml_credentials(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "config.yaml" path.write_text( "mqtt:\n username: yaml-user\n password: yaml-pass\n", encoding="utf-8", ) config = load_config( path, environ={"MQTT_USERNAME": "", "MQTT_PASSWORD": ""}, ) self.assertEqual(config.mqtt.username, "yaml-user") self.assertEqual(config.mqtt.password, "yaml-pass") def test_slot_count_cannot_break_stable_identities(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "config.yaml" path.write_text("storage:\n slot_count: 9\n", encoding="utf-8") with self.assertRaisesRegex(ConfigError, "fixed at 10"): load_config(path, environ={}) def test_invalid_router_unknown_state_policy_is_rejected(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "config.yaml" path.write_text("router:\n when_state_unknown: guess\n", encoding="utf-8") with self.assertRaisesRegex(ConfigError, "attempt, continue, or reject"): load_config(path, environ={}) def test_non_finite_timeouts_are_rejected(self): for value in (".nan", ".inf", "-.inf"): with self.subTest(value=value), tempfile.TemporaryDirectory() as directory: path = Path(directory) / "config.yaml" path.write_text( f"router:\n confirmation_timeout_seconds: {value}\n", encoding="utf-8", ) with self.assertRaisesRegex(ConfigError, "must be finite"): load_config(path, environ={}) def test_negative_router_echo_grace_is_rejected(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "config.yaml" path.write_text( "mqtt:\n router_echo_grace_seconds: -1\n", encoding="utf-8" ) with self.assertRaisesRegex(ConfigError, "router_echo_grace_seconds"): load_config(path, environ={}) def test_runtime_paths_must_be_absolute(self): cases = ( ("storage:\n media_base: relative/media\n", "storage.media_base"), ("storage:\n data_base: relative/data\n", "storage.data_base"), ("downloads:\n cookie_path: relative/cookies.txt\n", "downloads.cookie_path"), ) for contents, field in cases: with self.subTest(field=field), tempfile.TemporaryDirectory() as directory: path = Path(directory) / "config.yaml" path.write_text(contents, encoding="utf-8") with self.assertRaisesRegex(ConfigError, f"{field} must be an absolute path"): load_config(path, environ={}) class ImportLifecycleTests(unittest.TestCase): def test_controller_core_import_does_not_connect_or_create_runtime_paths(self): with tempfile.TemporaryDirectory() as directory: media = Path(directory) / "media" data = Path(directory) / "data" env = { **os.environ, "YTPLAYER_MEDIA_BASE": str(media), "YTPLAYER_DATA_BASE": str(data), } sys.modules.pop("controller_core", None) with mock.patch.dict(os.environ, env, clear=True): module = importlib.import_module("controller_core") self.assertFalse(media.exists()) self.assertFalse(data.exists()) self.assertIsNone(module.mqtt_client) self.assertIs(module.mqtt_started, False) sys.modules.pop("controller_core", None) if __name__ == "__main__": unittest.main()