#!/usr/bin/env python3 """Build the Jellyfin fixture and run Plezy's Android Maestro suites.""" from __future__ import annotations import argparse from dataclasses import dataclass import os from pathlib import Path import shlex import signal import shutil import subprocess import sys import time from typing import Mapping, Optional, Sequence, TextIO import urllib.error import urllib.request ROOT_DIR = Path(__file__).resolve().parent.parent APP_ID = "com.edde746.plezy" FAULTS = ("music-failure", "offline", "recovery") class RunnerError(RuntimeError): """A user-actionable runner failure.""" class RunnerSignal(Exception): def __init__(self, signum: int) -> None: super().__init__(f"Interrupted by signal {signum}") self.exit_status = 128 + signum def _raise_signal(signum: int, _frame: object) -> None: raise RunnerSignal(signum) @dataclass(frozen=True) class SuitePreset: flow_target: str maestro_config: Optional[str] jellyfin_log: str diagnostics_dir: str use_adb_reverse: bool = False uninstall_before_install: bool = False SUITES = { "basic": SuitePreset( flow_target=".maestro", maestro_config=None, jellyfin_log="build/maestro/jellyfin.log", diagnostics_dir="build/maestro/diagnostics", ), "catalog": SuitePreset( flow_target=".maestro/real_flows", maestro_config=None, jellyfin_log="build/maestro-real-jellyfin/jellyfin.log", diagnostics_dir="build/maestro-real-jellyfin/diagnostics", uninstall_before_install=True, ), "media": SuitePreset( flow_target=".maestro/media_flows", maestro_config=".maestro/media-config.yaml", jellyfin_log="build/maestro-media/jellyfin.log", diagnostics_dir="build/maestro/diagnostics", use_adb_reverse=True, uninstall_before_install=True, ), } @dataclass(frozen=True) class RunnerConfig: command: str jellyfin_host: str jellyfin_port: int proxy_port: int jellyfin_image: str skip_jellyfin: bool skip_jellyfin_build: bool skip_build: bool jellyfin_fault: Optional[str] use_adb_reverse: bool device_id: Optional[str] apk_path: Path flow_target: Path maestro_config: Optional[Path] uninstall_before_install: bool diagnostics_dir: Path jellyfin_log: Path proxy_log: Path proxy_journal: Path host_jellyfin_url: str jellyfin_url: Optional[str] jellyfin_build_attempts: int def _positive_int(value: str) -> int: try: parsed = int(value) except ValueError as error: raise argparse.ArgumentTypeError("must be a positive integer") from error if parsed < 1: raise argparse.ArgumentTypeError("must be a positive integer") return parsed def _add_bool_argument(parser: argparse.ArgumentParser, name: str, *, destination: str) -> None: group = parser.add_mutually_exclusive_group() group.add_argument(f"--{name}", dest=destination, action="store_true") group.add_argument(f"--no-{name}", dest=destination, action="store_false") parser.set_defaults(**{destination: None}) def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "command", choices=(*SUITES, "build-image"), nargs="?", default="basic", help="suite to run, or build-image to only build the Jellyfin fixture", ) parser.add_argument("--device", dest="device_id") parser.add_argument("--apk", dest="apk_path") parser.add_argument("--flow", dest="flow_target") parser.add_argument("--config", dest="maestro_config") parser.add_argument("--fault", choices=FAULTS, dest="jellyfin_fault") parser.add_argument("--jellyfin-host") parser.add_argument("--jellyfin-port", type=int) parser.add_argument("--proxy-port", type=int) parser.add_argument("--jellyfin-image") parser.add_argument("--diagnostics-dir") parser.add_argument("--jellyfin-log") parser.add_argument("--proxy-log") parser.add_argument("--proxy-journal") parser.add_argument("--host-jellyfin-url") parser.add_argument("--jellyfin-url") parser.add_argument("--jellyfin-build-attempts", type=_positive_int) _add_bool_argument(parser, "skip-jellyfin", destination="skip_jellyfin") _add_bool_argument(parser, "skip-jellyfin-build", destination="skip_jellyfin_build") _add_bool_argument(parser, "skip-build", destination="skip_build") _add_bool_argument(parser, "adb-reverse", destination="use_adb_reverse") _add_bool_argument(parser, "uninstall-before-install", destination="uninstall_before_install") return parser def _env_bool(environment: Mapping[str, str], name: str, default: bool) -> bool: value = environment.get(name) if value is None: return default normalized = value.strip().lower() if normalized in {"1", "true", "yes", "on"}: return True if normalized in {"0", "false", "no", "off"}: return False raise RunnerError(f"{name} must be one of 1, 0, true, false, yes, no, on, or off") def _option(cli_value: object, environment: Mapping[str, str], name: str, default: object) -> object: if cli_value is not None: return cli_value return environment.get(name, default) def _int_option( cli_value: Optional[int], environment: Mapping[str, str], name: str, default: int, ) -> int: value = _option(cli_value, environment, name, default) try: parsed = int(value) except (TypeError, ValueError) as error: raise RunnerError(f"{name} must be a positive integer") from error if parsed < 1: raise RunnerError(f"{name} must be a positive integer") return parsed def _bool_option( cli_value: Optional[bool], environment: Mapping[str, str], name: str, default: bool, ) -> bool: if cli_value is not None: return cli_value return _env_bool(environment, name, default) def _root_path(value: object) -> Path: path = Path(str(value)) return path if path.is_absolute() else ROOT_DIR / path def parse_config(argv: Optional[Sequence[str]] = None, environment: Optional[Mapping[str, str]] = None) -> RunnerConfig: args = _parser().parse_args(argv) env = os.environ if environment is None else environment preset = SUITES.get(args.command, SUITES["basic"]) jellyfin_host = str(_option(args.jellyfin_host, env, "MAESTRO_JELLYFIN_HOST", "127.0.0.1")) jellyfin_port = _int_option(args.jellyfin_port, env, "MAESTRO_JELLYFIN_PORT", 8096) proxy_port = _int_option(args.proxy_port, env, "MAESTRO_JELLYFIN_PROXY_PORT", jellyfin_port + 1) config_value = _option(args.maestro_config, env, "MAESTRO_CONFIG", preset.maestro_config or "") fault_value = _option(args.jellyfin_fault, env, "MAESTRO_JELLYFIN_FAULT", "") fault = str(fault_value) or None if fault is not None and fault not in FAULTS: raise RunnerError(f"Unsupported MAESTRO_JELLYFIN_FAULT: {fault}") build_attempts = _int_option( args.jellyfin_build_attempts, env, "MAESTRO_JELLYFIN_BUILD_ATTEMPTS", 2, ) return RunnerConfig( command=args.command, jellyfin_host=jellyfin_host, jellyfin_port=jellyfin_port, proxy_port=proxy_port, jellyfin_image=str( _option(args.jellyfin_image, env, "MAESTRO_JELLYFIN_IMAGE", "plezy-jellyfin-demo:local") ), skip_jellyfin=_bool_option(args.skip_jellyfin, env, "MAESTRO_SKIP_JELLYFIN", False), skip_jellyfin_build=_bool_option( args.skip_jellyfin_build, env, "MAESTRO_SKIP_JELLYFIN_BUILD", False, ), skip_build=_bool_option(args.skip_build, env, "MAESTRO_SKIP_BUILD", False), jellyfin_fault=fault, use_adb_reverse=_bool_option( args.use_adb_reverse, env, "MAESTRO_USE_ADB_REVERSE", preset.use_adb_reverse, ), device_id=str(_option(args.device_id, env, "MAESTRO_DEVICE_ID", "")) or None, apk_path=_root_path( _option( args.apk_path, env, "MAESTRO_APK_PATH", "build/app/outputs/flutter-apk/app-debug.apk", ) ), flow_target=_root_path( _option(args.flow_target, env, "MAESTRO_FLOW_TARGET", preset.flow_target) ), maestro_config=_root_path(config_value) if config_value else None, uninstall_before_install=_bool_option( args.uninstall_before_install, env, "MAESTRO_UNINSTALL_BEFORE_INSTALL", preset.uninstall_before_install, ), diagnostics_dir=_root_path( _option(args.diagnostics_dir, env, "MAESTRO_DIAGNOSTICS_DIR", preset.diagnostics_dir) ), jellyfin_log=_root_path( _option(args.jellyfin_log, env, "MAESTRO_JELLYFIN_LOG", preset.jellyfin_log) ), proxy_log=_root_path( _option( args.proxy_log, env, "MAESTRO_JELLYFIN_PROXY_LOG", "build/maestro/jellyfin-proxy.log", ) ), proxy_journal=_root_path( _option( args.proxy_journal, env, "MAESTRO_JELLYFIN_PROXY_JOURNAL", "build/maestro/jellyfin-proxy-journal.jsonl", ) ), host_jellyfin_url=str( _option( args.host_jellyfin_url, env, "MAESTRO_JELLYFIN_HOST_URL", f"http://{jellyfin_host}:{jellyfin_port}", ) ), jellyfin_url=str(_option(args.jellyfin_url, env, "MAESTRO_JELLYFIN_URL", "")) or None, jellyfin_build_attempts=build_attempts, ) def _format_command(command: Sequence[object]) -> str: values = [str(value) for value in command] if os.name == "nt": return subprocess.list2cmdline(values) return shlex.join(values) def _run_checked(command: Sequence[object], **kwargs: object) -> subprocess.CompletedProcess[str]: values = [str(value) for value in command] print(f"+ {_format_command(values)}", flush=True) return subprocess.run(values, cwd=ROOT_DIR, check=True, text=True, **kwargs) def _require_commands(names: Sequence[str]) -> None: missing = [name for name in names if shutil.which(name) is None] if missing: raise RunnerError(f"Required command not found: {', '.join(missing)}") def flutter_build_command() -> tuple[str, ...]: return ( "flutter", "build", "apk", "--debug", "--dart-define=PLEZY_MAESTRO_E2E=true", ) def build_jellyfin_image(config: RunnerConfig) -> None: _require_commands(("docker",)) command = ( "docker", "build", "--file", ROOT_DIR / ".maestro/jellyfin-demo/Dockerfile", "--tag", config.jellyfin_image, ROOT_DIR, ) for attempt in range(1, config.jellyfin_build_attempts + 1): try: _run_checked(command) return except subprocess.CalledProcessError: if attempt == config.jellyfin_build_attempts: raise RunnerError( f"Jellyfin image build failed after {config.jellyfin_build_attempts} attempts" ) print(f"Jellyfin image build attempt {attempt} failed; retrying", file=sys.stderr) time.sleep(5) class MaestroRunner: def __init__(self, config: RunnerConfig) -> None: self.config = config self.device_id = config.device_id self.container_name: Optional[str] = None self.proxy_process: Optional[subprocess.Popen[str]] = None self.proxy_output: Optional[TextIO] = None self.reverse_configured = False self.device_service_port = config.jellyfin_port self.host_jellyfin_url = config.host_jellyfin_url self.device_settings: dict[tuple[str, str], str] = {} @property def adb_prefix(self) -> list[str]: prefix = ["adb"] if self.device_id: prefix.extend(("-s", self.device_id)) return prefix def run(self) -> None: required = ["adb", "maestro"] if not self.config.skip_build: required.append("flutter") if not self.config.skip_jellyfin: required.append("docker") _require_commands(required) self._select_device() self._prepare_output_directories() if not self.config.skip_jellyfin: if not self.config.skip_jellyfin_build: build_jellyfin_image(self.config) self._start_jellyfin() self._wait_for_health(self.host_jellyfin_url, attempts=180, interval=1, service="Jellyfin") if self.config.jellyfin_fault: self._start_proxy() if not self.config.skip_build: _run_checked(("flutter", "pub", "get")) _run_checked(flutter_build_command()) self._prepare_device() _run_checked((*self.adb_prefix, "install", "-r", self.config.apk_path)) _run_checked(self.maestro_command()) def maestro_command(self) -> list[str]: default_url = f"http://10.0.2.2:{self.device_service_port}" if self.config.use_adb_reverse: default_url = f"http://127.0.0.1:{self.device_service_port}" jellyfin_url = self.config.jellyfin_url or default_url command = ["maestro", "test", "-e", f"JELLYFIN_URL={jellyfin_url}"] if self.config.jellyfin_fault == "offline": command.extend(("-e", f"JELLYFIN_CONTROL_URL={self.host_jellyfin_url}")) if self.device_id: command.extend(("--device", self.device_id)) if self.config.maestro_config: command.extend(("--config", str(self.config.maestro_config))) command.append(str(self.config.flow_target)) return command def _prepare_output_directories(self) -> None: for path in ( self.config.diagnostics_dir, self.config.jellyfin_log.parent, self.config.proxy_log.parent, self.config.proxy_journal.parent, ): path.mkdir(parents=True, exist_ok=True) def _select_device(self) -> None: if not self.config.use_adb_reverse or self.device_id: return result = subprocess.run( ("adb", "devices"), cwd=ROOT_DIR, check=True, capture_output=True, text=True, ) devices = [] for line in result.stdout.splitlines()[1:]: fields = line.split() if len(fields) >= 2 and fields[1] == "device": devices.append(fields[0]) if len(devices) != 1: raise RunnerError("Set --device to the Android device serial when using --adb-reverse") self.device_id = devices[0] def _start_jellyfin(self) -> None: name = f"plezy-maestro-jellyfin-{self.config.jellyfin_port}-{os.getpid()}" result = _run_checked( ( "docker", "run", "--detach", "--rm", "--name", name, "--publish", f"{self.config.jellyfin_host}:{self.config.jellyfin_port}:8096", self.config.jellyfin_image, ), capture_output=True, ) self.container_name = result.stdout.strip() or name def _wait_for_health(self, base_url: str, *, attempts: int, interval: float, service: str) -> None: health_url = f"{base_url.rstrip('/')}/health" last_error: Optional[Exception] = None for _ in range(attempts): try: with urllib.request.urlopen(health_url, timeout=1) as response: status = response.read().decode(errors="replace").strip() if status.casefold() == "healthy": return last_error = RunnerError(f"{service} health status is {status or 'empty'}") except (OSError, urllib.error.URLError) as error: last_error = error time.sleep(interval) raise RunnerError(f"{service} did not become ready at {base_url}: {last_error}") def _start_proxy(self) -> None: self.proxy_output = self.config.proxy_log.open("w", encoding="utf-8") self.proxy_process = subprocess.Popen( ( sys.executable, str(ROOT_DIR / "scripts/maestro_jellyfin_proxy.py"), "--host", self.config.jellyfin_host, "--port", str(self.config.proxy_port), "--upstream", self.host_jellyfin_url, "--fault", self.config.jellyfin_fault or "", "--journal", str(self.config.proxy_journal), ), cwd=ROOT_DIR, stdout=self.proxy_output, stderr=subprocess.STDOUT, text=True, ) proxy_url = f"http://{self.config.jellyfin_host}:{self.config.proxy_port}" for _ in range(50): if self.proxy_process.poll() is not None: raise RunnerError(f"Jellyfin fault proxy exited early; see {self.config.proxy_log}") try: self._wait_for_health(proxy_url, attempts=1, interval=0, service="Jellyfin fault proxy") self.host_jellyfin_url = proxy_url self.device_service_port = self.config.proxy_port return except RunnerError: time.sleep(0.1) raise RunnerError(f"Jellyfin fault proxy did not become ready; see {self.config.proxy_log}") def _adb_run( self, *arguments: object, check: bool = True, quiet: bool = False, timeout: int = 30, ) -> subprocess.CompletedProcess[str]: kwargs: dict[str, object] = {} if quiet: kwargs.update(stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) return subprocess.run( [*self.adb_prefix, *(str(argument) for argument in arguments)], cwd=ROOT_DIR, check=check, text=True, timeout=timeout, **kwargs, ) def _adb_capture(self, *arguments: object) -> Optional[str]: command = [*self.adb_prefix, *(str(argument) for argument in arguments)] try: result = subprocess.run( command, cwd=ROOT_DIR, check=False, capture_output=True, text=True, timeout=10, ) except subprocess.TimeoutExpired as error: raise RunnerError(f"ADB command timed out: {_format_command(command)}") from error return result.stdout.strip() if result.returncode == 0 else None def _prepare_device(self) -> None: _run_checked((*self.adb_prefix, "wait-for-device"), timeout=60) for namespace, key in ( ("global", "stay_on_while_plugged_in"), ("secure", "immersive_mode_confirmations"), ("global", "hide_error_dialogs"), ): value = self._adb_capture("shell", "settings", "get", namespace, key) if value is not None: self.device_settings[(namespace, key)] = value self._adb_run( "shell", "settings", "put", "secure", "immersive_mode_confirmations", "confirmed", check=False, quiet=True, ) self._adb_run( "shell", "settings", "put", "global", "hide_error_dialogs", "1", check=False, quiet=True, ) self._adb_run("shell", "input", "keyevent", "KEYCODE_BACK", check=False, quiet=True) _run_checked((*self.adb_prefix, "shell", "svc", "power", "stayon", "true")) _run_checked((*self.adb_prefix, "shell", "input", "keyevent", "KEYCODE_WAKEUP")) _run_checked((*self.adb_prefix, "shell", "wm", "dismiss-keyguard")) if self.config.use_adb_reverse: _run_checked( ( *self.adb_prefix, "reverse", f"tcp:{self.device_service_port}", f"tcp:{self.device_service_port}", ) ) self.reverse_configured = True if self.config.uninstall_before_install: self._adb_run("uninstall", APP_ID, check=False, quiet=True) def collect_failure_diagnostics(self, exit_status: int) -> None: try: self.config.diagnostics_dir.mkdir(parents=True, exist_ok=True) state_path = self.config.diagnostics_dir / "run-state.txt" with state_path.open("w", encoding="utf-8") as output: output.write(f"exit_status={exit_status}\n") output.write(f"jellyfin_host_url={self.host_jellyfin_url}\n") output.write(f"jellyfin_container={self.container_name or ''}\n") output.write(f"jellyfin_fault={self.config.jellyfin_fault or ''}\n") output.write(f"proxy_pid={self.proxy_process.pid if self.proxy_process else ''}\n") try: with urllib.request.urlopen( f"{self.host_jellyfin_url.rstrip('/')}/health", timeout=2, ) as response: output.write(response.read().decode(errors="replace")) output.write("\n") except Exception as error: # Diagnostics must not hide the original failure. output.write(f"health_error={error}\n") if self.container_name: self._write_command_output( ("docker", "logs", self.container_name), self.config.jellyfin_log, ) self._write_command_output(("adb", "devices", "-l"), self.config.diagnostics_dir / "adb-devices.txt") for filename, arguments in ( ("device-properties.txt", ("shell", "getprop")), ("device-processes.txt", ("shell", "ps", "-A")), ("device-activities.txt", ("shell", "dumpsys", "activity", "activities")), ("device-windows.txt", ("shell", "dumpsys", "window", "windows")), ("device-logcat.txt", ("logcat", "-d", "-v", "threadtime")), ): self._write_command_output( (*self.adb_prefix, *arguments), self.config.diagnostics_dir / filename, ) if os.name == "nt" and shutil.which("tasklist"): self._write_command_output(("tasklist",), self.config.diagnostics_dir / "host-processes.txt") elif shutil.which("ps"): self._write_command_output(("ps", "-ef"), self.config.diagnostics_dir / "host-processes.txt") if shutil.which("lsof"): self._write_command_output( ("lsof", "-nP", f"-iTCP:{self.device_service_port}"), self.config.diagnostics_dir / "jellyfin-listeners.txt", ) except Exception as error: # Diagnostics are best effort. print(f"Failed to collect Maestro diagnostics: {error}", file=sys.stderr) def _write_command_output(self, command: Sequence[object], path: Path) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8") as output: subprocess.run( [str(value) for value in command], cwd=ROOT_DIR, check=False, stdout=output, stderr=subprocess.STDOUT, text=True, timeout=15, ) def _adb_best_effort(self, *arguments: object) -> None: try: self._adb_run(*arguments, check=False, quiet=True, timeout=5) except (OSError, subprocess.SubprocessError): pass def cleanup(self) -> None: for (namespace, key), value in self.device_settings.items(): operation = "delete" if value == "null" else "put" arguments = ["shell", "settings", operation, namespace, key] if operation == "put": arguments.append(value) self._adb_best_effort(*arguments) if self.reverse_configured: self._adb_best_effort( "reverse", "--remove", f"tcp:{self.device_service_port}", ) try: if self.proxy_process: self.proxy_process.terminate() try: self.proxy_process.wait(timeout=5) except subprocess.TimeoutExpired: self.proxy_process.kill() self.proxy_process.wait() finally: if self.proxy_output: self.proxy_output.close() if self.container_name: try: self._write_command_output( ("docker", "logs", self.container_name), self.config.jellyfin_log, ) except (OSError, subprocess.SubprocessError): pass try: subprocess.run( ("docker", "stop", "--time", "15", self.container_name), cwd=ROOT_DIR, check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, text=True, timeout=30, ) except (OSError, subprocess.SubprocessError): pass def run(config: RunnerConfig) -> int: if config.command == "build-image": build_jellyfin_image(config) return 0 runner = MaestroRunner(config) exit_status = 0 try: runner.run() except KeyboardInterrupt: exit_status = 130 print("Maestro run interrupted", file=sys.stderr) except RunnerSignal as error: exit_status = error.exit_status print(str(error), file=sys.stderr) except subprocess.CalledProcessError as error: exit_status = error.returncode or 1 print(f"Command failed ({exit_status}): {_format_command(error.cmd)}", file=sys.stderr) except (OSError, RunnerError) as error: exit_status = 1 print(error, file=sys.stderr) finally: if exit_status: runner.collect_failure_diagnostics(exit_status) runner.cleanup() return exit_status def main(argv: Optional[Sequence[str]] = None) -> int: signal.signal(signal.SIGINT, _raise_signal) signal.signal(signal.SIGTERM, _raise_signal) try: config = parse_config(argv) return run(config) except RunnerSignal as error: return error.exit_status except KeyboardInterrupt: return 130 except (OSError, RunnerError, subprocess.CalledProcessError) as error: print(error, file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())