test: stabilize deterministic integration coverage

scripts/maestro_jellyfin_proxy.py

@@ -18,6 +18,7 @@ import urllib.request
 _FAULT_PATHS = {
     "music-failure": lambda path: path.startswith("/Artists/AlbumArtists"),
     "recovery": lambda path: path.startswith("/Videos/") and "/stream" in path,
+    "offline": lambda _path: False,
 }
 _FORWARD_HEADERS = {
     "accept",
@@ -50,6 +51,7 @@ class ProxyState:
         self.journal = journal
         self._fault_injected = False
         self._sequence = 0
+        self._offline_enabled = False
         self._lock = threading.Lock()
         if journal is not None:
             journal.parent.mkdir(parents=True, exist_ok=True)
@@ -65,6 +67,14 @@ class ProxyState:
             self._fault_injected = True
             return True
 
+    def set_offline(self, enabled: bool) -> None:
+        with self._lock:
+            self._offline_enabled = enabled
+
+    def is_offline(self) -> bool:
+        with self._lock:
+            return self.fault == "offline" and self._offline_enabled
+
     def record(self, *, method: str, path: str, status: int, kind: str) -> None:
         if self.journal is None:
             return
@@ -111,15 +121,28 @@ class JellyfinProxyHandler(BaseHTTPRequestHandler):
         self._proxy()
 
     def _proxy(self) -> None:
-        if self.state.should_fault(self.path):
-            payload = json.dumps({"error": "temporary Maestro fault"}).encode("utf-8")
+        if urllib.parse.urlsplit(self.path).path == "/__maestro/offline":
+            self._set_offline()
+            return
+        if self.state.is_offline():
+            payload = json.dumps({"error": "Maestro offline mode"}).encode("utf-8")
+            self.state.record(method=self.command, path=self.path, status=503, kind="offline")
             self.send_response(HTTPStatus.SERVICE_UNAVAILABLE)
             self.send_header("Content-Type", "application/json")
             self.send_header("Content-Length", str(len(payload)))
             self.end_headers()
             if self.command != "HEAD":
                 self.wfile.write(payload)
+            return
+        if self.state.should_fault(self.path):
+            payload = json.dumps({"error": "temporary Maestro fault"}).encode("utf-8")
             self.state.record(method=self.command, path=self.path, status=503, kind="fault")
+            self.send_response(HTTPStatus.SERVICE_UNAVAILABLE)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(payload)))
+            self.end_headers()
+            if self.command != "HEAD":
+                self.wfile.write(payload)
             return
 
         content_length = int(self.headers.get("Content-Length", "0"))
@@ -149,6 +172,7 @@ class JellyfinProxyHandler(BaseHTTPRequestHandler):
             status = HTTPStatus.BAD_GATEWAY
             response_headers = {"Content-Type": "application/json"}
 
+        self.state.record(method=self.command, path=self.path, status=int(status), kind="request")
         self.send_response(status)
         for name, value in response_headers.items():
             if name.lower() in _RESPONSE_HEADERS:
@@ -160,7 +184,19 @@ class JellyfinProxyHandler(BaseHTTPRequestHandler):
                 self.wfile.write(payload)
             except (BrokenPipeError, ConnectionResetError):
                 pass
-        self.state.record(method=self.command, path=self.path, status=int(status), kind="request")
+
+    def _set_offline(self) -> None:
+        content_length = int(self.headers.get("Content-Length", "0"))
+        body = self.rfile.read(content_length) if content_length else b"{}"
+        try:
+            enabled = bool(json.loads(body).get("enabled", True))
+        except (AttributeError, json.JSONDecodeError):
+            self.send_error(HTTPStatus.BAD_REQUEST)
+            return
+        self.state.set_offline(enabled)
+        self.state.record(method=self.command, path=self.path, status=204, kind="control")
+        self.send_response(HTTPStatus.NO_CONTENT)
+        self.end_headers()
 
     def log_message(self, format: str, *args: Any) -> None:
         print(f"jellyfin-proxy: {format % args}")

scripts/run_maestro.py

@@ -20,7 +20,7 @@ import urllib.request
 
 ROOT_DIR = Path(__file__).resolve().parent.parent
 APP_ID = "com.edde746.plezy"
-FAULTS = ("music-failure", "recovery")
+FAULTS = ("music-failure", "offline", "recovery")
 
 
 class RunnerError(RuntimeError):
@@ -316,6 +316,16 @@ def _require_commands(names: Sequence[str]) -> None:
         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 = (
@@ -373,14 +383,14 @@ class MaestroRunner:
             if not self.config.skip_jellyfin_build:
                 build_jellyfin_image(self.config)
             self._start_jellyfin()
-        self._wait_for_health(self.host_jellyfin_url, attempts=120, interval=0.25, service="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", "apk", "--debug"))
+            _run_checked(flutter_build_command())
 
         self._prepare_device()
         _run_checked((*self.adb_prefix, "install", "-r", self.config.apk_path))
@@ -392,6 +402,8 @@ class MaestroRunner:
             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:
@@ -451,11 +463,13 @@ class MaestroRunner:
         for _ in range(attempts):
             try:
                 with urllib.request.urlopen(health_url, timeout=1) as response:
-                    response.read()
-                return
+                    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)
+            time.sleep(interval)
         raise RunnerError(f"{service} did not become ready at {base_url}: {last_error}")
 
     def _start_proxy(self) -> None:

scripts/run_maestro_ci.py

@@ -9,6 +9,12 @@ from collections.abc import Sequence
 import run_maestro
 
 
+ANDROID_15_INSTRUMENTATION_CLASSES = (
+    "androidx.media3.decoder.ffmpeg.PlezyFfmpegPlaybackTest,"
+    "com.edde746.plezy.exoplayer.PlezyAudioModePlaybackTest"
+)
+
+
 GROUPS: dict[str, tuple[tuple[str, ...], ...]] = {
     "android-15": (
         ("basic",),
@@ -54,6 +60,21 @@ GROUPS: dict[str, tuple[tuple[str, ...], ...]] = {
             "--diagnostics-dir",
             "build/maestro-recovery/music-diagnostics",
         ),
+        (
+            "basic",
+            "--fault",
+            "offline",
+            "--flow",
+            ".maestro/flows/09_download_offline_playback.yaml",
+            "--jellyfin-log",
+            "build/maestro-offline/jellyfin.log",
+            "--proxy-log",
+            "build/maestro-offline/jellyfin-proxy.log",
+            "--proxy-journal",
+            "build/maestro-offline/proxy-journal.jsonl",
+            "--diagnostics-dir",
+            "build/maestro-offline/diagnostics",
+        ),
         (
             "basic",
             "--fault",
@@ -82,7 +103,24 @@ GROUPS: dict[str, tuple[tuple[str, ...], ...]] = {
 }
 
 
+def run_android_15_instrumentation() -> None:
+    print("==> Android 15 filtered instrumentation", flush=True)
+    run_maestro._run_checked(
+        (
+            "android/gradlew",
+            "-p",
+            "android",
+            ":app:connectedDebugAndroidTest",
+            "-x",
+            ":app:compileFlutterBuildDebug",
+            f"-Pandroid.testInstrumentationRunnerArguments.class={ANDROID_15_INSTRUMENTATION_CLASSES}",
+        )
+    )
+
+
 def run_group(name: str) -> int:
+    if name == "android-15":
+        run_android_15_instrumentation()
     failed = False
     for arguments in GROUPS[name]:
         print(f"==> Maestro {' '.join(arguments)}", flush=True)

scripts/test_maestro_flow_contracts.py

@@ -0,0 +1,200 @@
+#!/usr/bin/env python3
+
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+import re
+import unittest
+from typing import Any
+
+
+ROOT_DIR = Path(__file__).resolve().parents[1]
+
+
+def _mapping_separator(value: str) -> int | None:
+    quote: str | None = None
+    escaped = False
+    for index, character in enumerate(value):
+        if escaped:
+            escaped = False
+            continue
+        if character == "\\" and quote is not None:
+            escaped = True
+            continue
+        if character in {'"', "'"}:
+            if quote is None:
+                quote = character
+            elif quote == character:
+                quote = None
+            continue
+        if character == ":" and quote is None and (
+            index + 1 == len(value) or value[index + 1].isspace()
+        ):
+            return index
+    return None
+
+
+def _scalar(value: str) -> Any:
+    if value.startswith(('"', "'")):
+        return ast.literal_eval(value)
+    if value == "true":
+        return True
+    if value == "false":
+        return False
+    if value in {"null", "~"}:
+        return None
+    if re.fullmatch(r"-?\d+", value):
+        return int(value)
+    return value
+
+
+def _parse_node(lines: list[tuple[int, str]], index: int, indent: int) -> tuple[Any, int]:
+    if lines[index][0] != indent:
+        raise ValueError(f"unexpected indentation at line {index + 1}")
+    if lines[index][1].startswith("-"):
+        result: list[Any] = []
+        while index < len(lines) and lines[index][0] == indent and lines[index][1].startswith("-"):
+            item = lines[index][1][1:].lstrip()
+            index += 1
+            separator = _mapping_separator(item)
+            if separator is None:
+                result.append(_scalar(item))
+                continue
+
+            key = item[:separator]
+            value = item[separator + 1 :].lstrip()
+            if value:
+                result.append({key: _scalar(value)})
+            elif index < len(lines) and lines[index][0] > indent:
+                child, index = _parse_node(lines, index, lines[index][0])
+                result.append({key: child})
+            else:
+                result.append({key: None})
+        return result, index
+
+    result_map: dict[str, Any] = {}
+    while index < len(lines) and lines[index][0] == indent and not lines[index][1].startswith("-"):
+        item = lines[index][1]
+        separator = _mapping_separator(item)
+        if separator is None:
+            raise ValueError(f"expected mapping at line {index + 1}")
+        key = item[:separator]
+        value = item[separator + 1 :].lstrip()
+        index += 1
+        if value:
+            result_map[key] = _scalar(value)
+        elif index < len(lines) and lines[index][0] > indent:
+            result_map[key], index = _parse_node(lines, index, lines[index][0])
+        else:
+            result_map[key] = None
+    return result_map, index
+
+
+def load_flow(relative_path: str) -> list[dict[str, Any]]:
+    contents = (ROOT_DIR / relative_path).read_text(encoding="utf-8")
+    try:
+        commands = contents.split("---", maxsplit=1)[1]
+    except IndexError as error:
+        raise ValueError(f"{relative_path} has no Maestro command document") from error
+    lines = [
+        (len(line) - len(line.lstrip(" ")), line.lstrip(" "))
+        for line in commands.splitlines()
+        if line.strip() and not line.lstrip().startswith("#")
+    ]
+    parsed, next_index = _parse_node(lines, 0, 0)
+    if next_index != len(lines) or not isinstance(parsed, list):
+        raise ValueError(f"could not parse all commands in {relative_path}")
+    return parsed
+
+
+def command_name(step: dict[str, Any]) -> str:
+    if len(step) != 1:
+        raise ValueError(f"expected one command per step, got {step!r}")
+    return next(iter(step))
+
+
+def platform_pair(steps: list[dict[str, Any]], index: int) -> dict[str, list[dict[str, Any]]]:
+    pair = steps[index : index + 2]
+    if len(pair) != 2 or [command_name(step) for step in pair] != ["runFlow", "runFlow"]:
+        raise AssertionError(f"expected adjacent platform runFlow pair at command {index}")
+    branches = {step["runFlow"]["when"]["platform"]: step["runFlow"]["commands"] for step in pair}
+    if set(branches) != {"iOS", "Android"}:
+        raise AssertionError(f"expected iOS and Android branches, got {set(branches)}")
+    return branches
+
+
+class MaestroFlowContractTests(unittest.TestCase):
+    def test_tv_result_selector_cannot_select_the_query_field(self) -> None:
+        steps = load_flow(".maestro/regression_flows/05_tv_next_episode_back.yaml")
+        observation = next(
+            step["extendedWaitUntil"]
+            for step in steps
+            if command_name(step) == "extendedWaitUntil"
+            and isinstance(step["extendedWaitUntil"], dict)
+            and "TV show" in str(step["extendedWaitUntil"].get("visible", ""))
+        )
+        selector = observation["visible"]
+
+        self.assertIsNotNone(re.fullmatch(selector, "Maestro Show, TV show, unwatched"))
+        self.assertIsNotNone(re.fullmatch(selector, "Maestro Show, TV show, watched"))
+        self.assertIsNone(re.fullmatch(selector, "Maestro Show"))
+        self.assertIn({"tapOn": selector}, steps)
+
+    def test_offline_fault_is_injected_after_movies_tab_content_is_observed(self) -> None:
+        steps = load_flow(".maestro/flows/09_download_offline_playback.yaml")
+        movies_index = next(
+            index
+            for index, step in enumerate(steps)
+            if step.get("tapOn") == {"text": "Movies", "waitToSettleTimeoutMs": 3000}
+        )
+        observation_index = next(
+            index
+            for index, step in enumerate(steps[movies_index + 1 :], movies_index + 1)
+            if step.get("extendedWaitUntil", {}).get("visible") == "(?s).*Alpha Archive.*"
+        )
+        offline_index = next(
+            index
+            for index, step in enumerate(steps)
+            if step.get("runScript", {}).get("file") == "../scripts/set_jellyfin_offline.js"
+        )
+
+        self.assertLess(movies_index, observation_index)
+        self.assertLess(observation_index, offline_index)
+        self.assertEqual(steps[observation_index]["extendedWaitUntil"]["timeout"], 60000)
+
+    def test_tv_next_episode_dismissal_has_platform_specific_controls(self) -> None:
+        steps = load_flow(".maestro/regression_flows/05_tv_next_episode_back.yaml")
+        next_episode_index = next(
+            index
+            for index, step in enumerate(steps)
+            if step.get("extendedWaitUntil", {}).get("visible") == "Next Episode"
+        )
+        branches = platform_pair(steps, next_episode_index + 1)
+
+        self.assertEqual(branches["iOS"], [{"tapOn": "Cancel"}])
+        self.assertEqual(branches["Android"], [{"pressKey": "back"}])
+
+    def test_offline_flow_back_and_close_controls_are_platform_specific(self) -> None:
+        steps = load_flow(".maestro/flows/09_download_offline_playback.yaml")
+        platform_pair_indices = [
+            index
+            for index in range(len(steps) - 1)
+            if command_name(steps[index]) == "runFlow"
+            and command_name(steps[index + 1]) == "runFlow"
+            and steps[index]["runFlow"].get("when", {}).get("platform") == "iOS"
+            and steps[index + 1]["runFlow"].get("when", {}).get("platform") == "Android"
+        ]
+        self.assertEqual(len(platform_pair_indices), 2)
+        detail_close = platform_pair(steps, platform_pair_indices[0])
+        player_close = platform_pair(steps, platform_pair_indices[1])
+
+        self.assertEqual(detail_close["iOS"], [{"tapOn": {"point": "6%, 9%"}}])
+        self.assertEqual(detail_close["Android"], ["back"])
+        self.assertEqual(player_close["iOS"][0], {"tapOn": {"point": "90%, 5%"}})
+        self.assertEqual([command for command in player_close["Android"] if command == "back"], ["back", "back"])
+        self.assertNotIn("back", steps)
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/test_maestro_jellyfin_proxy.py

@@ -47,6 +47,16 @@ class _UpstreamHandler(BaseHTTPRequestHandler):
     def log_message(self, format: str, *args: object) -> None:
         pass
 
+class _ResponseFailureHandler:
+    def __init__(self, state: ProxyState, path: str) -> None:
+        self.state = state
+        self.path = path
+        self.command = "GET"
+        self.headers: dict[str, str] = {}
+
+    def send_response(self, status: int) -> None:
+        raise BrokenPipeError(f"client disconnected before status {status}")
+
 
 class JellyfinProxyTests(unittest.TestCase):
     def setUp(self) -> None:
@@ -121,6 +131,40 @@ class JellyfinProxyTests(unittest.TestCase):
         finally:
             self._stop_proxy(proxy, thread)
 
+    def test_offline_control_blocks_requests_until_reenabled(self) -> None:
+        proxy, thread, base_url, journal = self._start_proxy("offline")
+        try:
+            enable = urllib.request.Request(
+                base_url + "/__maestro/offline",
+                data=b'{"enabled":true}',
+                method="POST",
+                headers={"Content-Type": "application/json"},
+            )
+            with urllib.request.urlopen(enable) as response:
+                self.assertEqual(response.status, 204)
+
+            with self.assertRaises(urllib.error.HTTPError) as failure:
+                urllib.request.urlopen(base_url + "/Items")
+            self.assertEqual(failure.exception.code, 503)
+            failure.exception.close()
+            self.assertEqual(_UpstreamHandler.requests, [])
+
+            disable = urllib.request.Request(
+                base_url + "/__maestro/offline",
+                data=b'{"enabled":false}',
+                method="POST",
+                headers={"Content-Type": "application/json"},
+            )
+            with urllib.request.urlopen(disable) as response:
+                self.assertEqual(response.status, 204)
+            with urllib.request.urlopen(base_url + "/Items") as response:
+                self.assertEqual(response.status, 200)
+
+            events = [json.loads(line) for line in journal.read_text(encoding="utf-8").splitlines()]
+            self.assertEqual([event["kind"] for event in events], ["control", "offline", "control", "request"])
+        finally:
+            self._stop_proxy(proxy, thread)
+
     def test_music_fault_does_not_affect_other_requests(self) -> None:
         proxy, thread, base_url, _ = self._start_proxy("music-failure")
         try:
@@ -135,6 +179,27 @@ class JellyfinProxyTests(unittest.TestCase):
         finally:
             self._stop_proxy(proxy, thread)
 
+    def test_records_events_before_response_write(self) -> None:
+        upstream_url = f"http://127.0.0.1:{self.upstream.server_port}"
+        cases = [
+            ("fault", "recovery", "/Videos/movie/stream.mp4?Static=true", "fault", 503),
+            ("request", None, "/Items", "request", 200),
+        ]
+
+        for name, fault, path, expected_kind, expected_status in cases:
+            with self.subTest(name=name):
+                journal = Path(self.temp_dir.name) / f"{name}.jsonl"
+                state = ProxyState(upstream_url, fault, journal)
+                handler = _ResponseFailureHandler(state, path)
+
+                with self.assertRaises(BrokenPipeError):
+                    JellyfinProxyHandler._proxy(handler)
+
+                events = [json.loads(line) for line in journal.read_text(encoding="utf-8").splitlines()]
+                self.assertEqual(len(events), 1)
+                self.assertEqual(events[0]["kind"], expected_kind)
+                self.assertEqual(events[0]["status"], expected_status)
+
 
 if __name__ == "__main__":
     unittest.main()

scripts/test_run_maestro.py

@@ -9,7 +9,7 @@ from pathlib import Path
 import subprocess
 import sys
 import unittest
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 sys.path.insert(0, str(Path(__file__).resolve().parent))
 
@@ -94,6 +94,30 @@ class CommandTests(unittest.TestCase):
             ],
         )
 
+    def test_offline_fault_exposes_host_proxy_control_url(self) -> None:
+        config = run_maestro.parse_config(["basic", "--fault", "offline", "--adb-reverse"], {})
+        runner = run_maestro.MaestroRunner(config)
+        runner.host_jellyfin_url = "http://127.0.0.1:8097"
+        runner.device_service_port = 8097
+
+        command = runner.maestro_command()
+
+        self.assertIn("JELLYFIN_URL=http://127.0.0.1:8097", command)
+        self.assertIn("JELLYFIN_CONTROL_URL=http://127.0.0.1:8097", command)
+
+    def test_flutter_build_enables_stable_physical_device_controls(self) -> None:
+        self.assertEqual(
+            run_maestro.flutter_build_command(),
+            (
+                "flutter",
+                "build",
+                "apk",
+                "--debug",
+                "--dart-define=PLEZY_MAESTRO_E2E=true",
+            ),
+        )
+
+
     def test_explicit_device_url_wins_over_network_mode(self) -> None:
         config = run_maestro.parse_config(
             ["basic", "--adb-reverse", "--jellyfin-url", "http://device.test:9000"],
@@ -134,6 +158,21 @@ class LifecycleTests(unittest.TestCase):
         self.assertEqual(run_command.call_count, 2)
         sleep.assert_called_once_with(5)
 
+    def test_health_wait_rejects_degraded_until_healthy(self) -> None:
+        runner = run_maestro.MaestroRunner(run_maestro.parse_config([], {}))
+        degraded = MagicMock()
+        degraded.__enter__.return_value.read.return_value = b"Degraded"
+        healthy = MagicMock()
+        healthy.__enter__.return_value.read.return_value = b" Healthy\n"
+
+        with (
+            patch.object(run_maestro.urllib.request, "urlopen", side_effect=[degraded, healthy]),
+            patch.object(run_maestro.time, "sleep") as sleep,
+        ):
+            runner._wait_for_health("http://jellyfin.test", attempts=2, interval=0.25, service="Jellyfin")
+
+        sleep.assert_called_once_with(0.25)
+
 
 class CiGroupTests(unittest.TestCase):
     def test_android_15_group_runs_every_suite_after_failure(self) -> None:
@@ -142,12 +181,14 @@ class CiGroupTests(unittest.TestCase):
 
         with (
             patch.object(run_maestro_ci.run_maestro, "main", side_effect=statuses) as run,
+            patch.object(run_maestro_ci, "run_android_15_instrumentation") as instrumentation,
             redirect_stdout(io.StringIO()),
         ):
             exit_status = run_maestro_ci.run_group("android-15")
 
         self.assertEqual(exit_status, 1)
         self.assertEqual(run.call_count, expected_runs)
+        instrumentation.assert_called_once_with()
 
     def test_group_recipes_are_valid_runner_invocations(self) -> None:
         for recipes in run_maestro_ci.GROUPS.values():
@@ -158,12 +199,14 @@ class CiGroupTests(unittest.TestCase):
     def test_group_stops_after_interruption(self) -> None:
         with (
             patch.object(run_maestro_ci.run_maestro, "main", return_value=143) as run,
+            patch.object(run_maestro_ci, "run_android_15_instrumentation") as instrumentation,
             redirect_stdout(io.StringIO()),
         ):
             exit_status = run_maestro_ci.run_group("android-15")
 
         self.assertEqual(exit_status, 143)
         run.assert_called_once_with(("basic",))
+        instrumentation.assert_called_once_with()
 
 
 if __name__ == "__main__":