Files
Unraid-AppOps-Gateway-Public/backend/tests/test_managed_validation_parallel.py

135 lines
5.8 KiB
Python

import re
import importlib.util
import os
import subprocess
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[2]
WORKFLOW = ROOT / ".gitea" / "workflows" / "managed-validation.yml"
def test_managed_validation_keeps_parallel_lanes_and_full_gate():
text = WORKFLOW.read_text(encoding="utf-8")
for job in ("preflight:", "backend:", "frontend:", "security:", "polyglot:", "full:"):
assert f" {job}" in text
assert "needs: [preflight, backend, frontend, security, polyglot]" in text
assert "All managed validation lanes passed." in text
def test_preflight_preserves_docs_only_fast_path():
text = WORKFLOW.read_text(encoding="utf-8")
assert "runtime_change" in text
assert "*.md|*.mdx|docs/*" in text
def test_backend_budget_is_bounded_without_weakening_validation():
text = WORKFLOW.read_text(encoding="utf-8")
expected_timeouts = {
"preflight": 10,
"backend": 60,
"frontend": 30,
"security": 20,
"polyglot": 30,
"full": 10,
}
for job, expected_minutes in expected_timeouts.items():
match = re.search(
rf"(?ms)^ {re.escape(job)}:\n(?P<body>.*?)(?=^ [a-z][a-z0-9_-]*:\n|\Z)",
text,
)
assert match is not None, f"missing managed-validation job {job}"
timeout = re.search(r"(?m)^ timeout-minutes: (\d+)\s*$", match["body"])
assert timeout is not None, f"missing timeout for {job}"
assert int(timeout.group(1)) == expected_minutes
for lane in ("preflight", "backend", "frontend", "security", "polyglot", "full"):
assert f" {lane}:" in text
assert "needs: [preflight, backend, frontend, security, polyglot]" in text
assert "All managed validation lanes passed." in text
backend = re.search(
r"(?ms)^ backend:\n(?P<body>.*?)(?=^ [a-z][a-z0-9_-]*:\n|\Z)",
text,
)
assert backend is not None
backend_body = backend["body"]
assert 'if [[ "${profile}" == test || "${profile}" == full ]]; then' in backend_body
pytest_commands = [
line.strip() for line in backend_body.splitlines() if "-m pytest " in line
]
assert pytest_commands == [
'"${py}" -m pytest backend/tests -q --durations=20 --maxfail=1'
]
assert "continue-on-error" not in backend_body
assert "--ignore" not in backend_body
assert "--deselect" not in backend_body
def _job_body(job):
text = WORKFLOW.read_text(encoding="utf-8")
match = re.search(rf"(?ms)^ {re.escape(job)}:\n(?P<body>.*?)(?=^ [a-z][a-z0-9_-]*:\n|\Z)", text)
assert match is not None
return match["body"]
def test_light_checks_use_existing_single_checks_profile():
spec = importlib.util.spec_from_file_location(
"appops_light_checks_profiles", ROOT / "ops/gitea-deploy-broker/runner_profiles.py"
)
profiles = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = profiles
spec.loader.exec_module(profiles)
try:
for job in ("preflight", "full"):
label = re.search(r"(?m)^ runs-on: (\S+)\s*$", _job_body(job)).group(1)
assert tuple(profile.id for profile in profiles.profiles_for_labels((label,))) == ("checks",)
assert profiles.RUNNER_PROFILES["checks"].memory_mib == 1536
for job in ("backend", "frontend", "security", "polyglot"):
assert " runs-on: linux-validation\n" in _job_body(job)
finally:
sys.modules.pop(spec.name, None)
def _bash_script(job):
body = _job_body(job)
script = body.split(" run: |\n", 1)[1]
return "\n".join(line[10:] if line.startswith(" ") else line for line in script.splitlines())
@pytest.mark.skipif(os.name == "nt", reason="Execute actual workflow Bash on Linux")
@pytest.mark.parametrize("failed_lane", [None, "PREFLIGHT", "BACKEND", "FRONTEND", "SECURITY", "POLYGLOT"])
def test_light_full_gate_executes_all_required_lane_results(tmp_path, failed_lane):
environment = {"PATH": os.defpath, **{name: "success" for name in ("PREFLIGHT", "BACKEND", "FRONTEND", "SECURITY", "POLYGLOT")}}
if failed_lane:
environment[failed_lane] = "failure"
result = subprocess.run(["bash", "-c", _bash_script("full")], cwd=tmp_path, env=environment,
capture_output=True, text=True, timeout=10)
assert (result.returncode == 0) is (failed_lane is None)
assert ("All managed validation lanes passed." in result.stdout) is (failed_lane is None)
@pytest.mark.skipif(os.name == "nt", reason="Execute actual workflow Bash/Git on Linux")
@pytest.mark.parametrize("change", ["clean", "whitespace", "conflict"])
def test_light_preflight_executes_git_and_rejects_bad_source(tmp_path, change):
environment = {"PATH": os.defpath, "HOME": str(tmp_path), "GITHUB_OUTPUT": str(tmp_path / "output")}
for args in (["init", "--quiet"], ["config", "user.name", "CI fixture"],
["config", "user.email", "ci@example.invalid"]):
subprocess.run(["git", *args], cwd=tmp_path, env=environment, check=True, capture_output=True)
readme = tmp_path / "README.md"
readme.write_text("Clean source\n")
subprocess.run(["git", "add", "README.md"], cwd=tmp_path, env=environment, check=True, capture_output=True)
subprocess.run(["git", "commit", "--quiet", "-m", "fixture"], cwd=tmp_path, env=environment, check=True, capture_output=True)
if change == "whitespace":
readme.write_text("Trailing whitespace \n")
elif change == "conflict":
readme.write_text("<<<<<<< fixture\n")
result = subprocess.run(["bash", "-c", _bash_script("preflight")], cwd=tmp_path, env=environment,
capture_output=True, text=True, timeout=10)
assert (result.returncode == 0) is (change == "clean")
if change == "clean":
assert "runtime_change=true" in (tmp_path / "output").read_text()