Publish curated Unraid-AppOps-Gateway source from a194da2defbd709ea58bd4c9503966dbd37b11f2

backend/tests/test_managed_validation_parallel.py

@@ -1,6 +1,12 @@
 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"
 
@@ -61,3 +67,68 @@ def test_backend_budget_is_bounded_without_weakening_validation():
     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()