From 9091f2728371a247714eafa0d54aae3d74e147da Mon Sep 17 00:00:00 2001 From: Media Date: Thu, 6 Aug 2026 17:24:20 -0500 Subject: [PATCH] test: characterize contracts and make startup explicit --- app.py | 8 +- config.yaml | 59 ++++++- controller/__init__.py | 5 + controller/config.py | 274 +++++++++++++++++++++++++++++++++ controller/contracts.py | 48 ++++++ controller/ha.py | 98 ++++++++++++ controller_core.py | 120 +++++++++------ tests/__init__.py | 1 + tests/test_config_contracts.py | 138 +++++++++++++++++ 9 files changed, 695 insertions(+), 56 deletions(-) create mode 100644 controller/__init__.py create mode 100644 controller/config.py create mode 100644 controller/contracts.py create mode 100644 controller/ha.py create mode 100644 tests/__init__.py create mode 100644 tests/test_config_contracts.py diff --git a/app.py b/app.py index 04cb751..6ff9232 100644 --- a/app.py +++ b/app.py @@ -1,7 +1,7 @@ from flask import Flask, render_template_string, request, redirect, url_for, flash import os -from controller_core import enqueue_download +from controller_core import enqueue_download, start_controller, stop_controller app = Flask(__name__) app.secret_key = os.environ.get("FLASK_SECRET_KEY", "dev-secret") @@ -250,4 +250,8 @@ def submit(): return redirect(url_for("index")) if __name__ == "__main__": - app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 7531))) \ No newline at end of file + start_controller() + try: + app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 7531))) + finally: + stop_controller() diff --git a/config.yaml b/config.yaml index 1356fdb..0a9515f 100644 --- a/config.yaml +++ b/config.yaml @@ -3,13 +3,58 @@ mqtt: port: 1883 username: null password: null - client_id_prefix: "ytplayer" + client_id: yt-controller + keepalive_seconds: 30 + reconnect_min_seconds: 1 + reconnect_max_seconds: 30 + command_queue_size: 512 topics: - cmd: "youtube/player/cmd" - status: "youtube/player/status" + player_command: youtube/player/cmd + player_status: youtube/player/status + player_state: youtube/player/state + player_telemetry: youtube/player/telemetry + controller_command: youtube/controller/cmd + ingest_progress: youtube/ingest/progress + playback_base: youtube/media/playback + router_mode_set: audio/player/mode/set + router_mode_state: audio/player/mode/state + # Optional additive result topic. Existing consumers do not depend on it. + controller_result: null -player: - display: ":0" - xdg_runtime_dir: "/run/user/1000" - mpv_binary: "/usr/bin/mpv" +storage: + media_base: /mnt/Media/YTPlayer + data_base: /controller-data + database_name: controller.sqlite3 + legacy_yaml_name: playlists.yaml + slot_count: 10 + retention_seconds: 172800 + expiry_interval_seconds: 60 + # Marks uncertain playback for operator review; it never removes protection. + uncertain_review_after_seconds: 900 + +downloads: + yt_dlp_binary: yt-dlp + cookie_path: /run/secrets/yt-dlp-cookies.txt + cookie_required: false + max_workers: 2 + process_timeout_seconds: 7200 + max_height: 1080 + +router: + required_mode: video + confirmation_timeout_seconds: 3 + # attempt: request video and wait for the finite confirmation timeout + # continue: request video but do not wait when no state has been observed + # reject: reject playback until a Router state has been observed + when_state_unknown: attempt + continue_on_confirmation_failure: true + fallback_restore_mode: null + +homeassistant: + enabled: true + discovery_prefix: homeassistant + +web: + host: 0.0.0.0 + port: 7531 diff --git a/controller/__init__.py b/controller/__init__.py new file mode 100644 index 0000000..2bbe3c1 --- /dev/null +++ b/controller/__init__.py @@ -0,0 +1,5 @@ +"""YTPlayer Controller v2 package.""" + +from .config import ControllerConfig, ConfigError, load_config + +__all__ = ["ControllerConfig", "ConfigError", "load_config"] diff --git a/controller/config.py b/controller/config.py new file mode 100644 index 0000000..714be34 --- /dev/null +++ b/controller/config.py @@ -0,0 +1,274 @@ +"""Validated configuration for Controller v2.""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Mapping + +import yaml + +from .contracts import SLOT_IDS, Topics + + +BASE_DIR = Path(__file__).resolve().parent.parent +DEFAULT_CONFIG_PATH = BASE_DIR / "config.yaml" + + +class ConfigError(ValueError): + """Raised when Controller configuration is invalid.""" + + +@dataclass(frozen=True) +class MqttConfig: + host: str = "10.40.0.2" + port: int = 1883 + username: str | None = None + password: str | None = None + client_id: str = "yt-controller" + keepalive_seconds: int = 30 + reconnect_min_seconds: int = 1 + reconnect_max_seconds: int = 30 + command_queue_size: int = 512 + + +@dataclass(frozen=True) +class StorageConfig: + media_base: Path = Path("/mnt/Media/YTPlayer") + data_base: Path = Path("/controller-data") + database_name: str = "controller.sqlite3" + legacy_yaml_name: str = "playlists.yaml" + retention_seconds: int = 48 * 3600 + expiry_interval_seconds: float = 60.0 + uncertain_review_after_seconds: float = 15 * 60.0 + + @property + def database_path(self) -> Path: + return self.data_base / self.database_name + + @property + def legacy_yaml_path(self) -> Path: + return self.data_base / self.legacy_yaml_name + + +@dataclass(frozen=True) +class DownloadConfig: + yt_dlp_binary: str = "yt-dlp" + cookie_path: Path | None = Path("/run/secrets/yt-dlp-cookies.txt") + cookie_required: bool = False + max_workers: int = 2 + process_timeout_seconds: float = 2 * 3600.0 + max_height: int = 1080 + + +@dataclass(frozen=True) +class RouterConfig: + required_mode: str = "video" + confirmation_timeout_seconds: float = 3.0 + when_state_unknown: str = "attempt" + continue_on_confirmation_failure: bool = True + fallback_restore_mode: str | None = None + + +@dataclass(frozen=True) +class HomeAssistantConfig: + enabled: bool = True + discovery_prefix: str = "homeassistant" + + +@dataclass(frozen=True) +class WebConfig: + host: str = "0.0.0.0" + port: int = 7531 + + +@dataclass(frozen=True) +class ControllerConfig: + mqtt: MqttConfig = field(default_factory=MqttConfig) + topics: Topics = field(default_factory=Topics) + storage: StorageConfig = field(default_factory=StorageConfig) + downloads: DownloadConfig = field(default_factory=DownloadConfig) + router: RouterConfig = field(default_factory=RouterConfig) + homeassistant: HomeAssistantConfig = field(default_factory=HomeAssistantConfig) + web: WebConfig = field(default_factory=WebConfig) + + @property + def slot_ids(self) -> tuple[int, ...]: + return SLOT_IDS + + +def _mapping(value: Any, name: str) -> Mapping[str, Any]: + if value is None: + return {} + if not isinstance(value, Mapping): + raise ConfigError(f"{name} must be a mapping") + return value + + +def _text(value: Any, name: str, *, allow_none: bool = False) -> str | None: + if value is None and allow_none: + return None + if not isinstance(value, str) or not value.strip(): + raise ConfigError(f"{name} must be a non-empty string") + return value.strip() + + +def _integer(value: Any, name: str, *, minimum: int = 1, maximum: int | None = None) -> int: + if isinstance(value, bool): + raise ConfigError(f"{name} must be an integer") + try: + result = int(value) + except (TypeError, ValueError, OverflowError) as exc: + raise ConfigError(f"{name} must be an integer") from exc + if result < minimum or (maximum is not None and result > maximum): + upper = f" and <= {maximum}" if maximum is not None else "" + raise ConfigError(f"{name} must be >= {minimum}{upper}") + return result + + +def _number(value: Any, name: str, *, minimum: float = 0.0) -> float: + if isinstance(value, bool): + raise ConfigError(f"{name} must be a number") + try: + result = float(value) + except (TypeError, ValueError, OverflowError) as exc: + raise ConfigError(f"{name} must be a number") from exc + if result < minimum: + raise ConfigError(f"{name} must be >= {minimum}") + return result + + +def _boolean(value: Any, name: str) -> bool: + if isinstance(value, bool): + return value + if isinstance(value, str): + normalized = value.strip().lower() + if normalized in {"1", "true", "yes", "on"}: + return True + if normalized in {"0", "false", "no", "off"}: + return False + raise ConfigError(f"{name} must be a boolean") + + +def _optional_path(value: Any, name: str) -> Path | None: + if value in (None, ""): + return None + text = _text(value, name) + assert text is not None + return Path(text) + + +def _env_value(env: Mapping[str, str], key: str, fallback: Any) -> Any: + return env[key] if key in env else fallback + + +def load_config( + path: str | Path | None = None, + *, + environ: Mapping[str, str] | None = None, +) -> ControllerConfig: + """Load YAML relative to the repository and apply documented env overrides.""" + + config_path = Path(path) if path is not None else DEFAULT_CONFIG_PATH + raw: Mapping[str, Any] = {} + if config_path.exists(): + loaded = yaml.safe_load(config_path.read_text(encoding="utf-8")) + raw = _mapping(loaded, "configuration") + + env = os.environ if environ is None else environ + mqtt_raw = _mapping(raw.get("mqtt"), "mqtt") + topics_raw = _mapping(raw.get("topics"), "topics") + storage_raw = _mapping(raw.get("storage"), "storage") + downloads_raw = _mapping(raw.get("downloads"), "downloads") + router_raw = _mapping(raw.get("router"), "router") + ha_raw = _mapping(raw.get("homeassistant"), "homeassistant") + web_raw = _mapping(raw.get("web"), "web") + + mqtt = MqttConfig( + host=_text(_env_value(env, "MQTT_HOST", mqtt_raw.get("host", "10.40.0.2")), "mqtt.host"), + port=_integer(_env_value(env, "MQTT_PORT", mqtt_raw.get("port", 1883)), "mqtt.port", maximum=65535), + username=_text(_env_value(env, "MQTT_USERNAME", mqtt_raw.get("username")), "mqtt.username", allow_none=True), + password=_text(_env_value(env, "MQTT_PASSWORD", mqtt_raw.get("password")), "mqtt.password", allow_none=True), + client_id=_text(_env_value(env, "MQTT_CLIENT_ID", mqtt_raw.get("client_id", "yt-controller")), "mqtt.client_id"), + keepalive_seconds=_integer(mqtt_raw.get("keepalive_seconds", 30), "mqtt.keepalive_seconds"), + reconnect_min_seconds=_integer(mqtt_raw.get("reconnect_min_seconds", 1), "mqtt.reconnect_min_seconds"), + reconnect_max_seconds=_integer(mqtt_raw.get("reconnect_max_seconds", 30), "mqtt.reconnect_max_seconds"), + command_queue_size=_integer(mqtt_raw.get("command_queue_size", 512), "mqtt.command_queue_size"), + ) + if mqtt.reconnect_max_seconds < mqtt.reconnect_min_seconds: + raise ConfigError("mqtt.reconnect_max_seconds must be >= reconnect_min_seconds") + if (mqtt.username is None) != (mqtt.password is None): + raise ConfigError("mqtt.username and mqtt.password must either both be set or both be null") + + topics = Topics( + player_command=_text(topics_raw.get("player_command", "youtube/player/cmd"), "topics.player_command"), + player_status=_text(topics_raw.get("player_status", "youtube/player/status"), "topics.player_status"), + player_state=_text(topics_raw.get("player_state", "youtube/player/state"), "topics.player_state"), + player_telemetry=_text(topics_raw.get("player_telemetry", "youtube/player/telemetry"), "topics.player_telemetry"), + controller_command=_text(topics_raw.get("controller_command", "youtube/controller/cmd"), "topics.controller_command"), + ingest_progress=_text(topics_raw.get("ingest_progress", "youtube/ingest/progress"), "topics.ingest_progress"), + playback_base=_text(topics_raw.get("playback_base", "youtube/media/playback"), "topics.playback_base"), + router_mode_set=_text(topics_raw.get("router_mode_set", "audio/player/mode/set"), "topics.router_mode_set"), + router_mode_state=_text(topics_raw.get("router_mode_state", "audio/player/mode/state"), "topics.router_mode_state"), + controller_result=_text(topics_raw.get("controller_result"), "topics.controller_result", allow_none=True), + ) + + configured_slots = _integer(storage_raw.get("slot_count", 10), "storage.slot_count") + if configured_slots != len(SLOT_IDS): + raise ConfigError("storage.slot_count is fixed at 10 for stable slot identities") + storage = StorageConfig( + media_base=Path(_text(_env_value(env, "YTPLAYER_MEDIA_BASE", storage_raw.get("media_base", "/mnt/Media/YTPlayer")), "storage.media_base")), + data_base=Path(_text(_env_value(env, "YTPLAYER_DATA_BASE", storage_raw.get("data_base", "/controller-data")), "storage.data_base")), + database_name=_text(storage_raw.get("database_name", "controller.sqlite3"), "storage.database_name"), + legacy_yaml_name=_text(storage_raw.get("legacy_yaml_name", "playlists.yaml"), "storage.legacy_yaml_name"), + retention_seconds=_integer(_env_value(env, "YTPLAYER_RETENTION_SECONDS", storage_raw.get("retention_seconds", 48 * 3600)), "storage.retention_seconds"), + expiry_interval_seconds=_number(storage_raw.get("expiry_interval_seconds", 60), "storage.expiry_interval_seconds", minimum=0.1), + uncertain_review_after_seconds=_number(storage_raw.get("uncertain_review_after_seconds", 15 * 60), "storage.uncertain_review_after_seconds", minimum=0.0), + ) + if Path(storage.database_name).name != storage.database_name: + raise ConfigError("storage.database_name must be a filename") + if Path(storage.legacy_yaml_name).name != storage.legacy_yaml_name: + raise ConfigError("storage.legacy_yaml_name must be a filename") + + cookie_raw = _env_value(env, "YTPLAYER_COOKIE_PATH", downloads_raw.get("cookie_path", "/run/secrets/yt-dlp-cookies.txt")) + downloads = DownloadConfig( + yt_dlp_binary=_text(downloads_raw.get("yt_dlp_binary", "yt-dlp"), "downloads.yt_dlp_binary"), + cookie_path=_optional_path(cookie_raw, "downloads.cookie_path"), + cookie_required=_boolean(downloads_raw.get("cookie_required", False), "downloads.cookie_required"), + max_workers=_integer(downloads_raw.get("max_workers", 2), "downloads.max_workers", maximum=10), + process_timeout_seconds=_number(downloads_raw.get("process_timeout_seconds", 2 * 3600), "downloads.process_timeout_seconds", minimum=1.0), + max_height=_integer(downloads_raw.get("max_height", 1080), "downloads.max_height"), + ) + if downloads.cookie_required and downloads.cookie_path is None: + raise ConfigError("downloads.cookie_path is required when cookie_required is true") + + unknown_behavior = _text(router_raw.get("when_state_unknown", "attempt"), "router.when_state_unknown") + if unknown_behavior not in {"attempt", "continue", "reject"}: + raise ConfigError("router.when_state_unknown must be attempt, continue, or reject") + fallback = _text(router_raw.get("fallback_restore_mode"), "router.fallback_restore_mode", allow_none=True) + router = RouterConfig( + required_mode=_text(router_raw.get("required_mode", "video"), "router.required_mode"), + confirmation_timeout_seconds=_number(router_raw.get("confirmation_timeout_seconds", 3.0), "router.confirmation_timeout_seconds", minimum=0.1), + when_state_unknown=unknown_behavior, + continue_on_confirmation_failure=_boolean(router_raw.get("continue_on_confirmation_failure", True), "router.continue_on_confirmation_failure"), + fallback_restore_mode=fallback, + ) + + homeassistant = HomeAssistantConfig( + enabled=_boolean(ha_raw.get("enabled", True), "homeassistant.enabled"), + discovery_prefix=_text(ha_raw.get("discovery_prefix", "homeassistant"), "homeassistant.discovery_prefix"), + ) + web = WebConfig( + host=_text(web_raw.get("host", "0.0.0.0"), "web.host"), + port=_integer(_env_value(env, "PORT", web_raw.get("port", 7531)), "web.port", maximum=65535), + ) + return ControllerConfig( + mqtt=mqtt, + topics=topics, + storage=storage, + downloads=downloads, + router=router, + homeassistant=homeassistant, + web=web, + ) diff --git a/controller/contracts.py b/controller/contracts.py new file mode 100644 index 0000000..131f56f --- /dev/null +++ b/controller/contracts.py @@ -0,0 +1,48 @@ +"""Stable MQTT contracts shared by the Controller adapters.""" + +from __future__ import annotations + +from dataclasses import dataclass + + +SLOT_IDS = tuple(range(1, 11)) + + +@dataclass(frozen=True) +class Topics: + """MQTT topic names. + + Defaults preserve the established Player, Home Assistant, and Companion + integrations. Player ``state`` is fixed by the Player Node contract. + """ + + player_command: str = "youtube/player/cmd" + player_status: str = "youtube/player/status" + player_state: str = "youtube/player/state" + player_telemetry: str = "youtube/player/telemetry" + controller_command: str = "youtube/controller/cmd" + ingest_progress: str = "youtube/ingest/progress" + playback_base: str = "youtube/media/playback" + router_mode_set: str = "audio/player/mode/set" + router_mode_state: str = "audio/player/mode/state" + controller_result: str | None = None + + def slot_base(self, slot: int) -> str: + if slot not in SLOT_IDS: + raise ValueError(f"slot must be in 1..10, got {slot!r}") + return f"{self.playback_base}/{slot}" + + +PLAYER_COMMAND_QOS = 1 +PLAYER_COMMAND_RETAIN = False +PLAYER_STATUS_QOS = 1 +PLAYER_STATE_QOS = 0 +PLAYER_TELEMETRY_QOS = 0 + +# Existing Controller-owned HA/Companion state was retained at QoS 0. +LEGACY_STATE_QOS = 0 +LEGACY_STATE_RETAIN = True + +# Router commands are ephemeral commands just like Player commands. +ROUTER_COMMAND_QOS = 1 +ROUTER_COMMAND_RETAIN = False diff --git a/controller/ha.py b/controller/ha.py new file mode 100644 index 0000000..16d9a7b --- /dev/null +++ b/controller/ha.py @@ -0,0 +1,98 @@ +"""Home Assistant discovery and legacy slot-publication helpers.""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from typing import Any, Callable, Iterable + +from .config import HomeAssistantConfig +from .contracts import LEGACY_STATE_QOS, LEGACY_STATE_RETAIN, SLOT_IDS, Topics + + +Publish = Callable[..., Any] + + +HA_DEVICE = { + "identifiers": ["ytplayer-controller"], + "name": "YouTube Controller", + "manufacturer": "River of Glory", + "model": "YT Controller", +} + + +@dataclass(frozen=True) +class SlotAnnouncement: + slot: int + title: str + state: str + count: int + items: tuple[str, ...] = () + + +def discovery_messages( + topics: Topics, + config: HomeAssistantConfig, +) -> Iterable[tuple[str, str, int, bool]]: + """Yield the existing thirty discovery documents unchanged.""" + + for slot in SLOT_IDS: + base = topics.slot_base(slot) + object_id = f"yt_slot_{slot}" + entities = ( + ( + "title", + { + "name": f"YT Slot {slot} Title", + "state_topic": f"{base}/title", + "unique_id": f"{object_id}_title", + "icon": "mdi:playlist-music", + "device": HA_DEVICE, + }, + ), + ( + "state", + { + "name": f"YT Slot {slot} State", + "state_topic": f"{base}/state", + "unique_id": f"{object_id}_state", + "icon": "mdi:circle-small", + "device": HA_DEVICE, + }, + ), + ( + "count", + { + "name": f"YT Slot {slot} Count", + "state_topic": f"{base}/count", + "unique_id": f"{object_id}_count", + "icon": "mdi:numeric", + "device": HA_DEVICE, + }, + ), + ) + for suffix, payload in entities: + yield ( + f"{config.discovery_prefix}/sensor/{object_id}_{suffix}/config", + json.dumps(payload), + LEGACY_STATE_QOS, + LEGACY_STATE_RETAIN, + ) + + +def slot_messages( + announcement: SlotAnnouncement, + topics: Topics, + *, + clear_items_through: int = 0, +) -> Iterable[tuple[str, str, int, bool]]: + """Yield retained legacy slot topics, including stale item cleanup.""" + + base = topics.slot_base(announcement.slot) + yield f"{base}/title", announcement.title, LEGACY_STATE_QOS, LEGACY_STATE_RETAIN + yield f"{base}/state", announcement.state, LEGACY_STATE_QOS, LEGACY_STATE_RETAIN + yield f"{base}/count", str(announcement.count), LEGACY_STATE_QOS, LEGACY_STATE_RETAIN + for index, title in enumerate(announcement.items, start=1): + yield f"{base}/items/{index}", title, LEGACY_STATE_QOS, LEGACY_STATE_RETAIN + for index in range(len(announcement.items) + 1, clear_items_through + 1): + yield f"{base}/items/{index}", "", LEGACY_STATE_QOS, LEGACY_STATE_RETAIN diff --git a/controller_core.py b/controller_core.py index 4d5f893..a20c944 100644 --- a/controller_core.py +++ b/controller_core.py @@ -10,27 +10,28 @@ from typing import Dict, List, Optional import yaml import paho.mqtt.client as mqtt +from controller.config import load_config + # --- Config / paths --- -MEDIA_BASE = Path(os.environ.get("YTPLAYER_MEDIA_BASE", "/mnt/Media/YTPlayer")) -DATA_BASE = Path(os.environ.get("YTPLAYER_DATA_BASE", "/controller-data")) +CONFIG = load_config() + +MEDIA_BASE = CONFIG.storage.media_base +DATA_BASE = CONFIG.storage.data_base STATE_PATH = DATA_BASE / "playlists.yaml" -MAX_SLOTS = int(os.environ.get("YTPLAYER_MAX_SLOTS", "10")) -RETENTION_SECONDS = int(os.environ.get("YTPLAYER_RETENTION_SECONDS", str(48 * 3600))) +MAX_SLOTS = len(CONFIG.slot_ids) +RETENTION_SECONDS = CONFIG.storage.retention_seconds -MQTT_HOST = os.environ.get("MQTT_HOST", "10.40.0.2") -MQTT_PORT = int(os.environ.get("MQTT_PORT", "1883")) -MQTT_CLIENT_ID = os.environ.get("MQTT_CLIENT_ID", "yt-controller") +MQTT_HOST = CONFIG.mqtt.host +MQTT_PORT = CONFIG.mqtt.port +MQTT_CLIENT_ID = CONFIG.mqtt.client_id -TOPIC_INGEST_PROGRESS = "youtube/ingest/progress" -TOPIC_PLAYBACK_BASE = "youtube/media/playback" # e.g. youtube/media/playback/1/title -TOPIC_PLAYER_STATUS = "youtube/player/status" # JSON from player_service -TOPIC_PLAYER_CMD = "youtube/player/cmd" # commands to player_service -TOPIC_CONTROLLER_CMD = "youtube/controller/cmd" # commands from HA / StreamDeck - -MEDIA_BASE.mkdir(parents=True, exist_ok=True) -DATA_BASE.mkdir(parents=True, exist_ok=True) +TOPIC_INGEST_PROGRESS = CONFIG.topics.ingest_progress +TOPIC_PLAYBACK_BASE = CONFIG.topics.playback_base +TOPIC_PLAYER_STATUS = CONFIG.topics.player_status +TOPIC_PLAYER_CMD = CONFIG.topics.player_command +TOPIC_CONTROLLER_CMD = CONFIG.topics.controller_command # --- MQTT client --- @@ -43,6 +44,9 @@ def _on_connect(client, userdata, flags, rc, properties=None): client.subscribe(TOPIC_PLAYER_STATUS, qos=1) client.subscribe(TOPIC_CONTROLLER_CMD, qos=1) print(f"[controller] subscribed to {TOPIC_PLAYER_STATUS} and {TOPIC_CONTROLLER_CMD}") + if CONFIG.homeassistant.enabled: + ha_discovery_publish() + reannounce_slots() else: print(f"[mqtt] connect failed rc={rc}") @@ -111,26 +115,9 @@ def _handle_control_cmd(payload: str): def _on_disconnect(client, userdata, rc, properties=None): print(f"[mqtt] disconnected rc={rc}") -mqtt_client = mqtt.Client(client_id=MQTT_CLIENT_ID, clean_session=True) - -# If your broker ever needs auth, you'd uncomment: -# mqtt_client.username_pw_set(os.environ.get("MQTT_USERNAME"), os.environ.get("MQTT_PASSWORD")) - -mqtt_client.enable_logger() # send Paho logs to Python logging -mqtt_client.on_connect = _on_connect -mqtt_client.on_disconnect = _on_disconnect -mqtt_client.on_message = _on_message - -print(f"[controller] attempting MQTT connect to {MQTT_HOST}:{MQTT_PORT} …") -mqtt_client.connect(MQTT_HOST, MQTT_PORT, keepalive=30) - -mqtt_client.loop_start() -print("[controller] MQTT loop started") - -mqtt_client.on_message = _on_message -mqtt_client.subscribe(TOPIC_PLAYER_STATUS, qos=1) -mqtt_client.subscribe(TOPIC_CONTROLLER_CMD, qos=1) -print(f"[controller] subscribed to {TOPIC_PLAYER_STATUS} and {TOPIC_CONTROLLER_CMD}") +mqtt_client = None +mqtt_started = False +mqtt_lifecycle_lock = threading.Lock() def mqtt_publish(topic: str, payload, **kwargs): @@ -141,7 +128,10 @@ def mqtt_publish(topic: str, payload, **kwargs): if not isinstance(payload, str): payload = json.dumps(payload) print(f"[mqtt] publish {topic} -> {payload}") - mqtt_client.publish(topic, payload, **kwargs) + if mqtt_client is None: + print(f"[mqtt] publish skipped before Controller startup: {topic}") + return None + return mqtt_client.publish(topic, payload, **kwargs) # --- State model --- @@ -282,8 +272,6 @@ class ControllerState: state = ControllerState() -state.load() -state.prune_expired() # --- Helpers --- @@ -698,7 +686,7 @@ def handle_play_request(cmd: Dict) -> None: # --- HA Discovery --- -HA_DISCOVERY_PREFIX = os.environ.get("HA_DISCOVERY_PREFIX", "homeassistant") +HA_DISCOVERY_PREFIX = CONFIG.homeassistant.discovery_prefix HA_DEVICE = { "identifiers": ["ytplayer-controller"], "name": "YouTube Controller", @@ -826,13 +814,51 @@ def enqueue_download(url: str, submitter: Optional[str], note: Optional[str]) -> } -# --- Post-define initialization that needs functions above --- +# --- Explicit lifecycle (imports are intentionally side-effect free) --- -print("[controller] running HA discovery publish…") -ha_discovery_publish() -print("[controller] ha_discovery_publish() done") +def start_controller() -> None: + """Initialize storage and start non-blocking MQTT retry behavior.""" + global mqtt_client, mqtt_started + with mqtt_lifecycle_lock: + if mqtt_started: + return -# Re-announce any existing slots so HA / Deck see them after a restart -for s, slot in state.slots.items(): - publish_slot_topics(slot) -print("[controller] re-published slot states") + MEDIA_BASE.mkdir(parents=True, exist_ok=True) + DATA_BASE.mkdir(parents=True, exist_ok=True) + state.load() + + mqtt_client = mqtt.Client(client_id=MQTT_CLIENT_ID, clean_session=True) + if CONFIG.mqtt.username is not None: + mqtt_client.username_pw_set(CONFIG.mqtt.username, CONFIG.mqtt.password) + mqtt_client.enable_logger() + mqtt_client.on_connect = _on_connect + mqtt_client.on_disconnect = _on_disconnect + mqtt_client.on_message = _on_message + mqtt_client.reconnect_delay_set( + min_delay=CONFIG.mqtt.reconnect_min_seconds, + max_delay=CONFIG.mqtt.reconnect_max_seconds, + ) + + print(f"[controller] attempting asynchronous MQTT connect to {MQTT_HOST}:{MQTT_PORT} …") + mqtt_client.connect_async( + MQTT_HOST, + MQTT_PORT, + keepalive=CONFIG.mqtt.keepalive_seconds, + ) + mqtt_client.loop_start() + mqtt_started = True + print("[controller] MQTT loop started") + + +def stop_controller() -> None: + """Stop the MQTT client if explicit startup occurred.""" + global mqtt_client, mqtt_started + with mqtt_lifecycle_lock: + if not mqtt_started or mqtt_client is None: + return + try: + mqtt_client.disconnect() + finally: + mqtt_client.loop_stop() + mqtt_client = None + mqtt_started = False diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..65dbb95 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Controller regression tests.""" diff --git a/tests/test_config_contracts.py b/tests/test_config_contracts.py new file mode 100644 index 0000000..2fa7e43 --- /dev/null +++ b/tests/test_config_contracts.py @@ -0,0 +1,138 @@ +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(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) + + 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.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_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={}) + + +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()