fix(ci): make cross-platform checks deterministic
This commit is contained in:
edde746
+208
-5
@@ -2,10 +2,12 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
from contextlib import redirect_stderr, redirect_stdout
|
||||
from dataclasses import replace
|
||||
import io
|
||||
from pathlib import Path
|
||||
import shlex
|
||||
import subprocess
|
||||
import sys
|
||||
import unittest
|
||||
@@ -17,6 +19,112 @@ 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"
|
||||
E2E_WORKFLOW = ROOT_DIR / ".github/workflows/e2e.yml"
|
||||
SCRIPT_TEST_DISPATCHERS = (
|
||||
LOCAL_SCRIPT_TEST_DISPATCHER,
|
||||
CI_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():
|
||||
try:
|
||||
command = shlex.split(line.strip(), 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:
|
||||
for dispatcher in (LOCAL_SCRIPT_TEST_DISPATCHER, CI_SCRIPT_TEST_DISPATCHER):
|
||||
with self.subTest(dispatcher=dispatcher):
|
||||
self.assertEqual(
|
||||
_dispatched_script_tests(dispatcher).count(
|
||||
"test_maestro_real_jellyfin.py"
|
||||
),
|
||||
1,
|
||||
)
|
||||
|
||||
|
||||
class ParseConfigTests(unittest.TestCase):
|
||||
def test_basic_defaults(self) -> None:
|
||||
config = run_maestro.parse_config([], {})
|
||||
@@ -198,11 +306,106 @@ class CiGroupTests(unittest.TestCase):
|
||||
self.assertEqual(exit_status, 0)
|
||||
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_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 (
|
||||
|
||||
Reference in New Issue
Block a user