"""Validate Jinja2 expressions in Ansible files by rendering them. Extracts ``{{ ... }}`` expressions from Ansible YAML files and renders each one with Ansible's Jinja2 environment using mock variables. Catches errors like reversed filter arguments, undefined filters, and syntax errors before pushing to CI. The check is intentionally lightweight — it doesn't need real Ansible facts or variables. It provides common mock values (now(), ansible_*, etc.) and renders each expression in isolation. Expressions that fail with undefined variables that aren't in the mock set are skipped (not all variables can be predicted). Usage:: python -m devx.tools.check_jinja_expr python -m devx.tools.check_jinja_expr --path ansible/playbooks/deploy-observability.yml Exit code 0 if all renderable expressions pass, 1 if any fail. """ from __future__ import annotations import re import sys from pathlib import Path import click from jinja2 import Environment from jinja2.exceptions import TemplateSyntaxError, UndefinedError REPO_ROOT = Path.cwd() def _default_ansible_dirs() -> list[Path]: """Return the default directories to scan for Ansible files.""" return [ REPO_ROOT / "ansible" / "playbooks", REPO_ROOT / "ansible" / "roles", ] # Mock context for rendering Jinja expressions. MOCK_CONTEXT: dict[str, object] = { "now": lambda fmt=None: ( "2026-01-01T00:00:00+00:00" if fmt else type( "Now", (), { "timestamp": lambda self: 1735689600.0, "strftime": lambda self, fmt: "2026-01-01T00:00:00+00:00", }, )() ), "ansible_date_time": { "iso8601": "2026-01-01T00:00:00+00:00", "epoch": "1735689600", }, "ansible_facts": { "service_mgr": "systemd", "architecture": "x86_64", "distribution_release": "noble", "virtualization_type": "none", "interfaces": ["eth0", "lo"], "hostname": "test-host", }, "ansible_host": "10.0.0.1", "env": "staging", "environment": "staging", "customer_id": "test", "zitadel_domain": "zitadel.test", "_env_name": "staging", "_observability_data_root": "/opt", "skip_zitadel_stack": False, "skip_htpasswd": False, "skip_observability_stack": False, "backup_enabled": True, "app_filter": "", "app_domain": "test.example.com", "oidc_client_id": "test-client-id", "oidc_client_secret": "test-secret", # nosec B105 — mock value for Jinja rendering, not a real secret "s3_backup_bucket": "test-bucket", "s3_endpoint": "https://s3.test", "s3_access_key": "test-key", "s3_secret_key": "test-secret", # nosec B105 — mock value for Jinja rendering, not a real secret } # Pattern to find {{ ... }} expressions (non-greedy, single-line). EXPR_PATTERN = re.compile(r"\{\{(.*?)\}\}", re.DOTALL) def _find_yaml_files(path: Path) -> list[Path]: """Find Ansible YAML files (tasks, playbooks, handlers) in a path.""" if path.is_file(): return [path] files: list[Path] = [] for pattern in ["**/*.yml", "**/*.yaml"]: files.extend(path.glob(pattern)) # Exclude molecule scenarios — they have their own variables. return [f for f in files if "molecule" not in f.parts] def _extract_expressions(content: str) -> list[str]: """Extract Jinja expressions from file content. Filters out Go template syntax (``{{.Field}}``) used in docker inspect --format strings, and single-character fragments from quoted strings that aren't real Jinja expressions. """ expressions = [] for match in EXPR_PATTERN.finditer(content): raw = match.group(1) # Skip multi-line expressions (often have YAML formatting artifacts). if "\n" in raw: continue expr = raw.strip() # Skip empty, control flow, and single-char fragments. if not expr or expr.startswith("%") or len(expr) <= 1: continue # Skip Go template syntax (docker inspect --format). if expr.startswith(".") or "println" in expr: continue # Skip expressions containing Go template dot-access patterns. if ".State." in expr or ".NetworkSettings." in expr: continue # Skip expressions with unbalanced parens/brackets/braces — # the regex captured only part of a larger expression where # }} appears inside a dict literal (e.g. default({'k': {}})). if expr.count("(") != expr.count(")"): continue if expr.count("{") != expr.count("}"): continue if expr.count("[") != expr.count("]"): continue expressions.append(expr) return expressions def _render_expression(expr: str) -> tuple[bool, str]: """Try to render a Jinja expression. Returns (success, error_msg).""" try: env = Environment(autoescape=False, keep_trailing_newline=True) # nosec B701 — Ansible Jinja, not web-facing # noqa: S701 # Add common Ansible filters so expressions can render. # strftime: Ansible's signature is strftime(string_format, second, utc) # where string_format is the piped value. If the piped value looks like # a number (epoch) and second looks like a format string, the args are # reversed — this is the exact bug from OBL-INFRA-508. def _strftime(string_format: str, second: float | None = None, utc: bool = False) -> str: if isinstance(string_format, (int, float)) and isinstance(second, str) and "%" in second: raise ValueError( # noqa: TRY301 "Invalid value for epoch value — strftime filter arguments " "are reversed. The format string must be the piped value: " "'%format%' | strftime(epoch), not epoch | strftime('%format%')" ) return str(string_format) env.filters["strftime"] = _strftime env.filters["b64decode"] = lambda x: x env.filters["b64encode"] = lambda x: x env.filters["regex_replace"] = lambda x, pattern, replacement="": x env.filters["int"] = lambda x, default=0: ( int(x) if isinstance(x, (int, float, str)) and str(x).lstrip("-").isdigit() else default ) env.filters["bool"] = bool env.filters["basename"] = lambda x: str(x).rsplit("/", 1)[-1] env.filters["dirname"] = lambda x: str(x).rsplit("/", 1)[0] if "/" in str(x) else "." env.filters["combine"] = lambda *args, **kwargs: args[0] env.filters["from_json"] = lambda x: x env.filters["to_json"] = lambda x: x env.filters["ternary"] = lambda x, true_val, false_val=None: true_val if x else false_val env.filters["dict2items"] = lambda x: [ {"key": k, "value": v} for k, v in (x.items() if isinstance(x, dict) else []) ] env.filters["map"] = lambda x, attribute=None: x env.filters["default"] = lambda x, default_value="", boolean=False: x if x else default_value env.filters["from_yaml"] = lambda x: x env.filters["difference"] = lambda x, y: x env.filters["join"] = lambda x, sep="": sep.join(str(i) for i in (x if isinstance(x, list) else [x])) env.filters["list"] = lambda x: list(x) if isinstance(x, (list, tuple)) else [x] env.filters["length"] = lambda x: len(x) if hasattr(x, "__len__") else 0 env.filters["items"] = lambda x: list(x.items()) if isinstance(x, dict) else [] env.filters["first"] = lambda x: x[0] if isinstance(x, (list, str)) and x else x env.filters["last"] = lambda x: x[-1] if isinstance(x, (list, str)) and x else x env.filters["upper"] = lambda x: str(x).upper() env.filters["lower"] = lambda x: str(x).lower() env.filters["replace"] = lambda x, old, new: str(x).replace(old, new) env.filters["split"] = lambda x, sep=None: str(x).split(sep) if sep else str(x).split() env.filters["trim"] = lambda x: str(x).strip() env.filters["sort"] = lambda x: sorted(x) if isinstance(x, list) else x env.filters["unique"] = lambda x: list(set(x)) if isinstance(x, list) else x env.filters["count"] = lambda x: len(x) if hasattr(x, "__len__") else 0 env.filters["float"] = lambda x, default=0.0: ( float(x) if isinstance(x, (int, float, str)) and str(x).replace(".", "").lstrip("-").isdigit() else default ) env.filters["string"] = str env.filters["indent"] = lambda x, width=4: str(x) env.filters["to_nice_json"] = str env.filters["to_nice_yaml"] = str env.filters["from_yaml_all"] = lambda x: x env.filters["groupby"] = lambda x: x env.filters["dictsort"] = lambda x: list(x.items()) if isinstance(x, dict) else [] env.filters["max"] = lambda x: max(x) if isinstance(x, list) and x else x env.filters["min"] = lambda x: min(x) if isinstance(x, list) and x else x env.filters["reverse"] = lambda x: list(reversed(x)) if isinstance(x, list) else x env.filters["flatten"] = lambda x: x env.filters["product"] = lambda x: x env.filters["zip"] = lambda x: x env.filters["subelements"] = lambda x: x env.filters["json_query"] = lambda x: x env.filters["type_debug"] = lambda x: type(x).__name__ env.globals["lookup"] = lambda *args, **kwargs: "" env.globals["query"] = lambda *args, **kwargs: [] template = env.from_string("{{ " + expr + " }}") result = template.render(**MOCK_CONTEXT) except TemplateSyntaxError as e: return False, f"Syntax error: {e.message}" except UndefinedError as e: # Undefined variable — skip, we can't mock everything. return True, f"Skipped (undefined: {e})" except Exception as e: # Check if it's a filter argument error. error_msg = str(e) if "Invalid value for epoch" in error_msg: return False, f"strftime filter argument error: {error_msg}" # Other errors might be due to missing mock variables — skip. return True, f"Skipped ({type(e).__name__}: {error_msg})" else: return True, result def _check_file(filepath: Path, repo_root: Path) -> list[str]: """Check all Jinja expressions in a file. Returns list of violations.""" violations = [] content = filepath.read_text() expressions = _extract_expressions(content) for expr in expressions: success, msg = _render_expression(expr) if not success: try: rel_path = filepath.relative_to(repo_root) except ValueError: rel_path = filepath violations.append(f"{rel_path}: `{{{{ {expr} }}}}` — {msg}") return violations @click.command() @click.option( "--path", type=click.Path(exists=True, path_type=Path), help="Check a specific file or directory (default: ansible/playbooks + ansible/roles).", ) @click.option( "--ansible-dir", "ansible_dirs", type=click.Path(exists=True, path_type=Path), multiple=True, default=None, help="Override the default ansible directories (can be repeated). Defaults to ansible/playbooks and ansible/roles.", ) def main(path: Path | None, ansible_dirs: tuple[Path, ...]) -> None: """Validate Jinja2 expressions in Ansible files.""" dirs = list(ansible_dirs) if ansible_dirs else _default_ansible_dirs() if path: files = _find_yaml_files(path) else: files: list[Path] = [] for d in dirs: files.extend(_find_yaml_files(d)) all_violations: list[str] = [] for f in files: all_violations.extend(_check_file(f, REPO_ROOT)) if all_violations: click.echo("[check-jinja-expr] FAIL: invalid Jinja expressions found:") for v in all_violations: click.echo(f" - {v}") click.echo("\nFix: test expressions with `ansible localhost -m debug -a 'msg={{ }}'`") sys.exit(1) else: click.echo("[check-jinja-expr] OK: all Jinja expressions render correctly.") if __name__ == "__main__": # pragma: no cover main()