247 lines
11 KiB
Python
247 lines
11 KiB
Python
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)
|
|
self.assertEqual(config.downloads.export_directory, Path("exports"))
|
|
self.assertEqual(config.downloads.export_max_height, 1080)
|
|
self.assertEqual(config.downloads.export_max_workers, 1)
|
|
self.assertEqual(config.downloads.max_queued_exports, 20)
|
|
|
|
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={})
|
|
|
|
def test_export_directory_must_stay_relative_to_media_root(self):
|
|
invalid = ("/tmp/exports", "../exports", "nested/../exports", r"nested\\exports")
|
|
for value in invalid:
|
|
with self.subTest(value=value), tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "config.yaml"
|
|
path.write_text(
|
|
f"downloads:\n export_directory: {value!r}\n",
|
|
encoding="utf-8",
|
|
)
|
|
with self.assertRaisesRegex(ConfigError, "safe relative directory"):
|
|
load_config(path, environ={})
|
|
|
|
def test_export_directory_cannot_overlap_player_slot_storage(self):
|
|
invalid = ("slot1", "SLOT10/exports", ".controller-trash/exports")
|
|
for value in invalid:
|
|
with self.subTest(value=value), tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "config.yaml"
|
|
path.write_text(
|
|
f"downloads:\n export_directory: {value!r}\n",
|
|
encoding="utf-8",
|
|
)
|
|
with self.assertRaisesRegex(ConfigError, "slot storage"):
|
|
load_config(path, environ={})
|
|
|
|
def test_export_limits_are_bounded_positive_integers(self):
|
|
cases = (
|
|
("export_max_height", 0),
|
|
("export_max_workers", 0),
|
|
("export_max_workers", 11),
|
|
("max_queued_exports", 0),
|
|
("max_queued_exports", 101),
|
|
)
|
|
for field, value in cases:
|
|
with self.subTest(field=field), tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "config.yaml"
|
|
path.write_text(
|
|
f"downloads:\n {field}: {value}\n",
|
|
encoding="utf-8",
|
|
)
|
|
with self.assertRaisesRegex(ConfigError, field):
|
|
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()
|