plezy/scripts/test_run_maestro.py

#!/usr/bin/env python3

from __future__ import annotations

from contextlib import redirect_stderr, redirect_stdout
from dataclasses import replace
import io
from pathlib import Path
import subprocess
import sys
import unittest
from unittest.mock import MagicMock, patch

sys.path.insert(0, str(Path(__file__).resolve().parent))

import run_maestro  # noqa: E402
import run_maestro_ci  # noqa: E402


class ParseConfigTests(unittest.TestCase):
    def test_basic_defaults(self) -> None:
        config = run_maestro.parse_config([], {})

        self.assertEqual(config.command, "basic")
        self.assertEqual(config.flow_target, run_maestro.ROOT_DIR / ".maestro")
        self.assertIsNone(config.maestro_config)
        self.assertFalse(config.use_adb_reverse)
        self.assertFalse(config.uninstall_before_install)

    def test_suite_presets_replace_shell_wrappers(self) -> None:
        catalog = run_maestro.parse_config(["catalog"], {})
        media = run_maestro.parse_config(["media"], {})

        self.assertEqual(catalog.flow_target, run_maestro.ROOT_DIR / ".maestro/real_flows")
        self.assertEqual(
            catalog.diagnostics_dir,
            run_maestro.ROOT_DIR / "build/maestro-real-jellyfin/diagnostics",
        )
        self.assertTrue(catalog.uninstall_before_install)
        self.assertEqual(media.flow_target, run_maestro.ROOT_DIR / ".maestro/media_flows")
        self.assertEqual(media.maestro_config, run_maestro.ROOT_DIR / ".maestro/media-config.yaml")
        self.assertTrue(media.use_adb_reverse)
        self.assertTrue(media.uninstall_before_install)

    def test_cli_options_override_compatible_environment_values(self) -> None:
        config = run_maestro.parse_config(
            [
                "media",
                "--no-adb-reverse",
                "--flow",
                "custom/flow.yaml",
                "--device",
                "cli-device",
            ],
            {
                "MAESTRO_USE_ADB_REVERSE": "1",
                "MAESTRO_FLOW_TARGET": "environment/flow.yaml",
                "MAESTRO_DEVICE_ID": "environment-device",
                "MAESTRO_SKIP_BUILD": "true",
            },
        )

        self.assertFalse(config.use_adb_reverse)
        self.assertEqual(config.flow_target, run_maestro.ROOT_DIR / "custom/flow.yaml")
        self.assertEqual(config.device_id, "cli-device")
        self.assertTrue(config.skip_build)

    def test_invalid_environment_values_fail_early(self) -> None:
        with self.assertRaisesRegex(run_maestro.RunnerError, "MAESTRO_SKIP_BUILD"):
            run_maestro.parse_config([], {"MAESTRO_SKIP_BUILD": "sometimes"})
        with self.assertRaisesRegex(run_maestro.RunnerError, "MAESTRO_JELLYFIN_PORT"):
            run_maestro.parse_config([], {"MAESTRO_JELLYFIN_PORT": "invalid"})
        with self.assertRaisesRegex(run_maestro.RunnerError, "MAESTRO_JELLYFIN_BUILD_ATTEMPTS"):
            run_maestro.parse_config([], {"MAESTRO_JELLYFIN_BUILD_ATTEMPTS": "0"})


class CommandTests(unittest.TestCase):
    def test_media_command_contains_resolved_preset_and_device_url(self) -> None:
        config = run_maestro.parse_config(["media", "--device", "emulator-5554"], {})
        command = run_maestro.MaestroRunner(config).maestro_command()

        self.assertEqual(
            command,
            [
                "maestro",
                "test",
                "-e",
                "JELLYFIN_URL=http://127.0.0.1:8096",
                "--device",
                "emulator-5554",
                "--config",
                str(run_maestro.ROOT_DIR / ".maestro/media-config.yaml"),
                str(run_maestro.ROOT_DIR / ".maestro/media_flows"),
            ],
        )

    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"],
            {},
        )

        command = run_maestro.MaestroRunner(config).maestro_command()

        self.assertIn("JELLYFIN_URL=http://device.test:9000", command)


class LifecycleTests(unittest.TestCase):
    def test_runner_failure_collects_diagnostics_and_cleans_up(self) -> None:
        config = run_maestro.parse_config([], {})
        with patch.object(run_maestro, "MaestroRunner") as runner_type:
            runner = runner_type.return_value
            runner.run.side_effect = run_maestro.RunnerError("failed")
            with redirect_stderr(io.StringIO()):
                exit_status = run_maestro.run(config)

        self.assertEqual(exit_status, 1)
        runner.collect_failure_diagnostics.assert_called_once_with(1)
        runner.cleanup.assert_called_once_with()

    def test_image_build_retries_once(self) -> None:
        config = replace(run_maestro.parse_config(["build-image"], {}), jellyfin_build_attempts=2)
        failure = subprocess.CalledProcessError(1, ["docker", "build"])
        success = subprocess.CompletedProcess(["docker", "build"], 0)

        with (
            patch.object(run_maestro, "_require_commands"),
            patch.object(run_maestro, "_run_checked", side_effect=[failure, success]) as run_command,
            patch.object(run_maestro.time, "sleep") as sleep,
            redirect_stderr(io.StringIO()),
        ):
            run_maestro.build_jellyfin_image(config)

        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:
        expected_runs = len(run_maestro_ci.GROUPS["android-15"])
        statuses = [0, 1, *([0] * (expected_runs - 2))]

        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():
            for arguments in recipes:
                with self.subTest(arguments=arguments):
                    run_maestro.parse_config(arguments, {})

    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__":
    unittest.main()