edde746
0c58b2dc55
Two workflow guards had drifted from the code they describe, so `scripts/ci_guard_checks.sh` failed on a clean tree. The Flutter release-tag pin moved out of build.yml into the shared setup-flutter-git composite action, but the checker read that action from a fixed repository path while its test mutated a workflow fixture. The mutation could not reach the checker, so the rejection test asserted against an unmodified run. The checker now resolves the action beside the workflow it is given, and the test materialises a `.github` tree so the pin is genuinely exercised. The script-test roster likewise moved into ci_guard_checks.sh, which discovers `scripts/test_*.py` by glob; the dispatch guard still expected each one to be named explicitly in ci_checks.sh and ci.yml. It now reads that glob and checks both aggregates delegate to the shared roster.
439 lines
17 KiB
Python
Executable File
439 lines
17 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
from contextlib import redirect_stderr, redirect_stdout
|
|
from dataclasses import replace
|
|
import io
|
|
from pathlib import Path
|
|
import re
|
|
import shlex
|
|
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
|
|
|
|
|
|
ROOT_DIR = Path(__file__).resolve().parent.parent
|
|
SCRIPTS_DIR = ROOT_DIR / "scripts"
|
|
LOCAL_SCRIPT_TEST_DISPATCHER = SCRIPTS_DIR / "ci_checks.sh"
|
|
CI_SCRIPT_TEST_DISPATCHER = ROOT_DIR / ".github/workflows/ci.yml"
|
|
# The guard roster both dispatchers above delegate to. It discovers the script
|
|
# tests by glob, so a new scripts/test_*.py is picked up without being listed.
|
|
GUARD_SCRIPT_TEST_DISPATCHER = SCRIPTS_DIR / "ci_guard_checks.sh"
|
|
E2E_WORKFLOW = ROOT_DIR / ".github/workflows/e2e.yml"
|
|
SCRIPT_TEST_DISPATCHERS = (
|
|
LOCAL_SCRIPT_TEST_DISPATCHER,
|
|
CI_SCRIPT_TEST_DISPATCHER,
|
|
GUARD_SCRIPT_TEST_DISPATCHER,
|
|
SCRIPTS_DIR / "ci_website_checks.sh",
|
|
)
|
|
REGRESSION_FLOWS_DIR = ROOT_DIR / ".maestro/regression_flows"
|
|
|
|
|
|
def _is_main_guard(expression: ast.expr) -> bool:
|
|
if (
|
|
not isinstance(expression, ast.Compare)
|
|
or len(expression.ops) != 1
|
|
or not isinstance(expression.ops[0], ast.Eq)
|
|
or len(expression.comparators) != 1
|
|
):
|
|
return False
|
|
left = expression.left
|
|
right = expression.comparators[0]
|
|
return (
|
|
isinstance(left, ast.Name)
|
|
and left.id == "__name__"
|
|
and isinstance(right, ast.Constant)
|
|
and right.value == "__main__"
|
|
) or (
|
|
isinstance(right, ast.Name)
|
|
and right.id == "__name__"
|
|
and isinstance(left, ast.Constant)
|
|
and left.value == "__main__"
|
|
)
|
|
|
|
|
|
def _executable_script_tests() -> set[str]:
|
|
executable = set()
|
|
for path in SCRIPTS_DIR.glob("test_*.py"):
|
|
module = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
|
|
if any(isinstance(node, ast.If) and _is_main_guard(node.test) for node in module.body):
|
|
executable.add(path.name)
|
|
return executable
|
|
|
|
|
|
def _dispatched_script_tests(path: Path) -> list[str]:
|
|
dispatched = []
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
stripped = line.strip()
|
|
# `for guard_test in scripts/test_*.py; do` dispatches the whole roster.
|
|
loop = re.match(r"for\s+\w+\s+in\s+(scripts/test_[^;\s]*\.py)\s*;?\s*(?:do)?$", stripped)
|
|
if loop:
|
|
dispatched.extend(sorted(match.name for match in ROOT_DIR.glob(loop.group(1))))
|
|
continue
|
|
try:
|
|
command = shlex.split(stripped, comments=True)
|
|
except ValueError:
|
|
continue
|
|
if len(command) < 2 or Path(command[0]).name not in {"python", "python3"}:
|
|
continue
|
|
script_name = Path(command[1]).name
|
|
if script_name.startswith("test_") and script_name.endswith(".py"):
|
|
dispatched.append(script_name)
|
|
return dispatched
|
|
|
|
|
|
def _dispatched_maestro_ci_targets(path: Path) -> set[str]:
|
|
targets = set()
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
try:
|
|
command = shlex.split(line.strip(), comments=True)
|
|
except ValueError:
|
|
continue
|
|
for index, token in enumerate(command[:-1]):
|
|
if Path(token).name == "run_maestro_ci.py":
|
|
targets.add(command[index + 1])
|
|
return targets
|
|
|
|
|
|
def _registered_regression_flows(
|
|
targets: dict[str, tuple[tuple[str, ...], ...]],
|
|
) -> set[str]:
|
|
registered = set()
|
|
for recipes in targets.values():
|
|
for arguments in recipes:
|
|
flow_target = run_maestro.parse_config(arguments, {}).flow_target
|
|
if flow_target.parent == REGRESSION_FLOWS_DIR:
|
|
registered.add(flow_target.relative_to(ROOT_DIR).as_posix())
|
|
return registered
|
|
|
|
|
|
class ScriptTestDispatchTests(unittest.TestCase):
|
|
def test_every_executable_script_test_has_a_dispatcher(self) -> None:
|
|
dispatched = {
|
|
script
|
|
for dispatcher in SCRIPT_TEST_DISPATCHERS
|
|
for script in _dispatched_script_tests(dispatcher)
|
|
}
|
|
|
|
self.assertSetEqual(dispatched, _executable_script_tests())
|
|
|
|
def test_real_jellyfin_fixture_test_is_in_local_and_ci_guards(self) -> None:
|
|
# Both aggregates reach the roster through the shared guard script, so
|
|
# the fixture test is covered exactly when the glob picks it up.
|
|
for dispatcher in (LOCAL_SCRIPT_TEST_DISPATCHER, CI_SCRIPT_TEST_DISPATCHER):
|
|
with self.subTest(dispatcher=dispatcher):
|
|
self.assertIn(
|
|
f"bash {GUARD_SCRIPT_TEST_DISPATCHER.relative_to(ROOT_DIR).as_posix()}",
|
|
dispatcher.read_text(encoding="utf-8"),
|
|
)
|
|
|
|
self.assertEqual(
|
|
_dispatched_script_tests(GUARD_SCRIPT_TEST_DISPATCHER).count("test_maestro_real_jellyfin.py"),
|
|
1,
|
|
)
|
|
|
|
|
|
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,
|
|
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)
|
|
|
|
def test_android_15_instrumentation_has_an_isolated_target(self) -> None:
|
|
with (
|
|
patch.object(run_maestro_ci, "run_android_15_instrumentation") as instrumentation,
|
|
redirect_stdout(io.StringIO()),
|
|
):
|
|
exit_status = run_maestro_ci.run_target(run_maestro_ci.ANDROID_15_INSTRUMENTATION_TARGET)
|
|
|
|
self.assertEqual(exit_status, 0)
|
|
instrumentation.assert_called_once_with()
|
|
|
|
def test_all_automatic_and_manual_recipes_are_valid_runner_invocations(
|
|
self,
|
|
) -> None:
|
|
target_sets = (
|
|
("automatic", run_maestro_ci.GROUPS),
|
|
("destructive-manual", run_maestro_ci.DESTRUCTIVE_MANUAL_TARGETS),
|
|
)
|
|
for classification, targets in target_sets:
|
|
for target, recipes in targets.items():
|
|
for arguments in recipes:
|
|
with self.subTest(
|
|
classification=classification,
|
|
target=target,
|
|
arguments=arguments,
|
|
):
|
|
run_maestro.parse_config(arguments, {})
|
|
|
|
def test_top_level_regression_flow_inventory_is_complete(self) -> None:
|
|
inventory = {
|
|
path.relative_to(ROOT_DIR).as_posix()
|
|
for path in REGRESSION_FLOWS_DIR.glob("*.yaml")
|
|
}
|
|
automatic = _registered_regression_flows(run_maestro_ci.GROUPS)
|
|
destructive_manual = _registered_regression_flows(
|
|
run_maestro_ci.DESTRUCTIVE_MANUAL_TARGETS
|
|
)
|
|
|
|
self.assertFalse(automatic & destructive_manual)
|
|
self.assertSetEqual(automatic | destructive_manual, inventory)
|
|
|
|
def test_profile_regressions_are_only_in_the_destructive_manual_target(self) -> None:
|
|
profile_flows = {
|
|
".maestro/regression_flows/01_profile_switch_isolation.yaml",
|
|
".maestro/regression_flows/02_profile_teardown.yaml",
|
|
}
|
|
manual_profile_flows = _registered_regression_flows(
|
|
{
|
|
"profile-regressions": run_maestro_ci.DESTRUCTIVE_MANUAL_TARGETS[
|
|
"profile-regressions"
|
|
]
|
|
}
|
|
)
|
|
|
|
self.assertSetEqual(manual_profile_flows, profile_flows)
|
|
self.assertTrue(
|
|
profile_flows.isdisjoint(
|
|
_registered_regression_flows(run_maestro_ci.GROUPS)
|
|
)
|
|
)
|
|
|
|
def test_destructive_manual_targets_are_not_automatic_pr_targets(self) -> None:
|
|
automatic_pr_targets = _dispatched_maestro_ci_targets(E2E_WORKFLOW)
|
|
|
|
self.assertTrue(
|
|
set(run_maestro_ci.DESTRUCTIVE_MANUAL_TARGETS).isdisjoint(
|
|
automatic_pr_targets
|
|
)
|
|
)
|
|
|
|
def test_destructive_manual_recipes_have_distinct_diagnostics(self) -> None:
|
|
for target, recipes in run_maestro_ci.DESTRUCTIVE_MANUAL_TARGETS.items():
|
|
with self.subTest(target=target):
|
|
configs = [
|
|
run_maestro.parse_config(arguments, {}) for arguments in recipes
|
|
]
|
|
self.assertEqual(
|
|
len({config.jellyfin_log for config in configs}), len(configs)
|
|
)
|
|
self.assertEqual(
|
|
len({config.diagnostics_dir for config in configs}), len(configs)
|
|
)
|
|
|
|
def test_destructive_manual_target_requires_disposable_emulator_opt_in(
|
|
self,
|
|
) -> None:
|
|
with (
|
|
patch.object(run_maestro_ci.run_maestro, "main") as run,
|
|
redirect_stderr(io.StringIO()) as error_output,
|
|
):
|
|
exit_status = run_maestro_ci.main(["profile-regressions"])
|
|
|
|
self.assertEqual(exit_status, 2)
|
|
run.assert_not_called()
|
|
self.assertIn("Refusing destructive manual target", error_output.getvalue())
|
|
self.assertIn("--disposable-emulator", error_output.getvalue())
|
|
|
|
def test_destructive_manual_target_runs_after_explicit_opt_in(self) -> None:
|
|
recipes = run_maestro_ci.DESTRUCTIVE_MANUAL_TARGETS["profile-regressions"]
|
|
with (
|
|
patch.object(run_maestro_ci.run_maestro, "main", return_value=0) as run,
|
|
redirect_stdout(io.StringIO()),
|
|
):
|
|
exit_status = run_maestro_ci.main(
|
|
["profile-regressions", "--disposable-emulator"]
|
|
)
|
|
|
|
self.assertEqual(exit_status, 0)
|
|
self.assertEqual(
|
|
[invocation.args[0] for invocation in run.call_args_list], list(recipes)
|
|
)
|
|
|
|
def test_group_stops_after_interruption(self) -> None:
|
|
with (
|
|
patch.object(run_maestro_ci.run_maestro, "main", return_value=143) as run,
|
|
redirect_stdout(io.StringIO()),
|
|
):
|
|
exit_status = run_maestro_ci.run_group("android-15")
|
|
|
|
self.assertEqual(exit_status, 143)
|
|
run.assert_called_once_with(("basic",))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|