3"""End-to-end workflow scan fixtures for ``check_ci_parity.py``.
5The production checker owns the policy. This companion owns the temporary
6workflow trees that prove the complete scan fires in both directions.
9from __future__
import annotations
12from pathlib
import Path
14import check_ci_parity
as parity
18def _workflow_yaml(trigger_block: str, *, soft: str =
"", job_if: str =
"") -> str:
19 """Render a minimal one-gate workflow for the scan selftest."""
26 " runs-on: ubuntu-latest\n"
28 " - name: probe gate\n"
30 " run: just quality::local::gate probe-gate\n"
34def _cases() -> list[tuple[str, str, str, bool]]:
35 """Return the automatic/manual reachability fixtures."""
36 push =
"on:\n push:\n branches: [dev]\n"
37 dispatch =
"on:\n workflow_dispatch:\n"
38 no_trigger =
"# on:\n# push:\n"
42 (
"push-triggered gate",
"fast", _workflow_yaml(push),
False),
44 "triggers all commented out (hil-all's shape)",
46 _workflow_yaml(no_trigger),
50 "dispatch-only workflow for a fast gate",
52 _workflow_yaml(dispatch),
56 "dispatch-only workflow for a manual gate",
58 _workflow_yaml(dispatch),
62 "continue-on-error step for a fast gate",
64 _workflow_yaml(push, soft=
" continue-on-error: true\n"),
68 "job disabled by `if: false`",
70 _workflow_yaml(push, job_if=
" if: false\n"),
76def scan_selftest() -> int:
77 """Prove the real workflow scan and reachability checks both ways."""
79 registry_name =
"probe-gate"
80 for label, speed, workflow_text, must_fire
in _cases():
81 with tempfile.TemporaryDirectory()
as tmp:
83 (directory /
"probe.yml").write_text(workflow_text, encoding=
"utf-8")
84 registry = {registry_name: speed}
85 errors, bindings = parity.check_workflows(registry, directory)
86 errors.extend(parity.reachability_errors(registry, bindings))
88 ok = fired == must_fire
89 failures += 0
if ok
else 1
90 expectation =
"must fire" if must_fire
else "must stay quiet"
91 print(f
" [{'ok' if ok else 'FAIL'}] scan: {label} ({expectation})")
95 with tempfile.TemporaryDirectory()
as tmp:
96 errors, _ = parity.check_workflows({registry_name:
"fast"}, Path(tmp))
98 failures += 0
if ok
else 1
99 print(f
" [{'ok' if ok else 'FAIL'}] scan: an empty workflow directory is refused")
103 parsed = yaml.safe_load(
"on:\n push:\n branches: [dev]\njobs: {}\n")
104 ok = parity.workflow_triggers(parsed) == {
"push"}
105 failures += 0
if ok
else 1
106 print(f
" [{'ok' if ok else 'FAIL'}] scan: bare `on:` parses as a trigger map, not a bool key")