import ast import re import unittest from pathlib import Path from typing import Any import yaml ROOT = Path(__file__).resolve().parents[1] def _read(name: str) -> str: return (ROOT / name).read_text(encoding="utf-8") def _docker_logical_lines(text: str) -> list[str]: logical: list[str] = [] pending = "" for raw_line in text.splitlines(): stripped = raw_line.strip() if not stripped or (not pending and stripped.startswith("#")): continue pending = f"{pending} {stripped}".strip() if pending.endswith("\\"): pending = pending[:-1].rstrip() continue logical.append(pending) pending = "" if pending: logical.append(pending) return logical def _environment_map(value: Any) -> dict[str, str | None]: if isinstance(value, dict): return {str(key): None if item is None else str(item) for key, item in value.items()} result: dict[str, str | None] = {} for entry in value or []: key, separator, item = str(entry).partition("=") result[key] = item if separator else None return result def _volume_mounts(value: Any) -> list[tuple[str | None, str, bool]]: mounts: list[tuple[str | None, str, bool]] = [] for entry in value or []: if isinstance(entry, dict): target = str(entry.get("target") or "") if target: mounts.append( ( None if entry.get("source") is None else str(entry["source"]), target, bool(entry.get("read_only", False)), ) ) continue # Greedy source matching tolerates Compose substitutions containing # `:-` or `:?` while keeping the absolute container target distinct. match = re.match(r"^(.*):(/[^:]+)(?::([^:]+))?$", str(entry)) if match: mode = match.group(3) or "" mounts.append((match.group(1), match.group(2), "ro" in mode.split(","))) return mounts def _secret_targets(value: Any) -> set[str]: targets: set[str] = set() for entry in value or []: if isinstance(entry, dict): source = str(entry.get("source") or "") target = str(entry.get("target") or source) else: source = str(entry) target = source if not target: continue targets.add(target if target.startswith("/") else f"/run/secrets/{target}") return targets class RequirementPinTests(unittest.TestCase): def test_runtime_requirements_are_exactly_pinned(self): path = ROOT / "requirements.txt" self.assertTrue(path.is_file(), "requirements.txt is missing") requirements: dict[str, str] = {} pattern = re.compile( r"^([A-Za-z0-9_.-]+)(?:\[[A-Za-z0-9_,.-]+\])?==([^;\s]+)(?:\s*;.*)?$" ) for line_number, raw_line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1): line = raw_line.split("#", 1)[0].strip() if not line: continue match = pattern.fullmatch(line) self.assertIsNotNone( match, f"requirements.txt:{line_number} is not an exact == pin: {line!r}", ) assert match is not None normalized = re.sub(r"[-_.]+", "-", match.group(1)).lower() requirements[normalized] = match.group(2) for project in ("flask", "gunicorn", "paho-mqtt", "pyyaml", "yt-dlp"): self.assertIn(project, requirements, f"missing runtime dependency {project}") self.assertNotIn("*", requirements[project]) class DockerfileTests(unittest.TestCase): def setUp(self): self.path = ROOT / "Dockerfile" self.assertTrue(self.path.is_file(), "Dockerfile is missing") self.text = self.path.read_text(encoding="utf-8") self.lines = _docker_logical_lines(self.text) def test_image_copies_the_v2_runtime(self): copies = "\n".join( line for line in self.lines if line.upper().startswith("COPY ") ) for required in ( "controller", "config.yaml", "app.py", "wsgi.py", "gunicorn.conf.py", ): self.assertIn(required, copies, f"Dockerfile does not copy {required}") def test_image_runs_non_root_on_the_controller_port(self): users = [line.split(None, 1)[1] for line in self.lines if line.upper().startswith("USER ")] self.assertTrue(users, "Dockerfile must set a non-root USER") self.assertNotIn(users[-1].strip().lower(), {"root", "0", "0:0"}) self.assertRegex(self.text, r"(?im)^\s*EXPOSE\s+7531(?:/tcp)?\s*$") def test_image_has_an_http_healthcheck_and_runs_gunicorn(self): self.assertRegex(self.text, r"(?im)^\s*HEALTHCHECK\b") self.assertIn("/healthz", self.text) launch = "\n".join( line for line in self.lines if line.upper().startswith(("CMD ", "ENTRYPOINT ")) ).lower() self.assertIn("gunicorn", launch) def test_image_has_no_unbounded_yt_dlp_upgrade_or_hidden_global_config(self): self.assertNotRegex( self.text, r"(?i)(?:--upgrade|-U)\s+[\"']?(?:yt-dlp|yt_dlp)", ) self.assertNotIn("/etc/yt-dlp.conf", self.text) def test_image_verifies_a_supported_node_runtime_for_yt_dlp(self): self.assertRegex( self.text, r"(?im)^FROM\s+node:22\.\d+\.\d+-bookworm-slim\s+AS\s+", ) self.assertIn("NodeJsRuntime", self.text) self.assertIn("info.supported", self.text) class GunicornConfigTests(unittest.TestCase): def test_gunicorn_uses_exactly_one_worker_without_preload(self): path = ROOT / "gunicorn.conf.py" self.assertTrue(path.is_file(), "gunicorn.conf.py is missing") tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) assignments: dict[str, Any] = {} for node in tree.body: if not isinstance(node, (ast.Assign, ast.AnnAssign)): continue targets = node.targets if isinstance(node, ast.Assign) else [node.target] try: value = ast.literal_eval(node.value) except (TypeError, ValueError): continue for target in targets: if isinstance(target, ast.Name): assignments[target.id] = value self.assertEqual(assignments.get("workers"), 1) self.assertIs(assignments.get("preload_app"), False) self.assertIs(assignments.get("control_socket_disable"), True) self.assertEqual(assignments.get("worker_tmp_dir"), "/tmp") class ComposeRuntimeTests(unittest.TestCase): def setUp(self): self.path = ROOT / "docker-compose.yml" self.assertTrue(self.path.is_file(), "docker-compose.yml is missing") self.text = self.path.read_text(encoding="utf-8") document = yaml.safe_load(self.text) self.assertIsInstance(document, dict) self.document = document self.service = document.get("services", {}).get("controller") self.assertIsInstance(self.service, dict, "Compose controller service is missing") def test_compose_requires_a_real_flask_secret_and_publishes_port_7531(self): environment = _environment_map(self.service.get("environment")) secret = environment.get("FLASK_SECRET_KEY") self.assertIsNotNone(secret, "FLASK_SECRET_KEY environment setting is missing") assert secret is not None self.assertTrue(secret.startswith("${FLASK_SECRET_KEY")) self.assertIn("?", secret, "FLASK_SECRET_KEY must use required Compose substitution") self.assertNotIn("change-me", secret.lower()) self.assertNotIn("dev-secret", secret.lower()) published_targets: set[str] = set() for entry in self.service.get("ports") or []: if isinstance(entry, dict): published_targets.add(str(entry.get("target"))) else: match = re.search(r":(\d+)(?:/(?:tcp|udp))?$", str(entry)) if match: published_targets.add(match.group(1)) self.assertIn("7531", published_targets) def test_compose_passes_all_supported_mqtt_environment_overrides(self): environment = _environment_map(self.service.get("environment")) for name in ( "MQTT_HOST", "MQTT_PORT", "MQTT_USERNAME", "MQTT_PASSWORD", "MQTT_CLIENT_ID", ): with self.subTest(name=name): self.assertIn(name, environment) def test_compose_uses_dedicated_data_media_and_cookie_mounts(self): mounts = _volume_mounts(self.service.get("volumes")) by_target = {target: (source, read_only) for source, target, read_only in mounts} self.assertIn("/controller-data", by_target) self.assertIn("/mnt/Media/YTPlayer", by_target) self.assertNotEqual(by_target["/controller-data"][0], "/home/media/Scripts/Video-Player") cookie_path = "/run/secrets/yt-dlp-cookies.txt" mounted_cookie = cookie_path in by_target and by_target[cookie_path][1] mounted_secret_directory = ( "/run/secrets" in by_target and by_target["/run/secrets"][1] ) secret_cookie = cookie_path in _secret_targets(self.service.get("secrets")) self.assertTrue( mounted_cookie or mounted_secret_directory or secret_cookie, f"cookies must be read-only at exactly {cookie_path}", ) self.assertNotIn("/cookies_ro.txt", self.text) self.assertNotRegex( self.text, r"/home/media/Scripts/Video-Player(?=[:\s\"'])", ) def test_compose_enables_container_hardening_and_bounded_operations(self): self.assertIs(self.service.get("read_only"), True) self.assertIs(self.service.get("init"), True) cap_drop = self.service.get("cap_drop") or [] if isinstance(cap_drop, str): cap_drop = [cap_drop] self.assertIn("ALL", {str(value).upper() for value in cap_drop}) security_options = self.service.get("security_opt") or [] if isinstance(security_options, str): security_options = [security_options] self.assertTrue( any("no-new-privileges" in str(value).lower() for value in security_options) ) tmpfs = self.service.get("tmpfs") or [] if isinstance(tmpfs, str): tmpfs = [tmpfs] self.assertTrue(any(str(value).split(":", 1)[0] == "/tmp" for value in tmpfs)) self.assertTrue(any("noexec" in str(value) for value in tmpfs)) healthcheck = self.service.get("healthcheck") self.assertIsInstance(healthcheck, dict) self.assertIn("/healthz", str(healthcheck.get("test"))) logging_config = self.service.get("logging") self.assertIsInstance(logging_config, dict) options = logging_config.get("options") or {} self.assertTrue(options.get("max-size"), "logging max-size is missing") self.assertTrue(options.get("max-file"), "logging max-file is missing") grace = str(self.service.get("stop_grace_period") or "").rstrip("s") self.assertGreaterEqual(float(grace), 30.0) def test_compose_preserves_the_external_tabernet_network(self): service_networks = self.service.get("networks") or [] if isinstance(service_networks, dict): service_networks = service_networks.keys() self.assertIn("tabernet", service_networks) tabernet = self.document.get("networks", {}).get("tabernet") self.assertIsInstance(tabernet, dict) self.assertIs(tabernet.get("external"), True) class DockerIgnoreTests(unittest.TestCase): def test_dockerignore_excludes_secrets_and_python_caches(self): path = ROOT / ".dockerignore" self.assertTrue(path.is_file(), ".dockerignore is missing") patterns = { line.strip().lower() for line in path.read_text(encoding="utf-8").splitlines() if line.strip() and not line.lstrip().startswith("#") } self.assertTrue(any("__pycache__" in pattern for pattern in patterns)) self.assertTrue( any("pyc" in pattern or "py[cod]" in pattern for pattern in patterns) ) self.assertTrue(any(pattern in {".env", ".env*", "*.env"} for pattern in patterns)) self.assertTrue(any("secret" in pattern for pattern in patterns)) self.assertTrue(any("cookie" in pattern for pattern in patterns)) if __name__ == "__main__": unittest.main()