mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-06 02:54:26 +00:00
kevinveenbirkenbachandClaude Opus 5
5137465c6f
3529 invalid comments in 528 files, grandfathered in while the linter only saw the working-tree diff. Removal ran through the linter's own collectors, so detection and deletion cannot drift: a comment whose opener line is blank before it and whose closer line is blank after it loses all its lines, anything else loses only the comment region, and a line left holding a bare delimiter goes with it. Five files keep their comments behind a nocheck marker instead. Four are upstream configs whose inline docs are upstream's own (mautrix, systemd journald, nextcloud-oidc-login, MediaWiki). In littlejs the commented-out entries ARE the deactivation mechanism for an example tile, so deleting them would delete configuration. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
315 lines
11 KiB
Python
315 lines
11 KiB
Python
import unittest
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
try:
|
|
import yaml # noqa: F401 # imported only to assert pyyaml is installed
|
|
except ImportError: # pragma: no cover
|
|
raise SystemExit("Please `pip install pyyaml` to run this test.") from None
|
|
|
|
from utils.cache import PROJECT_ROOT
|
|
from utils.cache.files import iter_project_files
|
|
from utils.cache.yaml import load_yaml_all
|
|
from utils.roles.mapping import ROLE_FILE_TASKS_MAIN
|
|
|
|
# ---------- Helpers: YAML parsing ----------
|
|
|
|
|
|
def _load_yaml_file(path: str) -> list[dict[str, Any]]:
|
|
"""
|
|
Load a tasks YAML file.
|
|
Returns a list of top-level task dicts. If the file is empty, returns [].
|
|
Supports multi-doc YAML.
|
|
"""
|
|
docs = list(load_yaml_all(path)) or []
|
|
tasks: list[dict[str, Any]] = []
|
|
for doc in docs:
|
|
if doc is None:
|
|
continue
|
|
if isinstance(doc, list):
|
|
tasks.extend([t for t in doc if isinstance(t, dict)])
|
|
elif isinstance(doc, dict):
|
|
if "tasks" in doc and isinstance(doc["tasks"], list):
|
|
tasks.extend([t for t in doc["tasks"] if isinstance(t, dict)])
|
|
else:
|
|
tasks.append(doc)
|
|
return tasks
|
|
|
|
|
|
# ---------- Helpers: when / structure checks ----------
|
|
|
|
|
|
def _normalize_when(value: Any) -> list[str]:
|
|
"""
|
|
Normalize a 'when' value (string | list | bool | None) to a list of strings.
|
|
Non-string entries are ignored.
|
|
"""
|
|
if value is None or isinstance(value, bool):
|
|
return []
|
|
if isinstance(value, str):
|
|
v = value.strip()
|
|
return [v] if v else []
|
|
if isinstance(value, list):
|
|
out: list[str] = []
|
|
for item in value:
|
|
if isinstance(item, str):
|
|
s = item.strip()
|
|
if s:
|
|
out.append(s)
|
|
return out
|
|
return []
|
|
|
|
|
|
def _task_has_block_with_when(task: dict[str, Any]) -> bool:
|
|
return "block" in task and bool(_normalize_when(task.get("when")))
|
|
|
|
|
|
def _is_pure_guarded_tasks_file(tasks: list[dict[str, Any]]) -> tuple[list[str], bool]:
|
|
"""
|
|
A "pure guarded" tasks file has EXACTLY ONE top-level task,
|
|
that task contains a 'block', and that task has a 'when' condition.
|
|
Returns (guard_conditions, is_pure_guarded).
|
|
"""
|
|
if len(tasks) != 1:
|
|
return [], False
|
|
only_task = tasks[0]
|
|
if not _task_has_block_with_when(only_task):
|
|
return [], False
|
|
return _normalize_when(only_task.get("when")), True
|
|
|
|
|
|
# ---------- Helpers: discovery ----------
|
|
|
|
|
|
def _iter_all_tasks_files(repo_root: str) -> list[str]:
|
|
"""
|
|
Return all tasks/*.yml|*.yaml files in the project tree.
|
|
Filenames must sit directly inside a 'tasks' directory.
|
|
"""
|
|
files: list[str] = []
|
|
seen: set[str] = set()
|
|
for p in iter_project_files(extensions=(".yml", ".yaml")):
|
|
if Path(p).parent.name != "tasks":
|
|
continue
|
|
if p in seen:
|
|
continue
|
|
seen.add(p)
|
|
files.append(p)
|
|
return files
|
|
|
|
|
|
def _get_include_role_name(task: dict[str, Any]) -> str | None:
|
|
"""
|
|
If task is an include_role task, return the role 'name'.
|
|
Supports 'include_role' and 'ansible.builtin.include_role'.
|
|
"""
|
|
for key in ("include_role", "ansible.builtin.include_role"):
|
|
if key in task and isinstance(task[key], dict):
|
|
role_name = task[key].get("name")
|
|
if isinstance(role_name, str) and role_name.strip():
|
|
return role_name.strip()
|
|
return None
|
|
|
|
|
|
def _get_include_tasks_target(task: dict[str, Any]) -> str | None:
|
|
"""
|
|
If task is an include_tasks, return the path string as-is (could be relative).
|
|
Supports 'include_tasks' and 'ansible.builtin.include_tasks'.
|
|
Returns None if not found or not a string.
|
|
"""
|
|
for key in ("include_tasks", "ansible.builtin.include_tasks"):
|
|
if key in task:
|
|
val = task[key]
|
|
if isinstance(val, str):
|
|
return val.strip()
|
|
return None
|
|
|
|
|
|
def _contains_jinja(s: str) -> bool:
|
|
return "{{" in s or "{%" in s or "}}" in s or "%}" in s
|
|
|
|
|
|
def _resolve_include_tasks_path(include_value: str, including_file: str) -> str | None:
|
|
"""
|
|
Resolve an include_tasks path relative to the including file.
|
|
If it contains Jinja or does not resolve to an existing file, return None.
|
|
Tries exact path, then adds .yml / .yaml if extension missing.
|
|
"""
|
|
if _contains_jinja(include_value):
|
|
return None
|
|
|
|
candidates: list[str] = []
|
|
if Path(include_value).is_absolute():
|
|
candidates.append(include_value)
|
|
else:
|
|
base = str(Path(including_file).parent)
|
|
candidates.append(str(Path(base) / include_value))
|
|
|
|
final_candidates: list[str] = []
|
|
for c in candidates:
|
|
final_candidates.append(c)
|
|
candidate_path = Path(c)
|
|
if not candidate_path.suffix:
|
|
final_candidates.append(c + ".yml")
|
|
final_candidates.append(c + ".yaml")
|
|
|
|
for c in final_candidates:
|
|
if Path(c).is_file():
|
|
return c
|
|
|
|
return None
|
|
|
|
|
|
class PureGuardedIncludeTest(unittest.TestCase):
|
|
"""
|
|
Enforce short-circuit includes ONLY for "pure guarded" targets:
|
|
- Exactly one top-level task
|
|
- That task is a 'block'
|
|
- That task has a 'when'
|
|
Apply to both:
|
|
- include_role (roles/<role>/tasks/main.yml must be pure guarded)
|
|
- include_tasks (target tasks file must be pure guarded)
|
|
"""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.repo_root = str(PROJECT_ROOT)
|
|
|
|
cls.pure_guarded_roles: dict[str, tuple[list[str], str]] = {}
|
|
|
|
roles_dir = Path(cls.repo_root) / "roles"
|
|
for role_dir in roles_dir.iterdir():
|
|
if not role_dir.is_dir():
|
|
continue
|
|
main_path = role_dir / ROLE_FILE_TASKS_MAIN
|
|
if not main_path.is_file():
|
|
continue
|
|
main_path = str(main_path)
|
|
role_name = role_dir.name
|
|
try:
|
|
tasks = _load_yaml_file(main_path)
|
|
guards, pure = _is_pure_guarded_tasks_file(tasks)
|
|
if pure and guards:
|
|
cls.pure_guarded_roles[role_name] = (guards, main_path)
|
|
except Exception:
|
|
pass
|
|
|
|
cls.tasks_file_cache: dict[str, tuple[list[str], bool]] = {}
|
|
|
|
cls.all_tasks_files = _iter_all_tasks_files(cls.repo_root)
|
|
|
|
# ---------- Tests ----------
|
|
|
|
def test_include_role_short_circuits_when_target_is_pure_guarded(self):
|
|
failures: list[str] = []
|
|
|
|
if not self.pure_guarded_roles:
|
|
self.skipTest(
|
|
"No pure guarded roles found; nothing to validate for include_role."
|
|
)
|
|
|
|
for path in self.all_tasks_files:
|
|
try:
|
|
tasks = _load_yaml_file(path)
|
|
except Exception as e:
|
|
failures.append(f"[PARSE ERROR] {path}: {e}")
|
|
continue
|
|
|
|
for idx, task in enumerate(tasks):
|
|
role_name = _get_include_role_name(task)
|
|
if not role_name:
|
|
continue
|
|
|
|
role_entry = self.pure_guarded_roles.get(role_name)
|
|
if not role_entry:
|
|
continue
|
|
|
|
guards, main_path = role_entry
|
|
include_when = _normalize_when(task.get("when"))
|
|
|
|
if not include_when:
|
|
failures.append(
|
|
f"{path} (task #{idx + 1}) includes role '{role_name}' "
|
|
f"but lacks a 'when'. The role is pure guarded by {guards} in {main_path}. "
|
|
f"Add at least one of those guard expressions to the include to avoid loading the role unnecessarily."
|
|
)
|
|
continue
|
|
|
|
if not any(req in include_when for req in guards):
|
|
failures.append(
|
|
f"{path} (task #{idx + 1}) includes role '{role_name}' but its 'when' "
|
|
f"does not contain the role's guard.\n"
|
|
f"Role guard(s) from {main_path}: {guards}\n"
|
|
f"Include 'when': {include_when}\n"
|
|
"Add the role's guard condition to short-circuit when false."
|
|
)
|
|
|
|
if failures:
|
|
self.fail(
|
|
"Some include_role calls are missing pure-guard short-circuiting:\n\n"
|
|
+ "\n\n".join(failures)
|
|
)
|
|
|
|
def test_include_tasks_short_circuits_when_target_is_pure_guarded(self):
|
|
failures: list[str] = []
|
|
|
|
for including_path in self.all_tasks_files:
|
|
try:
|
|
including_tasks = _load_yaml_file(including_path)
|
|
except Exception as e:
|
|
failures.append(f"[PARSE ERROR] {including_path}: {e}")
|
|
continue
|
|
|
|
for idx, task in enumerate(including_tasks):
|
|
include_value = _get_include_tasks_target(task)
|
|
if not include_value:
|
|
continue
|
|
|
|
resolved = _resolve_include_tasks_path(include_value, including_path)
|
|
if not resolved:
|
|
continue
|
|
|
|
if resolved not in self.tasks_file_cache:
|
|
try:
|
|
target_tasks = _load_yaml_file(resolved)
|
|
guards, pure = _is_pure_guarded_tasks_file(target_tasks)
|
|
self.tasks_file_cache[resolved] = (guards, pure)
|
|
except Exception as e:
|
|
failures.append(
|
|
f"[PARSE ERROR] included by {including_path} (task #{idx + 1}): {resolved}: {e}"
|
|
)
|
|
self.tasks_file_cache[resolved] = ([], False)
|
|
|
|
guards, pure = self.tasks_file_cache.get(resolved, ([], False))
|
|
if not (pure and guards):
|
|
continue
|
|
|
|
include_when = _normalize_when(task.get("when"))
|
|
|
|
if not include_when:
|
|
failures.append(
|
|
f"{including_path} (task #{idx + 1}) includes tasks '{include_value}' "
|
|
f"-> {resolved}, which is PURE GUARDED by {guards}. "
|
|
f"Add at least one of those guard expressions to the include to avoid loading the file unnecessarily."
|
|
)
|
|
continue
|
|
|
|
if not any(req in include_when for req in guards):
|
|
failures.append(
|
|
f"{including_path} (task #{idx + 1}) includes tasks '{include_value}' "
|
|
f"-> {resolved}, but its 'when' does not contain the target's guard.\n"
|
|
f"Target guard(s): {guards}\n"
|
|
f"Include 'when': {include_when}\n"
|
|
"Add the guard condition to short-circuit when false."
|
|
)
|
|
|
|
if failures:
|
|
self.fail(
|
|
"Some include_tasks calls are missing pure-guard short-circuiting:\n\n"
|
|
+ "\n\n".join(failures)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|