Enforce Ansible availability via Nix and validate requirements.yml

- Add ansiblePkg as propagated dependency in flake.nix so ansible-galaxy is available on host
- Introduce strict requirements.yml validator for AnsibleRequirementsInstaller
- Accept roles entries with either 'name' or 'src'
- Ensure run() always validates requirements before installing dependencies
- Extend unit tests to cover valid, invalid and warning-only requirements.yml cases

See: https://chatgpt.com/share/69332bc4-a128-800f-a69c-fdc24c4cc7fe
This commit is contained in:
Kevin Veen-Birkenbach
2025-12-05 20:00:47 +01:00
parent f5475d86e2
commit a435745c02
3 changed files with 191 additions and 21 deletions

View File

@@ -9,7 +9,7 @@ This installer installs collections and roles via ansible-galaxy when found.
import os
import tempfile
from typing import Any, Dict
from typing import Any, Dict, List
import yaml
@@ -35,12 +35,79 @@ class AnsibleRequirementsInstaller(BaseInstaller):
print(f"Error loading {self.REQUIREMENTS_FILE} in {identifier}: {exc}")
return {}
def _validate_requirements(self, requirements: Dict[str, Any], identifier: str) -> None:
"""
Validate the requirements.yml structure.
Raises SystemExit on any validation error.
"""
errors: List[str] = []
if not isinstance(requirements, dict):
errors.append("Top-level structure must be a mapping.")
else:
allowed_keys = {"collections", "roles"}
unknown_keys = set(requirements.keys()) - allowed_keys
if unknown_keys:
print(
f"Warning: requirements.yml in {identifier} contains unknown keys: "
f"{', '.join(sorted(unknown_keys))}"
)
for section in ("collections", "roles"):
if section not in requirements:
continue
value = requirements[section]
if not isinstance(value, list):
errors.append(f"'{section}' must be a list.")
continue
for idx, entry in enumerate(value):
if isinstance(entry, str):
# short form "community.docker" etc.
continue
if isinstance(entry, dict):
# Collections: brauchen zwingend 'name'
if section == "collections":
if not entry.get("name"):
errors.append(
f"Entry #{idx} in '{section}' is a mapping "
f"but has no 'name' key."
)
else:
# Roles: 'name' ODER 'src' sind ok (beides gängig)
if not (entry.get("name") or entry.get("src")):
errors.append(
f"Entry #{idx} in '{section}' is a mapping but "
f"has neither 'name' nor 'src' key."
)
continue
errors.append(
f"Entry #{idx} in '{section}' has invalid type "
f"{type(entry).__name__}; expected string or mapping."
)
if errors:
print(f"Invalid requirements.yml in {identifier}:")
for err in errors:
print(f" - {err}")
raise SystemExit(
f"requirements.yml validation failed for {identifier}."
)
def run(self, ctx: RepoContext) -> None:
req_file = os.path.join(ctx.repo_dir, self.REQUIREMENTS_FILE)
requirements = self._load_requirements(req_file, ctx.identifier)
if not requirements or not isinstance(requirements, dict):
if not requirements:
return
# Validate structure before doing anything dangerous
self._validate_requirements(requirements, ctx.identifier)
if "collections" not in requirements and "roles" not in requirements:
return