From 1fc7a03c1a94190824129a60bbfebdc3c7eca61e Mon Sep 17 00:00:00 2001 From: emil User Date: Wed, 12 Aug 2026 12:43:37 +0000 Subject: [PATCH] DEVX-152: feat: sync missing features from v0.49.x line to master --- AGENTS.md | 5 + README.md | 17 +- docs/index.md | 18 +- src/devx/molecule/molecule_changed.py | 158 ++++++++ src/devx/tools/check_ansible_no_log.py | 232 +++++++++++ .../check_ansible_no_state_absent_on_db.py | 176 +++++++++ src/devx/tools/check_ansible_patterns.py | 345 ++++++++++++++++ src/devx/tools/check_jinja_expr.py | 292 ++++++++++++++ tests/unit/test_molecule_changed.py | 192 +++++++++ tests/unit/test_tools_check_ansible_no_log.py | 367 ++++++++++++++++++ ...ols_check_ansible_no_state_absent_on_db.py | 198 ++++++++++ .../unit/test_tools_check_ansible_patterns.py | 340 ++++++++++++++++ tests/unit/test_tools_check_jinja_expr.py | 264 +++++++++++++ 13 files changed, 2584 insertions(+), 20 deletions(-) create mode 100644 src/devx/molecule/molecule_changed.py create mode 100644 src/devx/tools/check_ansible_no_log.py create mode 100644 src/devx/tools/check_ansible_no_state_absent_on_db.py create mode 100644 src/devx/tools/check_ansible_patterns.py create mode 100644 src/devx/tools/check_jinja_expr.py create mode 100644 tests/unit/test_molecule_changed.py create mode 100644 tests/unit/test_tools_check_ansible_no_log.py create mode 100644 tests/unit/test_tools_check_ansible_no_state_absent_on_db.py create mode 100644 tests/unit/test_tools_check_ansible_patterns.py create mode 100644 tests/unit/test_tools_check_jinja_expr.py diff --git a/AGENTS.md b/AGENTS.md index 4386a2d..7179915 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -124,6 +124,10 @@ src/devx/ │ ├── check_docker_init.py # Check Docker Compose services with healthchecks have init: true │ ├── check_ansible_set_fact_to_json.py # Check set_fact tasks don't misuse to_json │ ├── check_alert_rules.py # Validate Prometheus alert rules with promtool +│ ├── check_ansible_no_log.py # Check Ansible tasks for missing no_log on secrets +│ ├── check_ansible_patterns.py # Detect dangerous failure-masking patterns +│ ├── check_jinja_expr.py # Validate Jinja2 expressions in Ansible files +│ ├── check_ansible_no_state_absent_on_db.py # Prevent state:absent on DB paths │ └── _shared.py # Shared tool utilities ├── opentofu.py # OpenTofu output helpers (get_tofu_output, get_tofu_vm_ip, get_tofu_vm_field) ├── utils/ # Shared utilities (reusable across projects) @@ -143,6 +147,7 @@ src/devx/ ├── distribute_molecule.py # Distribute molecule scenarios across runners (LPT scheduling, --roles-root for multi-role) ├── molecule_ci_guard.py # Run molecule with cross-runner fail-fast (--roles-root) ├── molecule_all.py # Run all molecule scenarios locally + ├── molecule_changed.py # Detect which Ansible roles changed and output molecule scenarios ├── start_docker.py # Ensure Docker daemon is running for molecule tests └── platforms.py # Supported molecule platforms ``` diff --git a/README.md b/README.md index b8a4ad7..0fba4d0 100644 --- a/README.md +++ b/README.md @@ -16,12 +16,12 @@ quality badges. [![CI](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions/workflows/ci.yml/badge.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions) [![License: GPL-3.0](https://img.shields.io/badge/license-GPL--3.0-blue)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/src/branch/master/LICENSE) -[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/e463f9a5d56cfcba17d7daa0dd73a387070ae464/coverage.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions) -[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/e463f9a5d56cfcba17d7daa0dd73a387070ae464/tests.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions) -[![Docs](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/e463f9a5d56cfcba17d7daa0dd73a387070ae464/docs.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki) -[![Code Quality](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/e463f9a5d56cfcba17d7daa0dd73a387070ae464/quality.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions) -[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/e463f9a5d56cfcba17d7daa0dd73a387070ae464/version.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases) -[![Python](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/e463f9a5d56cfcba17d7daa0dd73a387070ae464/python.svg)](https://www.python.org/downloads/) +[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/8c02351c3190af51b1dfe545a7abbec665c51559/coverage.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions) +[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/8c02351c3190af51b1dfe545a7abbec665c51559/tests.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions) +[![Docs](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/8c02351c3190af51b1dfe545a7abbec665c51559/docs.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki) +[![Code Quality](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/8c02351c3190af51b1dfe545a7abbec665c51559/quality.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions) +[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/8c02351c3190af51b1dfe545a7abbec665c51559/version.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases) +[![Python](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/8c02351c3190af51b1dfe545a7abbec665c51559/python.svg)](https://www.python.org/downloads/) ## Why devx? @@ -226,10 +226,6 @@ python -m devx.molecule.distribute_molecule --runner-index 1 --max-runners 3 python -m devx.molecule.distribute_molecule --list # list all scenarios python -m devx.molecule.distribute_molecule --list-platforms # list platforms -# Run molecule tests with cross-runner fail-fast -python -m devx.molecule.molecule_ci_guard pair1 pair2 -python -m devx.molecule.molecule_ci_guard --roles-root ansible/roles pair1 pair2 - # Run all molecule scenarios locally (sequential) python -m devx.molecule.molecule_all python -m devx.molecule.molecule_all --bin .venv/bin @@ -303,7 +299,6 @@ devx --version | `devx molecule all` | Run all molecule scenarios on all supported platforms | | `devx molecule discover-runners` | Discover available Gitea Actions runners | | `devx molecule distribute` | Distribute molecule test pairs across parallel runners | -| `devx molecule guard` | Run molecule tests with CI failure polling | See [CLI Commands](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki/CLI-Commands) in the wiki for full command documentation with examples. diff --git a/docs/index.md b/docs/index.md index d5ab33b..caef7cf 100644 --- a/docs/index.md +++ b/docs/index.md @@ -12,12 +12,12 @@ project to be reusable across all oblachno-oss repositories. [![CI](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions/workflows/ci.yml/badge.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions) [![License: GPL-3.0](https://img.shields.io/badge/license-GPL--3.0-blue)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/src/branch/master/LICENSE) -[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/e463f9a5d56cfcba17d7daa0dd73a387070ae464/coverage.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions) -[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/e463f9a5d56cfcba17d7daa0dd73a387070ae464/tests.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions) -[![Docs](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/e463f9a5d56cfcba17d7daa0dd73a387070ae464/docs.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki) -[![Code Quality](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/e463f9a5d56cfcba17d7daa0dd73a387070ae464/quality.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions) -[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/e463f9a5d56cfcba17d7daa0dd73a387070ae464/version.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases) -[![Python](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/e463f9a5d56cfcba17d7daa0dd73a387070ae464/python.svg)](https://www.python.org/downloads/) +[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/8c02351c3190af51b1dfe545a7abbec665c51559/coverage.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions) +[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/8c02351c3190af51b1dfe545a7abbec665c51559/tests.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions) +[![Docs](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/8c02351c3190af51b1dfe545a7abbec665c51559/docs.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki) +[![Code Quality](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/8c02351c3190af51b1dfe545a7abbec665c51559/quality.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions) +[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/8c02351c3190af51b1dfe545a7abbec665c51559/version.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases) +[![Python](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/8c02351c3190af51b1dfe545a7abbec665c51559/python.svg)](https://www.python.org/downloads/) ## Overview @@ -104,8 +104,8 @@ devx is a self-contained Python package under `src/devx/`: - **Dev tools** (`devx.tools`) — setup, install_tools, check_test_speed, configure_repo, generate_badges, generate_cliff_config, install_checkmake - **Molecule tools** (`devx.molecule`) — Optional, for projects with Ansible - roles: distribute_molecule, molecule_ci_guard, molecule_all, discover_runners, - start_docker, platforms + roles: distribute_molecule, molecule_all, discover_runners, start_docker, + platforms See [Architecture](Architecture) for the full package structure, module descriptions, design principles, and data flow diagrams. @@ -132,7 +132,7 @@ devx provides a `devx` CLI with three command groups: - `devx ci ` — CI/CD automation (17 commands) - `devx tools ` — Developer tools (9 commands) -- `devx molecule ` — Molecule testing (4 commands, optional) +- `devx molecule ` — Molecule testing (3 commands, optional) See [CLI Commands](CLI-Commands) for full command documentation with examples. diff --git a/src/devx/molecule/molecule_changed.py b/src/devx/molecule/molecule_changed.py new file mode 100644 index 0000000..66b4c6d --- /dev/null +++ b/src/devx/molecule/molecule_changed.py @@ -0,0 +1,158 @@ +"""Detect which Ansible roles changed and output their molecule scenarios. + +Usage:: + + python -m devx.molecule.molecule_changed --print-targets + python -m devx.molecule.molecule_changed --base origin/master --print-roles + +Outputs the list of make targets (e.g. molecule-docker-base) for roles +that have changed files vs the base ref. Used by ``make molecule-changed`` +to run only the molecule scenarios affected by the current diff. + +Role-to-target mapping is derived from the directory structure: + ansible/roles// → molecule- + +For roles with multiple scenarios (e.g. app_container has customer-apps, +nextcloud, postgres-upgrade, simple-app), the base target runs all +scenarios for that role. + +Playbooks that change also trigger molecule for the roles they include. +Shared infrastructure changes (ansible.cfg, requirements.yml, molecule/) +trigger all scenarios. +""" + +from __future__ import annotations + +import subprocess # nosec B404 — used to run git, a trusted binary +from pathlib import Path + +import click + +REPO_ROOT = Path.cwd() + +# Map role names to make targets. +ROLE_TARGET_MAP: dict[str, str] = { + "app_container": "molecule-app-container", + "app_hardening": "molecule-app-hardening", + "crowdsec": "molecule-crowdsec", + "disk_cleanup": "molecule-disk-cleanup", + "docker_base": "molecule-docker-base", + "observability": "molecule-observability", + "restore": "molecule-restore", + "sso_config": "molecule-sso-config", + "storage": "molecule-storage", + "zitadel": "molecule-zitadel", +} + +# Playbooks that map to molecule scenarios (via roles they include). +PLAYBOOK_ROLE_MAP: dict[str, list[str]] = { + "ansible/playbooks/deploy-observability.yml": ["observability", "docker_base", "zitadel", "crowdsec"], + "ansible/playbooks/deploy-customer.yml": ["app_container", "docker_base", "app_hardening", "sso_config"], + "ansible/playbooks/configure-oidc.yml": ["sso_config", "app_container"], + "ansible/playbooks/prepare-vms.yml": ["docker_base", "app_hardening", "storage", "disk_cleanup", "crowdsec"], +} + +# Shared infrastructure that affects all molecule tests. +SHARED_PATHS = ( + "ansible/ansible.cfg", + "ansible/requirements.yml", + "ansible/molecule/", +) + +# Minimum path parts for a role file: ansible/roles/ (3 parts). +# Files inside the role have more parts, but we only need the role name. +_MIN_ROLE_PATH_PARTS = 3 + + +def _run_git(args: list[str]) -> str: # pragma: no cover + """Run a git command and return stdout.""" + result = subprocess.run( # nosec + ["git", *args], + cwd=REPO_ROOT, + capture_output=True, + text=True, + check=False, + ) + return result.stdout + + +def get_changed_files(base: str) -> list[str]: + """Get list of changed files vs base ref.""" + for ref in [base, "master"]: + output = _run_git(["diff", "--name-only", f"{ref}...HEAD"]) + if output.strip(): + return sorted(output.strip().splitlines()) + return [] + + +def detect_changed_roles(changed_files: list[str]) -> set[str]: + """Detect which roles have changed files.""" + roles: set[str] = set() + + for filepath in changed_files: + # Check if file is in a role directory + if filepath.startswith("ansible/roles/"): + parts = filepath.split("/") + if len(parts) >= _MIN_ROLE_PATH_PARTS: + roles.add(parts[2]) + + # Check if file is a playbook that maps to roles + if filepath in PLAYBOOK_ROLE_MAP: + roles.update(PLAYBOOK_ROLE_MAP[filepath]) + + # Check shared infrastructure — triggers all roles + for shared in SHARED_PATHS: + if filepath.startswith(shared): + return set(ROLE_TARGET_MAP.keys()) + + return roles + + +def roles_to_targets(roles: set[str]) -> list[str]: + """Convert role names to make targets.""" + targets = [] + for role in sorted(roles): + target = ROLE_TARGET_MAP.get(role) + if target: + targets.append(target) + return targets + + +@click.command() +@click.option( + "--base", + default="origin/master", + help="Base ref to compare against (default: origin/master).", +) +@click.option( + "--print-targets", + is_flag=True, + help="Print make targets (e.g. molecule-docker-base).", +) +@click.option( + "--print-roles", + is_flag=True, + help="Print role names (default if no --print-targets).", +) +def main(base: str, print_targets: bool, print_roles: bool) -> None: + """Detect which Ansible roles changed and output molecule scenarios.""" + changed_files = get_changed_files(base) + if not changed_files: + click.echo("No changed files detected.", err=True) + return + + roles = detect_changed_roles(changed_files) + if not roles: + click.echo("No molecule scenarios affected by changes.", err=True) + return + + if print_targets: + for target in roles_to_targets(roles): + click.echo(target) + else: + for role in sorted(roles): + click.echo(role) + + +if __name__ == "__main__": # pragma: no cover + main() diff --git a/src/devx/tools/check_ansible_no_log.py b/src/devx/tools/check_ansible_no_log.py new file mode 100644 index 0000000..c2af05e --- /dev/null +++ b/src/devx/tools/check_ansible_no_log.py @@ -0,0 +1,232 @@ +"""Check Ansible tasks for missing no_log on secret-handling tasks. + +ansible-lint's built-in ``no-log-password`` rule only fires when a module +parameter is literally named ``*password*`` and there's a loop. It does +NOT catch: + +- Shell/command tasks that interpolate ``{{ _secrets.* }}`` or + ``{{ *password* }}`` variables +- Template/copy tasks that render secret values without ``no_log`` + +This script fills that gap by scanning all Ansible task files for +variables that look like secrets (``_secrets.*``, ``*password*``, +``*secret*``, ``*token*``, ``*api_key*``) and verifying that the task +has ``no_log`` set to a non-False value. + +Usage:: + + python -m devx.tools.check_ansible_no_log + python -m devx.tools.check_ansible_no_log --path ansible/roles/my_role + python -m devx.tools.check_ansible_no_log --ansible-dir ansible/roles + +Exit code 0 if all secret-handling tasks have no_log, 1 otherwise. +""" + +from __future__ import annotations + +import re +import sys +from pathlib import Path + +import click +import yaml + +REPO_ROOT = Path.cwd() +DEFAULT_ANSIBLE_DIR = REPO_ROOT / "ansible" + +# Patterns that indicate a task is handling secrets. +# We only match Jinja-interpolated variables ({{ ... }}) to avoid false +# positives from field names like "password" in module params or task names. +SECRET_PATTERNS = [ + # {{ _secrets.anything }} or {{ _secrets['anything'] }} + re.compile(r"\{\{[^}]*_secrets\.", re.IGNORECASE), + # {{ anything_password }} but NOT the word "password" in a string literal + re.compile(r"\{\{[^}]*password", re.IGNORECASE), + # {{ anything_secret }} + re.compile(r"\{\{[^}]*_secret\b", re.IGNORECASE), + # {{ anything_api_key }} + re.compile(r"\{\{[^}]*api_key", re.IGNORECASE), + # {{ anything_token }} (but not loop tokens like {{ loop_token }}) + re.compile(r"\{\{[^}]*(?:vault_token|auth_token|access_token|bot_token)", re.IGNORECASE), +] + +# Task keys whose values might contain secret references +TASK_VALUE_KEYS = { + "shell", + "command", + "ansible.builtin.shell", + "ansible.builtin.command", + "ansible.builtin.template", + "ansible.builtin.copy", + "ansible.builtin.debug", + "template", + "copy", + "debug", + "cmd", + "msg", + "content", +} + +# Keys that are NOT secret-bearing (task metadata, not values) +NON_VALUE_KEYS = { + "name", + "when", + "loop", + "loop_control", + "changed_when", + "failed_when", + "no_log", + "register", + "tags", + "vars", + "become", + "become_user", + "delegate_to", + "run_once", + "environment", + "with_items", + "with_dict", + "with_list", +} + + +def _contains_secret(value: object) -> bool: + """Recursively check if a value contains secret-like variable references.""" + if isinstance(value, str): + return any(p.search(value) for p in SECRET_PATTERNS) + if isinstance(value, dict): + return any(_contains_secret(v) for v in value.values()) + if isinstance(value, list): + return any(_contains_secret(item) for item in value) + return False + + +def _has_no_log(task: dict) -> bool: + """Check if a task has no_log set to a non-False value.""" + no_log = task.get("no_log", False) + # Jinja expressions (e.g. "{{ not debug_mode }}") count as set + return no_log is not False and no_log is not None + + +def _check_task(task: dict, file_path: Path, task_num: int) -> list[str]: + """Check a single task for missing no_log on secret values. + + Returns a list of violation messages (empty if OK). + """ + violations: list[str] = [] + + # Skip tasks that already have no_log + if _has_no_log(task): + return violations + + # Check all string values in the task for secret references + has_secrets = False + for key, value in task.items(): + if key in NON_VALUE_KEYS: + continue + # Check action module params (shell, command, copy, template, etc.) + if _contains_secret(value): + has_secrets = True + break + + if has_secrets: + task_name = task.get("name", "") + violations.append( + f"{file_path}:{task_num}: Task '{task_name}' references secrets " + f"but has no no_log. Add `no_log: true` or " + f'`no_log: "{{{{ not (debug_mode | default(false) | bool) }}}}"` ' + f"to prevent credential leakage in Ansible output." + ) + + return violations + + +def check_directory(ansible_dir: Path) -> list[str]: + """Check all Ansible task files in a directory tree.""" + all_violations: list[str] = [] + + # Find all task files + task_files = list(ansible_dir.rglob("tasks/*.yml")) + task_files += list(ansible_dir.rglob("tasks/*.yaml")) + # Also check playbook files + task_files += list(ansible_dir.glob("playbooks/*.yml")) + + for task_file in sorted(task_files): + # Skip molecule test files + if "molecule" in task_file.parts: + continue + + try: + with task_file.open() as f: + docs = list(yaml.safe_load_all(f)) + except (yaml.YAMLError, OSError): + continue + + for doc in docs: + if not doc: + continue + + # Task files are bare lists of tasks; playbook files are + # lists of plays (each play is a dict with 'hosts' key) + if isinstance(doc, list): + is_plays = isinstance(doc[0], dict) and "hosts" in doc[0] + if not is_plays: + for i, task in enumerate(doc): + if not isinstance(task, dict): + continue + all_violations.extend(_check_task(task, task_file, i + 1)) + continue + plays = doc + elif isinstance(doc, dict): + plays = [doc] + else: + continue + + for play in plays: + if not isinstance(play, dict): + continue + for task_section in ("tasks", "pre_tasks", "post_tasks", "handlers"): + tasks = play.get(task_section, []) + if not isinstance(tasks, list): + continue + for i, task in enumerate(tasks): + if not isinstance(task, dict): + continue + all_violations.extend(_check_task(task, task_file, i + 1)) + + return all_violations + + +@click.command() +@click.option( + "--path", + type=click.Path(exists=True, path_type=Path), + help="Check a specific file or directory (default: ansible/).", +) +@click.option( + "--ansible-dir", + type=click.Path(exists=True, path_type=Path), + default=None, + help="Override the default ansible directory (default: ansible/).", +) +def main(path: Path | None, ansible_dir: Path | None) -> None: + """Check that Ansible tasks handling secrets have no_log set.""" + target = path or ansible_dir or DEFAULT_ANSIBLE_DIR + if not target.is_dir(): + click.echo(f"Error: {target} is not a directory", err=True) + sys.exit(2) + + violations = check_directory(target) + + if violations: + click.echo(f"Found {len(violations)} task(s) handling secrets without no_log:\n") + for v in violations: + click.echo(f" {v}") + click.echo(f"\nTotal: {len(violations)} violation(s).") + sys.exit(1) + + click.echo(f"[check-ansible-no-log] All secret-handling tasks have no_log. ({target})") + + +if __name__ == "__main__": # pragma: no cover + main() diff --git a/src/devx/tools/check_ansible_no_state_absent_on_db.py b/src/devx/tools/check_ansible_no_state_absent_on_db.py new file mode 100644 index 0000000..07a3aa1 --- /dev/null +++ b/src/devx/tools/check_ansible_no_state_absent_on_db.py @@ -0,0 +1,176 @@ +"""Check Ansible tasks for ``state: absent`` on database data directories. + +This is a static analysis lint check that runs in CI (``make lint-ci``) +to prevent the class of bug that caused the 2026-07-22 production outage +(ADR-0028): a ``state: absent`` on a PostgreSQL data directory path that +fired on every deploy and wiped the ZITADEL database. + +The existing unit test ``scripts/tests/test_no_zitadel_db_wipe.py`` covers +the same concern as a regression test. This lint check runs earlier in +the pipeline (before tests) and covers ALL roles and playbooks, not just +the ZITADEL role. + +Allowed contexts (where DB recreation is legitimate): +- PostgreSQL major version upgrades (``upgrade-postgres``, ``PG_VERSION``) +- Explicit ``# lint:allow-state-absent`` comment on the task + +Usage:: + + python -m devx.tools.check_ansible_no_state_absent_on_db + python -m devx.tools.check_ansible_no_state_absent_on_db --path ansible/roles/zitadel/tasks/main.yml + +Exit code 0 if no violations found, 1 otherwise. +""" + +from __future__ import annotations + +import re +import sys +from pathlib import Path + +import click + +REPO_ROOT = Path.cwd() +DEFAULT_ANSIBLE_DIRS: list[Path] = [ + REPO_ROOT / "ansible" / "playbooks", + REPO_ROOT / "ansible" / "roles", +] + +# Database data directory path patterns. +# These match the DIRECTORY path, not individual files within it. +# Removing a stale config file (e.g. postgresql.conf) is safe; removing +# the entire data directory is not. +DB_PATH_PATTERNS = ( + re.compile(r"postgres/zitadel-db", re.IGNORECASE), + re.compile(r"postgres/\w+-db", re.IGNORECASE), + re.compile(r"/var/lib/postgresql/data", re.IGNORECASE), + re.compile(r"/var/lib/postgresql/data/\w+-db", re.IGNORECASE), +) + +# Destructive operations +DESTRUCTIVE_PATTERNS = ( + re.compile(r"state:\s*absent", re.IGNORECASE), + re.compile(r"rm\s+-rf.*\bdb\b", re.IGNORECASE), +) + +# Allowed contexts where DB recreation is legitimate +ALLOWED_CONTEXT_KEYWORDS = ( + "upgrade-postgres", + "PG_VERSION", + "pg_version", +) + +# Comment marker to explicitly allow state: absent on a specific task +ALLOW_MARKER = "lint:allow-state-absent" + + +def _find_task_files(base: Path) -> list[Path]: + """Find all YAML task files under a base directory, skipping molecule.""" + if base.is_file() and base.suffix in (".yml", ".yaml"): + return [base] + if not base.is_dir(): + return [] + files: list[Path] = [] + for f in sorted(base.rglob("*.yml")) + sorted(base.rglob("*.yaml")): + if "molecule" in f.parts: + continue + files.append(f) + return files + + +def _check_file(filepath: Path, repo_root: Path) -> list[str]: + """Check a YAML file for state: absent on DB data directory paths. + + Returns a list of violation messages (empty if clean). + """ + try: + content = filepath.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + return [] + + # Quick check: if no DB path pattern appears anywhere, skip + if not any(p.search(content) for p in DB_PATH_PATTERNS): + return [] + + try: + display_path = filepath.relative_to(repo_root) + except ValueError: + display_path = filepath + + violations: list[str] = [] + lines = content.splitlines() + + for i, line in enumerate(lines): + for db_pattern in DB_PATH_PATTERNS: + if not db_pattern.search(line): + continue + + # Check surrounding context (±5 lines) for destructive operations + context_start = max(0, i - 5) + context_end = min(len(lines), i + 6) + context = "\n".join(lines[context_start:context_end]) + + # Skip if in an allowed context (PG upgrade) + if any(kw in context for kw in ALLOWED_CONTEXT_KEYWORDS): + continue + + # Skip if the allow marker comment is in the context + if ALLOW_MARKER in context: + continue + + for dp in DESTRUCTIVE_PATTERNS: + if dp.search(context): + violations.append( + f"{display_path}:{i + 1} — destructive operation " + f"({dp.pattern!r}) near DB data directory path " + f"({db_pattern.pattern!r}). " + f"Database directories must never be wiped automatically (ADR-0028). " + f"If this is legitimate (e.g. PG upgrade), add " + f"#{ALLOW_MARKER} to the task." + ) + break + + 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: + """Check that no Ansible task uses state: absent on a DB data directory.""" + dirs = list(ansible_dirs) if ansible_dirs else DEFAULT_ANSIBLE_DIRS + if path: + files = _find_task_files(path) + else: + files: list[Path] = [] + for d in dirs: + files.extend(_find_task_files(d)) + + all_violations: list[str] = [] + for f in files: + all_violations.extend(_check_file(f, REPO_ROOT)) + + if all_violations: + click.echo("[check-ansible-no-state-absent-on-db] FAIL: destructive operations on DB paths:") + for v in all_violations: + click.echo(f" - {v}") + click.echo(f"\nTotal: {len(all_violations)} violation(s).") + click.echo("Database data directories must never be wiped automatically (ADR-0028).") + sys.exit(1) + else: + click.echo("[check-ansible-no-state-absent-on-db] OK: no destructive operations on DB paths.") + + +if __name__ == "__main__": # pragma: no cover + main() diff --git a/src/devx/tools/check_ansible_patterns.py b/src/devx/tools/check_ansible_patterns.py new file mode 100644 index 0000000..e07294d --- /dev/null +++ b/src/devx/tools/check_ansible_patterns.py @@ -0,0 +1,345 @@ +"""Check Ansible tasks for dangerous patterns that mask failures. + +This check addresses the gap identified in the testing-strategy audit: +the automated PR review only checks Python files, and ``ansible-lint`` +runs at ``profile: basic`` which does not catch dangerous patterns like: + +- ``|| true`` on tasks that are NOT cleanup/idempotency operations +- ``failed_when: false`` on critical tasks (e.g. DB operations) +- ``2>/dev/null`` on tasks where stderr contains important diagnostics + +Most ``|| true`` and ``2>/dev/null`` instances in the codebase are +legitimate (container removal, journalctl, apt-get, docker prune, SUID +removal). This check flags only instances that are NOT in a known-safe +context. Tasks can also opt out with a ``# lint:allow-failure-masking`` +comment. + +Usage:: + + python -m devx.tools.check_ansible_patterns + python -m devx.tools.check_ansible_patterns --path ansible/roles/app_container/tasks/main.yml + +Exit code 0 if no violations found, 1 otherwise. +""" + +from __future__ import annotations + +import re +import sys +from pathlib import Path + +import click +import yaml + +REPO_ROOT = Path.cwd() +DEFAULT_ANSIBLE_DIRS: list[Path] = [ + REPO_ROOT / "ansible" / "playbooks", + REPO_ROOT / "ansible" / "roles", +] + +# Comment marker to explicitly allow a pattern on a specific task +ALLOW_MARKER = "lint:allow-failure-masking" + +# Patterns that mask failures when used in shell/command tasks +OR_TRUE_PATTERN = re.compile(r"\|\|\s*true\b", re.IGNORECASE) +REDIRECT_DEVNULL_PATTERN = re.compile(r"2>/dev/null") + +# Module keys that accept shell/command strings +SHELL_MODULE_KEYS = frozenset( + { + "shell", + "command", + "ansible.builtin.shell", + "ansible.builtin.command", + "cmd", + "ansible.builtin.raw", + "raw", + } +) + +# Task keys whose values might contain shell commands +COMMAND_VALUE_KEYS = frozenset( + { + "shell", + "command", + "ansible.builtin.shell", + "ansible.builtin.command", + "cmd", + "raw", + "ansible.builtin.raw", + } +) + +# Legitimate contexts where || true or 2>/dev/null are safe. +# These are command prefixes or task names that indicate cleanup/idempotency. +LEGITIMATE_COMMAND_PREFIXES = ( + # Container/process removal (may not exist) + "docker rm", + "docker stop", + "docker rmi", + "docker network rm", + "docker volume rm", + "pkill", + "kill", + # Cleanup commands that are expected to sometimes fail + "journalctl --vacuum", + "apt-get clean", + "apt-get autoremove", + "docker image prune", + "docker container prune", + "docker volume prune", + "docker builder prune", + "find / -name", + # SUID removal (binaries may not exist) + "chmod", + "rm -f", + # Network connection checks (may fail if not connected) + "docker network connect", + # Prometheus snapshot API (may fail if no snapshot) + "curl.*api/v2/admin/tsdb/snapshot", +) + +LEGITIMATE_TASK_NAME_KEYWORDS = ( + "remove", + "cleanup", + "clean up", + "prune", + "purge", + "disconnect", + "stop", + "kill", + "strip suid", + "suid", + "vacuum", + "ensure.*absent", + "may not exist", + "if exists", + "optional", + "best effort", + "no-op", + "noop", + "idempotent", + "sync", +) + +# Tasks with failed_when: false that are critical and should not mask failures. +# Only flag operations that SHOULD fail loudly — writing secrets, provisioning +# users, creating OIDC apps. Do NOT flag stop/start/check/wait/migrate/restore +# operations where failed_when: false is legitimate (container may not exist, +# may already be stopped, etc.). +CRITICAL_TASK_KEYWORDS = ( + "password", + "secret", + "provision", + "oidc", +) + +# Task name keywords that indicate failed_when: false is legitimate +LEGITIMATE_FAILED_WHEN_KEYWORDS = ( + "stop", + "start", + "check", + "wait", + "migrate", + "restart", + "rebuild", + "restore", + "remove", + "cleanup", + "sync", + "download", + "extract", + "verify", +) + + +def _is_legitimate_or_true(command_str: str, task_name: str) -> bool: + """Check if a || true in a command is in a legitimate context.""" + # Check task name for legitimate keywords + name_lower = task_name.lower() + if any(re.search(kw, name_lower) for kw in LEGITIMATE_TASK_NAME_KEYWORDS): + return True + + # Check command prefix for legitimate patterns + cmd_lower = command_str.lower() + return any(re.search(prefix, cmd_lower) for prefix in LEGITIMATE_COMMAND_PREFIXES) + + +def _is_legitimate_devnull(command_str: str, task_name: str) -> bool: + """Check if a 2>/dev/null in a command is in a legitimate context.""" + # 2>/dev/null is almost always safe — it suppresses stderr noise. + # Only flag it if the task is critical (DB, backup, OIDC) AND + # there's no || true (which is the more dangerous pattern). + return _is_legitimate_or_true(command_str, task_name) + + +def _check_task(task: dict, filepath: Path, task_num: int, repo_root: Path) -> list[str]: + """Check a single task for dangerous failure-masking patterns.""" + violations: list[str] = [] + + try: + display_path = filepath.relative_to(repo_root) + except ValueError: + display_path = filepath + + task_name = task.get("name", "") + + # Check for the allow marker in the task name + # (YAML comments are not preserved by safe_load, so we check the + # task name for the marker as a workaround) + if ALLOW_MARKER in task_name: + return violations + + # Check for || true in command/shell values + for key in COMMAND_VALUE_KEYS: + value = task.get(key) + if value is None: + continue + value_str = str(value) + if OR_TRUE_PATTERN.search(value_str) and not _is_legitimate_or_true(value_str, task_name): + violations.append( + f"{display_path}:{task_num} — task '{task_name}' uses " + f"'|| true' in {key} which may mask real failures. " + f"If this is a cleanup/idempotency operation, rename the " + f"task to include 'remove'/'cleanup'/'prune' or add " + f"#{ALLOW_MARKER} to the task." + ) + + # Check for failed_when: false on critical tasks + failed_when = task.get("failed_when") + if failed_when is False: + name_lower = task_name.lower() + # Skip if the task name indicates a legitimate failed_when: false context + is_legitimate = any(kw in name_lower for kw in LEGITIMATE_FAILED_WHEN_KEYWORDS) + if not is_legitimate: + for kw in CRITICAL_TASK_KEYWORDS: + if kw in name_lower: + violations.append( + f"{display_path}:{task_num} — critical task '{task_name}' " + f"has failed_when: false, which masks failures on " + f"a {kw}-related operation. Remove failed_when: false " + f"or add #{ALLOW_MARKER} if masking is intentional." + ) + break + + return violations + + +def _check_file(filepath: Path, repo_root: Path) -> list[str]: + """Check a YAML file for dangerous failure-masking patterns.""" + try: + content = filepath.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + return [] + + # Quick check: if no patterns appear, skip + if not ( + OR_TRUE_PATTERN.search(content) or "failed_when: false" in content or REDIRECT_DEVNULL_PATTERN.search(content) + ): + return [] + + # Check for allow markers in comments + has_allow_marker = ALLOW_MARKER in content + + try: + docs = list(yaml.safe_load_all(content)) + except yaml.YAMLError: + return [] + + violations: list[str] = [] + + for doc in docs: + if not doc: + continue + if isinstance(doc, list): + for i, item in enumerate(doc): + if isinstance(item, dict): + if any(k in item for k in ("tasks", "pre_tasks", "post_tasks", "handlers")): + _check_tasks(item, filepath, violations, repo_root) + else: + violations.extend(_check_task(item, filepath, i + 1, repo_root)) + block = item.get("block") + if isinstance(block, list): + for j, bt in enumerate(block): + if isinstance(bt, dict): + violations.extend(_check_task(bt, filepath, i + j + 1, repo_root)) + elif isinstance(doc, dict): + _check_tasks(doc, filepath, violations, repo_root) + + # Filter out violations if the allow marker is present in the file + # (coarse-grained opt-out for files with many legitimate uses) + if has_allow_marker: + violations = [] + + return violations + + +def _check_tasks(doc: dict, filepath: Path, errors: list[str], repo_root: Path) -> None: + """Check top-level tasks and nested task sections in a playbook doc.""" + for section_key in ("tasks", "pre_tasks", "post_tasks", "handlers"): + section = doc.get(section_key) + if isinstance(section, list): + for i, task in enumerate(section): + if isinstance(task, dict): + errors.extend(_check_task(task, filepath, i + 1, repo_root)) + block = task.get("block") + if isinstance(block, list): + for j, bt in enumerate(block): + if isinstance(bt, dict): + errors.extend(_check_task(bt, filepath, i + j + 1, repo_root)) + + +def _find_task_files(base: Path) -> list[Path]: + """Find all YAML task files under a base directory, skipping molecule.""" + if base.is_file() and base.suffix in (".yml", ".yaml"): + return [base] + if not base.is_dir(): + return [] + files: list[Path] = [] + for f in sorted(base.rglob("*.yml")) + sorted(base.rglob("*.yaml")): + if "molecule" in f.parts: + continue + files.append(f) + return files + + +@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: + """Check Ansible tasks for dangerous failure-masking patterns.""" + dirs = list(ansible_dirs) if ansible_dirs else DEFAULT_ANSIBLE_DIRS + if path: + files = _find_task_files(path) + else: + files: list[Path] = [] + for d in dirs: + files.extend(_find_task_files(d)) + + all_violations: list[str] = [] + for f in files: + all_violations.extend(_check_file(f, REPO_ROOT)) + + if all_violations: + click.echo("[check-ansible-patterns] FAIL: dangerous failure-masking patterns found:") + for v in all_violations: + click.echo(f" - {v}") + click.echo(f"\nTotal: {len(all_violations)} violation(s).") + sys.exit(1) + else: + click.echo("[check-ansible-patterns] OK: no dangerous failure-masking patterns.") + + +if __name__ == "__main__": # pragma: no cover + main() diff --git a/src/devx/tools/check_jinja_expr.py b/src/devx/tools/check_jinja_expr.py new file mode 100644 index 0000000..40b8990 --- /dev/null +++ b/src/devx/tools/check_jinja_expr.py @@ -0,0 +1,292 @@ +"""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() diff --git a/tests/unit/test_molecule_changed.py b/tests/unit/test_molecule_changed.py new file mode 100644 index 0000000..eea39d6 --- /dev/null +++ b/tests/unit/test_molecule_changed.py @@ -0,0 +1,192 @@ +"""Unit tests for devx.molecule.molecule_changed. + +Verifies that the script correctly detects changed roles and maps +them to make targets. +""" + +from __future__ import annotations + +from unittest.mock import patch + +from click.testing import CliRunner + +from devx.molecule.molecule_changed import ( + detect_changed_roles, + get_changed_files, + main, + roles_to_targets, +) + + +def test_detect_role_change(): + """A file in ansible/roles// maps to that role.""" + files = ["ansible/roles/docker_base/tasks/main.yml"] + roles = detect_changed_roles(files) + assert "docker_base" in roles + + +def test_detect_playbook_change(): + """A playbook change maps to its included roles.""" + files = ["ansible/playbooks/deploy-observability.yml"] + roles = detect_changed_roles(files) + assert "observability" in roles + assert "docker_base" in roles + assert "zitadel" in roles + + +def test_detect_shared_infra_triggers_all(): + """ansible.cfg change triggers all roles.""" + files = ["ansible/ansible.cfg"] + roles = detect_changed_roles(files) + assert len(roles) == 10 # all roles + + +def test_detect_no_ansible_changes(): + """Non-Ansible files don't trigger any roles.""" + files = ["scripts/molecule_changed.py", "Makefile"] + roles = detect_changed_roles(files) + assert len(roles) == 0 + + +def test_roles_to_targets(): + """Role names map to make targets.""" + targets = roles_to_targets({"docker_base", "zitadel"}) + assert "molecule-docker-base" in targets + assert "molecule-zitadel" in targets + + +def test_roles_to_targets_unknown_role(): + """Unknown roles are silently skipped.""" + targets = roles_to_targets({"docker_base", "unknown_role"}) + assert targets == ["molecule-docker-base"] + + +def test_main_no_changes(): + """When no files changed, outputs message to stderr.""" + with patch("devx.molecule.molecule_changed.get_changed_files", return_value=[]): + runner = CliRunner() + result = runner.invoke(main, ["--print-targets"]) + assert result.exit_code == 0 + assert "No changed files" in result.output + + +def test_main_print_targets(): + """--print-targets outputs make targets.""" + with patch( + "devx.molecule.molecule_changed.get_changed_files", + return_value=["ansible/roles/docker_base/tasks/main.yml"], + ): + runner = CliRunner() + result = runner.invoke(main, ["--print-targets"]) + assert result.exit_code == 0 + assert "molecule-docker-base" in result.output + + +def test_main_print_roles(): + """--print-roles outputs role names.""" + with patch( + "devx.molecule.molecule_changed.get_changed_files", + return_value=["ansible/roles/zitadel/tasks/main.yml"], + ): + runner = CliRunner() + result = runner.invoke(main, ["--print-roles"]) + assert result.exit_code == 0 + assert "zitadel" in result.output + + +def test_main_no_ansible_changes(): + """When only non-Ansible files changed, outputs no scenarios message.""" + with patch( + "devx.molecule.molecule_changed.get_changed_files", + return_value=["scripts/molecule_changed.py"], + ): + runner = CliRunner() + result = runner.invoke(main, ["--print-targets"]) + assert result.exit_code == 0 + assert "No molecule scenarios" in result.output + + +def test_get_changed_files_with_mock(): + """get_changed_files returns files from git diff.""" + with patch("devx.molecule.molecule_changed._run_git", return_value="file1\nfile2\n"): + files = get_changed_files("origin/master") + assert files == ["file1", "file2"] + + +def test_get_changed_files_falls_back_to_master(): + """When base ref has no diff, falls back to master.""" + calls: list[list[str]] = [] + + def mock_git(args): + calls.append(args) + # First call (origin/master) returns empty, second (master) returns files + if "origin/master...HEAD" in args[2]: + return "" + return "ansible/roles/docker_base/tasks/main.yml\n" + + with patch("devx.molecule.molecule_changed._run_git", side_effect=mock_git): + files = get_changed_files("origin/master") + assert files == ["ansible/roles/docker_base/tasks/main.yml"] + assert len(calls) == 2 + + +def test_get_changed_files_empty(): + """When no changes in either ref, returns empty list.""" + with patch("devx.molecule.molecule_changed._run_git", return_value=""): + files = get_changed_files("origin/master") + assert files == [] + + +def test_detect_molecule_shared_path(): + """ansible/molecule/ change triggers all roles.""" + files = ["ansible/molecule/Dockerfile"] + roles = detect_changed_roles(files) + assert len(roles) == 10 + + +def test_detect_requirements_yml_triggers_all(): + """ansible/requirements.yml change triggers all roles.""" + files = ["ansible/requirements.yml"] + roles = detect_changed_roles(files) + assert len(roles) == 10 + + +def test_detect_configure_oidc_playbook(): + """configure-oidc.yml maps to sso_config and app_container.""" + files = ["ansible/playbooks/configure-oidc.yml"] + roles = detect_changed_roles(files) + assert "sso_config" in roles + assert "app_container" in roles + + +def test_detect_prepare_vms_playbook(): + """prepare-vms.yml maps to all base roles.""" + files = ["ansible/playbooks/prepare-vms.yml"] + roles = detect_changed_roles(files) + assert "docker_base" in roles + assert "app_hardening" in roles + assert "storage" in roles + assert "disk_cleanup" in roles + assert "crowdsec" in roles + + +def test_detect_deploy_customer_playbook(): + """deploy-customer.yml maps to its roles.""" + files = ["ansible/playbooks/deploy-customer.yml"] + roles = detect_changed_roles(files) + assert "app_container" in roles + assert "docker_base" in roles + assert "app_hardening" in roles + assert "sso_config" in roles + + +def test_main_default_base(): + """main() with no --base uses origin/master.""" + with patch( + "devx.molecule.molecule_changed.get_changed_files", + return_value=["ansible/roles/zitadel/tasks/main.yml"], + ) as mock: + runner = CliRunner() + result = runner.invoke(main, ["--print-roles"]) + assert result.exit_code == 0 + mock.assert_called_once_with("origin/master") diff --git a/tests/unit/test_tools_check_ansible_no_log.py b/tests/unit/test_tools_check_ansible_no_log.py new file mode 100644 index 0000000..f618f25 --- /dev/null +++ b/tests/unit/test_tools_check_ansible_no_log.py @@ -0,0 +1,367 @@ +"""Unit tests for devx.tools.check_ansible_no_log.""" + +from __future__ import annotations + +from pathlib import Path + +from click.testing import CliRunner + +from devx.tools.check_ansible_no_log import _check_task, check_directory, main + + +def _make_task(name: str, action: str, value: str, **extra: object) -> dict: + """Build a minimal task dict for testing.""" + task: dict = {"name": name, action: value} + task.update(extra) + return task + + +class TestCheckTask: + def test_task_with_secret_and_no_log_passes(self): + task = _make_task( + "Safe task", + "ansible.builtin.shell", + "echo {{ _secrets.mattermost_admin_password }}", + no_log=True, + ) + assert _check_task(task, Path("test.yml"), 1) == [] + + def test_task_with_secret_and_no_no_log_fails(self): + task = _make_task( + "Unsafe task", + "ansible.builtin.shell", + "echo {{ _secrets.mattermost_admin_password }}", + ) + violations = _check_task(task, Path("test.yml"), 1) + assert len(violations) == 1 + assert "no_log" in violations[0] + + def test_task_without_secret_passes(self): + task = _make_task( + "Normal task", + "ansible.builtin.shell", + "echo hello world", + ) + assert _check_task(task, Path("test.yml"), 1) == [] + + def test_task_with_jinja_no_log_passes(self): + task = _make_task( + "Safe task with jinja no_log", + "ansible.builtin.shell", + "echo {{ _secrets.mattermost_admin_password }}", + no_log="{{ not (debug_mode | default(false) | bool) }}", + ) + assert _check_task(task, Path("test.yml"), 1) == [] + + def test_task_with_password_in_name_only_no_false_positive(self): + """Task name contains 'password' but no secret value — should not flag.""" + task = _make_task( + "Configure passwdqc in common-password", + "ansible.builtin.lineinfile", + "password required pam_passwdqc.so min=disabled,disabled,16,12,8", + ) + assert _check_task(task, Path("test.yml"), 1) == [] + + def test_task_with_password_in_module_param_no_false_positive(self): + """Module param named 'password' but value is a literal — no Jinja.""" + task = { + "name": "Set user password", + "ansible.builtin.user": { + "name": "deploy", + "password_lock": True, + }, + } + assert _check_task(task, Path("test.yml"), 1) == [] + + def test_task_with_vault_password_variable_fails(self): + task = _make_task( + "Unsafe vault task", + "ansible.builtin.shell", + "echo {{ vault_zitadel_db_password }}", + ) + violations = _check_task(task, Path("test.yml"), 1) + assert len(violations) == 1 + + def test_task_with_nested_dict_secret_fails(self): + """Secrets in nested dict values (e.g. set_fact) should be caught.""" + task = { + "name": "Set secrets", + "ansible.builtin.set_fact": { + "db_password": "{{ vault_db_password }}", + "api_key": "{{ vault_api_key }}", + }, + } + violations = _check_task(task, Path("test.yml"), 1) + assert len(violations) == 1 + + def test_task_with_no_log_none_passes(self): + """no_log: None should count as not set (flagged).""" + task = _make_task( + "Unsafe task", + "ansible.builtin.shell", + "echo {{ _secrets.db_password }}", + no_log=None, + ) + violations = _check_task(task, Path("test.yml"), 1) + assert len(violations) == 1 + + def test_task_with_secret_in_list_value_fails(self): + """Secrets inside list values should be caught.""" + task = { + "name": "Task with list secret", + "ansible.builtin.set_fact": { + "items": ["{{ _secrets.api_key }}", "normal_value"], + }, + } + violations = _check_task(task, Path("test.yml"), 1) + assert len(violations) == 1 + + def test_task_with_api_key_secret_fails(self): + """api_key in Jinja expression should be caught.""" + task = _make_task( + "Unsafe task", + "ansible.builtin.shell", + "echo {{ my_api_key }}", + ) + violations = _check_task(task, Path("test.yml"), 1) + assert len(violations) == 1 + + def test_task_with_secret_in_jinja_fails(self): + """_secret in Jinja expression should be caught.""" + task = _make_task( + "Unsafe task", + "ansible.builtin.shell", + "echo {{ my_secret }}", + ) + violations = _check_task(task, Path("test.yml"), 1) + assert len(violations) == 1 + + def test_task_with_access_token_fails(self): + """access_token in Jinja expression should be caught.""" + task = _make_task( + "Unsafe task", + "ansible.builtin.shell", + "echo {{ my_access_token }}", + ) + violations = _check_task(task, Path("test.yml"), 1) + assert len(violations) == 1 + + def test_task_with_non_secret_non_dict_non_list_value(self): + """Non-str, non-dict, non-list values (e.g. int) should not crash.""" + task = _make_task( + "Task with int", + "ansible.builtin.shell", + "echo hello", + some_int=42, + ) + assert _check_task(task, Path("test.yml"), 1) == [] + + +class TestCheckDirectory: + def test_clean_directory_passes(self, tmp_path: Path): + """A directory with no secret-handling tasks should pass.""" + role_dir = tmp_path / "roles" / "test_role" + (role_dir / "tasks").mkdir(parents=True) + (role_dir / "tasks" / "main.yml").write_text( + "- name: Normal task\n ansible.builtin.shell: echo hello\n changed_when: false\n" + ) + assert check_directory(role_dir) == [] + + def test_unsafe_task_is_caught(self, tmp_path: Path): + """A task with secrets but no no_log should be flagged.""" + role_dir = tmp_path / "roles" / "test_role" + (role_dir / "tasks").mkdir(parents=True) + (role_dir / "tasks" / "main.yml").write_text( + "- name: Unsafe task\n ansible.builtin.shell: echo {{ _secrets.db_password }}\n changed_when: false\n" + ) + violations = check_directory(role_dir) + assert len(violations) == 1 + assert "Unsafe task" in violations[0] + + def test_molecule_files_are_skipped(self, tmp_path: Path): + """Molecule test files should not be scanned.""" + role_dir = tmp_path / "roles" / "test_role" + mol_dir = role_dir / "molecule" / "default" / "tasks" + mol_dir.mkdir(parents=True) + (mol_dir / "main.yml").write_text( + "- name: Unsafe task in molecule\n ansible.builtin.shell: echo {{ _secrets.db_password }}\n" + ) + assert check_directory(role_dir) == [] + + def test_playbook_format_is_parsed(self, tmp_path: Path): + """Playbook files (list of plays with 'hosts') should be parsed.""" + pb_dir = tmp_path / "playbooks" + pb_dir.mkdir(parents=True) + (pb_dir / "test.yml").write_text( + "---\n" + "- name: Test play\n" + " hosts: all\n" + " tasks:\n" + " - name: Unsafe task\n" + " ansible.builtin.shell: echo {{ _secrets.db_password }}\n" + ) + violations = check_directory(tmp_path) + assert len(violations) == 1 + assert "Unsafe task" in violations[0] + + def test_invalid_yaml_is_skipped(self, tmp_path: Path): + """Invalid YAML files should be skipped, not crash.""" + role_dir = tmp_path / "roles" / "test_role" + (role_dir / "tasks").mkdir(parents=True) + (role_dir / "tasks" / "main.yml").write_text("{{ invalid yaml: [") + assert check_directory(role_dir) == [] + + def test_empty_yaml_doc_is_skipped(self, tmp_path: Path): + """Empty YAML documents (None) should be skipped.""" + role_dir = tmp_path / "roles" / "test_role" + (role_dir / "tasks").mkdir(parents=True) + (role_dir / "tasks" / "main.yml").write_text("---\n") + assert check_directory(role_dir) == [] + + def test_non_dict_non_list_doc_is_skipped(self, tmp_path: Path): + """YAML docs that are neither dict nor list should be skipped.""" + role_dir = tmp_path / "roles" / "test_role" + (role_dir / "tasks").mkdir(parents=True) + (role_dir / "tasks" / "main.yml").write_text("just a string\n") + assert check_directory(role_dir) == [] + + def test_task_file_with_non_dict_task_skipped(self, tmp_path: Path): + """Non-dict items in a task list should be skipped.""" + role_dir = tmp_path / "roles" / "test_role" + (role_dir / "tasks").mkdir(parents=True) + (role_dir / "tasks" / "main.yml").write_text( + "- just a string\n- name: Safe task\n ansible.builtin.shell: echo hello\n" + ) + assert check_directory(role_dir) == [] + + def test_secret_in_list_value_is_caught(self, tmp_path: Path): + """Secrets inside list values should be caught.""" + role_dir = tmp_path / "roles" / "test_role" + (role_dir / "tasks").mkdir(parents=True) + (role_dir / "tasks" / "main.yml").write_text( + "- name: Task with list secret\n" + " ansible.builtin.set_fact:\n" + " items:\n" + ' - "{{ _secrets.api_key }}"\n' + " - normal_value\n" + ) + violations = check_directory(role_dir) + assert len(violations) == 1 + + def test_single_play_dict_format(self, tmp_path: Path): + """A playbook that's a bare dict (not list of plays) should be parsed.""" + pb_dir = tmp_path / "playbooks" + pb_dir.mkdir(parents=True) + (pb_dir / "test.yml").write_text( + "---\n" + "name: Single play\n" + "hosts: all\n" + "tasks:\n" + " - name: Unsafe task\n" + " ansible.builtin.shell: echo {{ _secrets.db_password }}\n" + ) + violations = check_directory(tmp_path) + assert len(violations) == 1 + + def test_play_with_non_dict_play_skipped(self, tmp_path: Path): + """Non-dict plays in a playbook list should be skipped.""" + pb_dir = tmp_path / "playbooks" + pb_dir.mkdir(parents=True) + # First play is valid (makes is_plays=True), second is a non-dict + (pb_dir / "test.yml").write_text( + "---\n" + "- name: Safe play\n" + " hosts: all\n" + " tasks:\n" + " - name: Safe task\n" + " ansible.builtin.shell: echo hello\n" + '- "just a string as second play"\n' + ) + assert check_directory(tmp_path) == [] + + def test_play_with_non_list_tasks_skipped(self, tmp_path: Path): + """Plays where tasks is not a list should be skipped.""" + pb_dir = tmp_path / "playbooks" + pb_dir.mkdir(parents=True) + (pb_dir / "test.yml").write_text('---\n- name: Play with bad tasks\n hosts: all\n tasks: "not a list"\n') + assert check_directory(tmp_path) == [] + + def test_play_with_non_dict_task_in_playbook(self, tmp_path: Path): + """Non-dict tasks in a playbook should be skipped.""" + pb_dir = tmp_path / "playbooks" + pb_dir.mkdir(parents=True) + (pb_dir / "test.yml").write_text( + "---\n" + "- name: Play\n" + " hosts: all\n" + " tasks:\n" + ' - "just a string"\n' + " - name: Safe task\n" + " ansible.builtin.shell: echo hello\n" + ) + assert check_directory(tmp_path) == [] + + def test_yaml_file_with_oserror_skipped(self, tmp_path: Path): + """YAML files that can't be opened should be skipped.""" + role_dir = tmp_path / "roles" / "test_role" + (role_dir / "tasks").mkdir(parents=True) + # Create a file that will cause OSError when opened + # (use a directory with .yml extension) + bad_file = role_dir / "tasks" / "main.yml" + bad_file.mkdir() + assert check_directory(role_dir) == [] + + +class TestMain: + def test_main_passes_on_clean_dir(self, tmp_path: Path): + """main() should exit 0 on a clean directory.""" + role_dir = tmp_path / "roles" / "test_role" + (role_dir / "tasks").mkdir(parents=True) + (role_dir / "tasks" / "main.yml").write_text( + "- name: Normal task\n ansible.builtin.shell: echo hello\n changed_when: false\n" + ) + runner = CliRunner() + result = runner.invoke(main, ["--path", str(role_dir)]) + assert result.exit_code == 0 + assert "OK" in result.output or "no_log" in result.output + + def test_main_fails_on_unsafe_dir(self, tmp_path: Path): + """main() should exit 1 when violations are found.""" + role_dir = tmp_path / "roles" / "test_role" + (role_dir / "tasks").mkdir(parents=True) + (role_dir / "tasks" / "main.yml").write_text( + "- name: Unsafe task\n ansible.builtin.shell: echo {{ _secrets.db_password }}\n" + ) + runner = CliRunner() + result = runner.invoke(main, ["--path", str(role_dir)]) + assert result.exit_code == 1 + assert "Unsafe task" in result.output + + def test_main_returns_2_on_missing_dir(self, tmp_path: Path): + """main() should exit 2 when the directory doesn't exist.""" + runner = CliRunner() + result = runner.invoke(main, ["--path", str(tmp_path / "nonexistent")]) + assert result.exit_code == 2 + + def test_main_with_ansible_dir_option(self, tmp_path: Path): + """main() --ansible-dir should work like --path.""" + role_dir = tmp_path / "roles" / "test_role" + (role_dir / "tasks").mkdir(parents=True) + (role_dir / "tasks" / "main.yml").write_text( + "- name: Unsafe task\n ansible.builtin.shell: echo {{ _secrets.db_password }}\n" + ) + runner = CliRunner() + result = runner.invoke(main, ["--ansible-dir", str(tmp_path)]) + assert result.exit_code == 1 + + def test_main_no_path_no_ansible_dir_uses_default(self, tmp_path: Path, monkeypatch): + """main() with no args uses DEFAULT_ANSIBLE_DIR.""" + import devx.tools.check_ansible_no_log as mod + + role_dir = tmp_path / "roles" / "test_role" + (role_dir / "tasks").mkdir(parents=True) + (role_dir / "tasks" / "main.yml").write_text("- name: Normal task\n ansible.builtin.shell: echo hello\n") + monkeypatch.setattr(mod, "DEFAULT_ANSIBLE_DIR", tmp_path) + runner = CliRunner() + result = runner.invoke(main, []) + assert result.exit_code == 0 diff --git a/tests/unit/test_tools_check_ansible_no_state_absent_on_db.py b/tests/unit/test_tools_check_ansible_no_state_absent_on_db.py new file mode 100644 index 0000000..dc8d055 --- /dev/null +++ b/tests/unit/test_tools_check_ansible_no_state_absent_on_db.py @@ -0,0 +1,198 @@ +"""Unit tests for devx.tools.check_ansible_no_state_absent_on_db.""" + +from __future__ import annotations + +from pathlib import Path + +from click.testing import CliRunner + +from devx.tools.check_ansible_no_state_absent_on_db import _check_file, _find_task_files, main + + +class TestCheckFile: + def test_clean_file_no_db_paths(self, tmp_path: Path): + """A file with no DB paths should produce no violations.""" + p = tmp_path / "test.yml" + p.write_text("- name: Safe task\n ansible.builtin.file:\n path: /opt/app/data\n state: directory\n") + assert _check_file(p, tmp_path) == [] + + def test_state_absent_on_zitadel_db_fails(self, tmp_path: Path): + """state: absent on zitadel-db path should be flagged.""" + p = tmp_path / "test.yml" + p.write_text( + "- name: Dangerous wipe\n ansible.builtin.file:\n path: /opt/postgres/zitadel-db\n state: absent\n" + ) + violations = _check_file(p, tmp_path) + assert len(violations) >= 1 + assert "state" in violations[0].lower() or "absent" in violations[0].lower() + + def test_state_absent_on_var_lib_postgresql_fails(self, tmp_path: Path): + """state: absent on /var/lib/postgresql/data should be flagged.""" + p = tmp_path / "test.yml" + p.write_text( + "- name: Dangerous wipe\n ansible.builtin.file:\n path: /var/lib/postgresql/data\n state: absent\n" + ) + violations = _check_file(p, tmp_path) + assert len(violations) >= 1 + + def test_state_absent_on_app_db_fails(self, tmp_path: Path): + """state: absent on any *-db path should be flagged.""" + p = tmp_path / "test.yml" + p.write_text( + "- name: Dangerous wipe\n ansible.builtin.file:\n path: /opt/postgres/gitea-db\n state: absent\n" + ) + violations = _check_file(p, tmp_path) + assert len(violations) >= 1 + + def test_state_absent_with_pg_upgrade_context_passes(self, tmp_path: Path): + """state: absent near DB path with upgrade-postgres context should pass.""" + p = tmp_path / "test.yml" + p.write_text( + "- name: PG upgrade — remove old data\n" + " ansible.builtin.file:\n" + " path: /opt/postgres/zitadel-db\n" + " state: absent\n" + " when: pg_version_changed | default(false)\n" + ) + violations = _check_file(p, tmp_path) + assert violations == [] + + def test_state_absent_with_pg_version_context_passes(self, tmp_path: Path): + """state: absent near DB path with PG_VERSION context should pass.""" + p = tmp_path / "test.yml" + p.write_text( + "- name: PG upgrade\n" + " ansible.builtin.file:\n" + " path: /opt/postgres/zitadel-db\n" + " state: absent\n" + " when: PG_VERSION is defined\n" + ) + violations = _check_file(p, tmp_path) + assert violations == [] + + def test_state_absent_with_allow_marker_passes(self, tmp_path: Path): + """state: absent with lint:allow-state-absent comment should pass.""" + p = tmp_path / "test.yml" + p.write_text( + "# lint:allow-state-absent\n" + "- name: Intentional wipe\n" + " ansible.builtin.file:\n" + " path: /opt/postgres/zitadel-db\n" + " state: absent\n" + ) + violations = _check_file(p, tmp_path) + assert violations == [] + + def test_state_present_on_db_path_passes(self, tmp_path: Path): + """state: present (not absent) on DB path should pass.""" + p = tmp_path / "test.yml" + p.write_text( + "- name: Safe task\n ansible.builtin.file:\n path: /opt/postgres/zitadel-db\n state: directory\n" + ) + assert _check_file(p, tmp_path) == [] + + def test_nonexistent_file_returns_empty(self): + """A nonexistent file should return no violations.""" + assert _check_file(Path("/nonexistent/path/file.yml"), Path.cwd()) == [] + + def test_rm_rf_db_fails(self, tmp_path: Path): + """rm -rf on a DB path should be flagged.""" + p = tmp_path / "test.yml" + p.write_text("- name: Dangerous wipe\n ansible.builtin.shell: rm -rf /opt/postgres/zitadel-db\n") + violations = _check_file(p, tmp_path) + assert len(violations) >= 1 + + def test_relative_path_outside_repo(self, tmp_path: Path): + """Files outside repo_root use the full path in display.""" + p = tmp_path / "test.yml" + p.write_text( + "- name: Dangerous wipe\n ansible.builtin.file:\n path: /opt/postgres/zitadel-db\n state: absent\n" + ) + violations = _check_file(p, Path("/other/repo")) + assert len(violations) >= 1 + assert str(tmp_path) in violations[0] or "test.yml" in violations[0] + + +class TestFindTaskFiles: + def test_find_yml_files_in_directory(self, tmp_path: Path): + """Should find .yml files in a directory.""" + (tmp_path / "tasks").mkdir() + (tmp_path / "tasks" / "main.yml").write_text("[]") + (tmp_path / "tasks" / "other.yaml").write_text("[]") + files = _find_task_files(tmp_path) + assert len(files) == 2 + + def test_skip_molecule_files(self, tmp_path: Path): + """Should skip files in molecule directories.""" + (tmp_path / "molecule").mkdir() + (tmp_path / "molecule" / "test.yml").write_text("[]") + (tmp_path / "main.yml").write_text("[]") + files = _find_task_files(tmp_path) + assert len(files) == 1 + assert "molecule" not in files[0].parts + + def test_single_file_input(self, tmp_path: Path): + """Should return the file itself if it's a .yml file.""" + f = tmp_path / "test.yml" + f.write_text("[]") + files = _find_task_files(f) + assert files == [f] + + def test_nonexistent_path_returns_empty(self): + """A path that is neither a file nor a dir should return [].""" + files = _find_task_files(Path("/nonexistent/path/that/does/not/exist")) + assert files == [] + + def test_non_yaml_file_skipped(self, tmp_path: Path): + """Non-YAML files should not be included.""" + f = tmp_path / "readme.txt" + f.write_text("not yaml") + assert _find_task_files(f) == [] + + +class TestMain: + def test_main_no_violations_exit_zero(self, tmp_path: Path): + """main() with a clean file should exit 0.""" + f = tmp_path / "clean.yml" + f.write_text("- name: Safe task\n ansible.builtin.file:\n path: /opt/app\n state: directory\n") + result = CliRunner().invoke(main, ["--path", str(f)]) + assert result.exit_code == 0 + assert "OK" in result.output + + def test_main_with_violations_exit_one(self, tmp_path: Path): + """main() with a state: absent on a DB path should exit 1.""" + f = tmp_path / "dangerous.yml" + f.write_text( + "- name: Dangerous wipe\n ansible.builtin.file:\n path: /opt/postgres/zitadel-db\n state: absent\n" + ) + result = CliRunner().invoke(main, ["--path", str(f)]) + assert result.exit_code == 1 + assert "FAIL" in result.output + + def test_main_path_to_clean_file(self, tmp_path: Path): + """main() --path pointing to a specific clean file should exit 0.""" + f = tmp_path / "tasks.yml" + f.write_text("- name: Safe\n ansible.builtin.file:\n path: /opt/app\n state: directory\n") + result = CliRunner().invoke(main, ["--path", str(f)]) + assert result.exit_code == 0 + + def test_main_default_dirs_no_violations(self, tmp_path: Path, monkeypatch): + """main() with no --path scans default dirs and exits 0.""" + import devx.tools.check_ansible_no_state_absent_on_db as mod + + (tmp_path / "clean.yml").write_text( + "- name: Safe\n ansible.builtin.file:\n path: /opt/app\n state: directory\n" + ) + monkeypatch.setattr(mod, "DEFAULT_ANSIBLE_DIRS", [tmp_path]) + result = CliRunner().invoke(main, []) + assert result.exit_code == 0 + assert "OK" in result.output + + def test_main_custom_ansible_dirs(self, tmp_path: Path): + """main() --ansible-dir should work.""" + f = tmp_path / "dangerous.yml" + f.write_text( + "- name: Dangerous wipe\n ansible.builtin.file:\n path: /opt/postgres/zitadel-db\n state: absent\n" + ) + result = CliRunner().invoke(main, ["--ansible-dir", str(tmp_path)]) + assert result.exit_code == 1 diff --git a/tests/unit/test_tools_check_ansible_patterns.py b/tests/unit/test_tools_check_ansible_patterns.py new file mode 100644 index 0000000..9b962f7 --- /dev/null +++ b/tests/unit/test_tools_check_ansible_patterns.py @@ -0,0 +1,340 @@ +"""Unit tests for devx.tools.check_ansible_patterns.""" + +from __future__ import annotations + +from pathlib import Path + +from click.testing import CliRunner + +from devx.tools.check_ansible_patterns import ( + _check_file, + _check_task, + _check_tasks, + _find_task_files, + _is_legitimate_devnull, + _is_legitimate_or_true, + main, +) + + +def _make_task(name: str, action: str, value: str, **extra: object) -> dict: + """Build a minimal task dict for testing.""" + task: dict = {"name": name, action: value} + task.update(extra) + return task + + +class TestIsLegitimateOrTrue: + def test_cleanup_task_name_is_legitimate(self): + assert _is_legitimate_or_true("docker rm old-container", "Remove old container") + + def test_prune_task_name_is_legitimate(self): + assert _is_legitimate_or_true("docker image prune -f", "Prune unused images") + + def test_docker_rm_command_is_legitimate(self): + assert _is_legitimate_or_true("docker rm -f mycontainer", "Some task") + + def test_provision_task_is_not_legitimate(self): + assert not _is_legitimate_or_true("curl -X POST https://api/app || true", "Provision OIDC client") + + def test_sync_task_name_is_legitimate(self): + assert _is_legitimate_or_true("psql -c 'ALTER USER' || true", "Sync PostgreSQL password") + + +class TestCheckTask: + def test_or_true_on_provision_task_fails(self, tmp_path: Path): + task = _make_task( + "Provision OIDC client", + "ansible.builtin.shell", + "curl -X POST https://zitadel/api || true", + ) + violations = _check_task(task, tmp_path / "test.yml", 1, tmp_path) + assert len(violations) >= 1 + assert "|| true" in violations[0] + + def test_or_true_on_cleanup_task_passes(self, tmp_path: Path): + task = _make_task( + "Remove old container", + "ansible.builtin.shell", + "docker rm -f old-container || true", + ) + assert _check_task(task, tmp_path / "test.yml", 1, tmp_path) == [] + + def test_failed_when_false_on_provision_fails(self, tmp_path: Path): + task = _make_task( + "Provision OIDC client", + "ansible.builtin.shell", + "curl -X POST https://zitadel/api", + failed_when=False, + ) + violations = _check_task(task, tmp_path / "test.yml", 1, tmp_path) + assert any("failed_when" in v for v in violations) + + def test_failed_when_false_on_stop_passes(self, tmp_path: Path): + task = _make_task( + "Stop ZITADEL containers", + "ansible.builtin.shell", + "docker stop zitadel", + failed_when=False, + ) + assert _check_task(task, tmp_path / "test.yml", 1, tmp_path) == [] + + def test_failed_when_false_on_check_passes(self, tmp_path: Path): + task = _make_task( + "Check if ZITADEL is running", + "ansible.builtin.shell", + "docker inspect zitadel", + failed_when=False, + ) + assert _check_task(task, tmp_path / "test.yml", 1, tmp_path) == [] + + def test_allow_marker_in_name_passes(self, tmp_path: Path): + task = _make_task( + "Provision OIDC #lint:allow-failure-masking", + "ansible.builtin.shell", + "curl -X POST https://zitadel/api || true", + failed_when=False, + ) + assert _check_task(task, tmp_path / "test.yml", 1, tmp_path) == [] + + def test_safe_task_no_violations(self, tmp_path: Path): + task = _make_task( + "Create directory", + "ansible.builtin.file", + "path=/opt/app state=directory", + ) + assert _check_task(task, tmp_path / "test.yml", 1, tmp_path) == [] + + def test_relative_path_outside_repo(self, tmp_path: Path): + """Files outside repo_root use the full path in display.""" + task = _make_task( + "Provision OIDC", + "ansible.builtin.shell", + "curl || true", + ) + other_dir = Path("/tmp/other") + violations = _check_task(task, other_dir / "test.yml", 1, tmp_path) + assert len(violations) >= 1 + + +class TestCheckFile: + def test_clean_file_passes(self, tmp_path: Path): + p = tmp_path / "test.yml" + p.write_text("- name: Safe task\n ansible.builtin.file:\n path: /opt/app\n state: directory\n") + assert _check_file(p, tmp_path) == [] + + def test_dangerous_pattern_detected(self, tmp_path: Path): + p = tmp_path / "test.yml" + p.write_text( + "- name: Provision OIDC\n" + " ansible.builtin.shell: |\n" + " curl -X POST https://api/app || true\n" + " failed_when: false\n" + ) + violations = _check_file(p, tmp_path) + assert len(violations) >= 1 + + def test_file_level_allow_marker_passes(self, tmp_path: Path): + p = tmp_path / "test.yml" + p.write_text( + "# lint:allow-failure-masking\n" + "- name: Provision OIDC\n" + " ansible.builtin.shell: |\n" + " curl -X POST https://api/app || true\n" + " failed_when: false\n" + ) + assert _check_file(p, tmp_path) == [] + + def test_nonexistent_file_returns_empty(self): + assert _check_file(Path("/nonexistent/path/file.yml"), Path.cwd()) == [] + + def test_yaml_parse_error_returns_empty(self, tmp_path: Path): + p = tmp_path / "test.yml" + p.write_text("name: Provision OIDC\n shell: curl || true\n: invalid: [") + assert _check_file(p, tmp_path) == [] + + def test_dict_doc_playbook_with_tasks(self, tmp_path: Path): + p = tmp_path / "playbook.yml" + p.write_text( + "- hosts: all\n" + " tasks:\n" + " - name: Provision OIDC\n" + " ansible.builtin.shell: curl -X POST https://api/app || true\n" + ) + violations = _check_file(p, tmp_path) + assert any("|| true" in v for v in violations) + + def test_dict_doc_with_pre_tasks_and_post_tasks(self, tmp_path: Path): + p = tmp_path / "playbook.yml" + p.write_text( + "- hosts: all\n" + " pre_tasks:\n" + " - name: Provision secret\n" + " ansible.builtin.shell: curl -X POST https://api/app || true\n" + " post_tasks:\n" + " - name: Provision OIDC\n" + " ansible.builtin.shell: curl -X POST https://api/app || true\n" + " handlers:\n" + " - name: Provision password\n" + " ansible.builtin.shell: curl -X POST https://api/app || true\n" + ) + violations = _check_file(p, tmp_path) + assert len(violations) >= 3 + + def test_block_tasks_in_list_item(self, tmp_path: Path): + p = tmp_path / "tasks.yml" + p.write_text( + "- name: Outer task\n" + " block:\n" + " - name: Provision OIDC\n" + " ansible.builtin.shell: curl -X POST https://api/app || true\n" + " - name: Provision secret\n" + " ansible.builtin.shell: curl -X POST https://api/app || true\n" + ) + violations = _check_file(p, tmp_path) + assert any("|| true" in v for v in violations) + + def test_empty_doc_skipped(self, tmp_path: Path): + p = tmp_path / "test.yml" + p.write_text("---\nnull\n---\n- name: Provision OIDC\n ansible.builtin.shell: curl || true\n") + violations = _check_file(p, tmp_path) + assert any("|| true" in v for v in violations) + + def test_pure_dict_doc_with_tasks(self, tmp_path: Path): + p = tmp_path / "playbook.yml" + p.write_text( + "hosts: all\n" + "tasks:\n" + " - name: Provision OIDC\n" + " ansible.builtin.shell: curl -X POST https://api/app || true\n" + ) + violations = _check_file(p, tmp_path) + assert any("|| true" in v for v in violations) + + +class TestIsLegitimateDevnull: + def test_cleanup_task_is_legitimate(self): + assert _is_legitimate_devnull("docker rm old-container 2>/dev/null", "Remove old container") + + def test_provision_task_is_not_legitimate(self): + assert not _is_legitimate_devnull("curl -X POST https://api/app 2>/dev/null", "Provision OIDC client") + + +class TestCheckTasks: + def test_tasks_section_checked(self, tmp_path: Path): + doc = { + "tasks": [ + {"name": "Provision OIDC", "ansible.builtin.shell": "curl || true"}, + ], + } + errors: list[str] = [] + _check_tasks(doc, tmp_path / "test.yml", errors, tmp_path) + assert any("|| true" in e for e in errors) + + def test_block_inside_tasks_section(self, tmp_path: Path): + doc = { + "tasks": [ + { + "name": "Outer", + "block": [ + {"name": "Provision secret", "ansible.builtin.shell": "curl || true"}, + ], + }, + ], + } + errors: list[str] = [] + _check_tasks(doc, tmp_path / "test.yml", errors, tmp_path) + assert any("|| true" in e for e in errors) + + def test_non_list_section_ignored(self, tmp_path: Path): + doc = {"tasks": "not a list"} + errors: list[str] = [] + _check_tasks(doc, tmp_path / "test.yml", errors, tmp_path) + assert errors == [] + + def test_non_dict_task_ignored(self, tmp_path: Path): + doc = {"tasks": ["just a string"]} + errors: list[str] = [] + _check_tasks(doc, tmp_path / "test.yml", errors, tmp_path) + assert errors == [] + + +class TestFindTaskFiles: + def test_single_file(self, tmp_path: Path): + p = tmp_path / "main.yml" + p.write_text("- name: test\n") + assert _find_task_files(p) == [p] + + def test_single_yaml_file(self, tmp_path: Path): + p = tmp_path / "main.yaml" + p.write_text("- name: test\n") + assert _find_task_files(p) == [p] + + def test_non_yaml_file_returns_empty(self, tmp_path: Path): + p = tmp_path / "main.txt" + p.write_text("hello\n") + assert _find_task_files(p) == [] + + def test_directory_finds_yaml_files(self, tmp_path: Path): + (tmp_path / "a.yml").write_text("- name: a\n") + (tmp_path / "sub").mkdir() + (tmp_path / "sub" / "b.yaml").write_text("- name: b\n") + (tmp_path / "ignore.txt").write_text("nope\n") + result = _find_task_files(tmp_path) + names = {f.name for f in result} + assert names == {"a.yml", "b.yaml"} + + def test_directory_skips_molecule(self, tmp_path: Path): + (tmp_path / "a.yml").write_text("- name: a\n") + (tmp_path / "molecule").mkdir() + (tmp_path / "molecule" / "scenario.yml").write_text("- name: mol\n") + result = _find_task_files(tmp_path) + assert all("molecule" not in f.parts for f in result) + + def test_nonexistent_path_returns_empty(self): + assert _find_task_files(Path("/nonexistent/path/xyz")) == [] + + +class TestMain: + def test_main_clean_file_exit_zero(self, tmp_path: Path): + p = tmp_path / "clean.yml" + p.write_text("- name: Safe task\n ansible.builtin.file:\n path: /opt/app\n state: directory\n") + result = CliRunner().invoke(main, ["--path", str(p)]) + assert result.exit_code == 0 + assert "OK" in result.output + + def test_main_violation_exit_one(self, tmp_path: Path): + p = tmp_path / "bad.yml" + p.write_text( + "- name: Provision OIDC\n" + " ansible.builtin.shell: curl -X POST https://api/app || true\n" + " failed_when: false\n" + ) + result = CliRunner().invoke(main, ["--path", str(p)]) + assert result.exit_code == 1 + assert "FAIL" in result.output + + def test_main_directory(self, tmp_path: Path): + (tmp_path / "clean.yml").write_text( + "- name: Safe task\n ansible.builtin.file:\n path: /opt\n state: directory\n" + ) + result = CliRunner().invoke(main, ["--path", str(tmp_path)]) + assert result.exit_code == 0 + + def test_main_default_dirs(self, tmp_path: Path, monkeypatch): + import devx.tools.check_ansible_patterns as mod + + (tmp_path / "clean.yml").write_text( + "- name: Safe task\n ansible.builtin.file:\n path: /opt\n state: directory\n" + ) + monkeypatch.setattr(mod, "DEFAULT_ANSIBLE_DIRS", [tmp_path]) + result = CliRunner().invoke(main, []) + assert result.exit_code == 0 + assert "OK" in result.output + + def test_main_custom_ansible_dirs(self, tmp_path: Path): + (tmp_path / "bad.yml").write_text( + "- name: Provision OIDC\n ansible.builtin.shell: curl -X POST https://api/app || true\n" + ) + result = CliRunner().invoke(main, ["--ansible-dir", str(tmp_path)]) + assert result.exit_code == 1 diff --git a/tests/unit/test_tools_check_jinja_expr.py b/tests/unit/test_tools_check_jinja_expr.py new file mode 100644 index 0000000..44a1546 --- /dev/null +++ b/tests/unit/test_tools_check_jinja_expr.py @@ -0,0 +1,264 @@ +"""Unit tests for devx.tools.check_jinja_expr. + +Verifies that the check correctly validates Jinja2 expressions, +catches reversed strftime filter arguments (the OBL-INFRA-508 bug), +and passes on valid expressions. +""" + +from __future__ import annotations + +from pathlib import Path +from unittest.mock import patch + +from click.testing import CliRunner + +from devx.tools.check_jinja_expr import ( + _check_file, + _default_ansible_dirs, + _extract_expressions, + _render_expression, + main, +) + + +def test_render_valid_expression(): + """Valid Jinja expression renders without error.""" + ok, _ = _render_expression("'%Y-%m-%dT%H:%M:%S+00:00' | strftime(1735689600)") + assert ok + + +def test_render_reversed_strftime_args(): + """Reversed strftime filter args are detected as an error.""" + ok, msg = _render_expression("(now().timestamp() | int + 3600) | strftime('%Y-%m-%dT%H:%M:%S+00:00')") + assert not ok + assert "reversed" in msg.lower() + + +def test_render_correct_strftime_args(): + """Correct strftime filter args pass.""" + ok, _ = _render_expression("'%Y-%m-%dT%H:%M:%S+00:00' | strftime((now().timestamp() | int) + 3600)") + assert ok + + +def test_render_unknown_filter(): + """Unknown filter is reported as an error.""" + ok, msg = _render_expression("'test' | nonexistent_filter") + assert not ok + assert "filter" in msg.lower() + + +def test_extract_skips_go_templates(): + """Go template syntax ({{.Field}}) is not extracted.""" + content = "cmd: docker inspect --format '{{.State.Running}}' container" + expressions = _extract_expressions(content) + assert len(expressions) == 0 + + +def test_extract_skips_single_char(): + """Single-character fragments are not extracted.""" + content = 'value: "{{ \' }}"' + expressions = _extract_expressions(content) + assert len(expressions) == 0 + + +def test_extract_skips_multiline(): + """Multi-line expressions are skipped.""" + content = 'value: "{{\n something\n}}"' + expressions = _extract_expressions(content) + assert len(expressions) == 0 + + +def test_extract_skips_unbalanced(): + """Expressions with unbalanced braces (from partial capture) are skipped.""" + content = "value: \"{{ default({'k': {}}, true) }}\"" + expressions = _extract_expressions(content) + # The regex captures {{ default({'k': {}} — unbalanced parens + # because the inner }} terminates the match early. + # All extracted expressions should have balanced braces. + for expr in expressions: + assert expr.count("{") == expr.count("}") + + +def test_extract_valid_expression(): + """Valid Jinja expressions are extracted.""" + content = "value: \"{{ my_var | default('x') }}\"" + expressions = _extract_expressions(content) + assert "my_var | default('x')" in expressions + + +def test_main_passes_on_clean_file(tmp_path: Path) -> None: + """A file with valid expressions passes.""" + test_file = tmp_path / "tasks.yml" + test_file.write_text("value: \"{{ my_var | default('x') }}\"\nother: \"{{ '%Y' | strftime(1735689600) }}\"\n") + runner = CliRunner() + result = runner.invoke(main, ["--path", str(test_file)]) + assert result.exit_code == 0 + + +def test_main_no_violations_empty_dir(tmp_path: Path) -> None: + """An empty directory passes.""" + runner = CliRunner() + result = runner.invoke(main, ["--path", str(tmp_path)]) + assert result.exit_code == 0 + + +def test_main_catches_reversed_strftime(tmp_path: Path) -> None: + """A file with reversed strftime args is flagged.""" + test_file = tmp_path / "test.yml" + test_file.write_text("value: \"{{ (now().timestamp() | int + 3600) | strftime('%Y-%m-%dT%H:%M:%S+00:00') }}\"\n") + runner = CliRunner() + result = runner.invoke(main, ["--path", str(test_file)]) + assert result.exit_code == 1 + assert "reversed" in result.output.lower() + + +def test_render_skips_undefined_var(): + """Undefined variables are skipped (MockDict returns mock for missing keys).""" + ok, _ = _render_expression("nonexistent_var_in_mock | upper") + assert ok + + +def test_render_skips_other_errors(): + """Non-filter errors from missing mocks are skipped.""" + ok, _ = _render_expression("some_undefined.attr.method()") + assert ok + + +def test_extract_skips_backtick(): + """Backtick fragments are skipped (caught by single-char check).""" + content = 'value: "{{ ` }}"' + expressions = _extract_expressions(content) + assert len(expressions) == 0 + + +def test_extract_skips_network_settings(): + """Expressions with .NetworkSettings. patterns are skipped.""" + content = 'value: "{{ foo.NetworkSettings.IPAddress }}"' + expressions = _extract_expressions(content) + assert len(expressions) == 0 + + +def test_extract_skips_unbalanced_parens(): + """Expressions with unbalanced parens are skipped.""" + content = 'value: "{{ foo(bar }}"' + expressions = _extract_expressions(content) + assert len(expressions) == 0 + + +def test_extract_skips_unbalanced_braces(): + """Expressions with unbalanced braces are skipped.""" + content = 'value: "{{ foo{bar }}"' + expressions = _extract_expressions(content) + assert len(expressions) == 0 + + +def test_extract_skips_unbalanced_brackets(): + """Expressions with unbalanced brackets are skipped.""" + content = 'value: "{{ foo[0 }}"' + expressions = _extract_expressions(content) + assert len(expressions) == 0 + + +def test_extract_skips_control_flow(): + """Control flow fragments starting with % are skipped.""" + content = 'value: "{{ % if x }}"' + expressions = _extract_expressions(content) + assert len(expressions) == 0 + + +def test_check_file_outside_repo(tmp_path: Path) -> None: + """Files outside REPO_ROOT are handled (no relative_to error).""" + test_file = tmp_path / "test.yml" + test_file.write_text("value: \"{{ (now().timestamp() | int + 3600) | strftime('%Y-%m-%dT%H:%M:%S+00:00') }}\"\n") + violations = _check_file(test_file, Path("/other/repo")) + assert len(violations) == 1 + assert "reversed" in violations[0].lower() + + +def test_render_mock_dict_missing_key(): + """MockDict returns a mock for missing keys (no UndefinedError).""" + ok, _ = _render_expression("undefined_var.some_attr | upper") + assert ok + + +def test_render_syntax_error(): + """Syntax errors are reported as failures.""" + ok, msg = _render_expression("{{ invalid syntax +") + assert not ok + assert "Syntax error" in msg + + +def test_render_unknown_filter_error(): + """Unknown filters are reported as failures (not skipped).""" + ok, msg = _render_expression("'test' | nonexistent_filter") + assert not ok + assert "filter" in msg.lower() + + +def test_render_generic_exception_skipped(): + """Non-filter exceptions from missing mocks are skipped.""" + # replace() with no args triggers TypeError (missing required args) + # which is not a filter-not-found or strftime error — should be skipped. + ok, msg = _render_expression("my_var | replace") + assert ok + assert "Skipped" in msg + + +def test_default_ansible_dirs(): + """_default_ansible_dirs returns playbooks and roles paths.""" + dirs = _default_ansible_dirs() + assert Path.cwd() / "ansible" / "playbooks" in dirs + assert Path.cwd() / "ansible" / "roles" in dirs + + +def test_main_default_dirs(tmp_path: Path) -> None: + """Running with no --path scans default dirs (uses small temp fixture).""" + (tmp_path / "playbooks").mkdir() + (tmp_path / "roles").mkdir() + (tmp_path / "playbooks" / "test.yml").write_text("value: \"{{ my_var | default('x') }}\"\n") + with patch( + "devx.tools.check_jinja_expr._default_ansible_dirs", + return_value=[tmp_path / "playbooks", tmp_path / "roles"], + ): + runner = CliRunner() + result = runner.invoke(main, []) + assert result.exit_code == 0 + + +def test_main_custom_ansible_dirs(tmp_path: Path) -> None: + """--ansible-dir option works.""" + (tmp_path / "test.yml").write_text("value: \"{{ my_var | default('x') }}\"\n") + runner = CliRunner() + result = runner.invoke(main, ["--ansible-dir", str(tmp_path)]) + assert result.exit_code == 0 + + +def test_extract_skips_println(): + """Expressions with 'println' (Go template) are skipped.""" + content = 'value: "{{ println something }}"' + expressions = _extract_expressions(content) + assert len(expressions) == 0 + + +def test_extract_skips_state_dot(): + """Expressions with .State. patterns are skipped.""" + content = 'value: "{{ foo.State.Running }}"' + expressions = _extract_expressions(content) + assert len(expressions) == 0 + + +def test_render_now_with_format(): + """now() with a format argument works.""" + ok, _ = _render_expression("now('%Y-%m-%d')") + assert ok + + +def test_find_yaml_files_skips_molecule(tmp_path: Path) -> None: + """Molecule directories are excluded from file search.""" + from devx.tools.check_jinja_expr import _find_yaml_files + + (tmp_path / "tasks.yml").write_text("value: test\n") + (tmp_path / "molecule").mkdir() + (tmp_path / "molecule" / "test.yml").write_text("value: test\n") + files = _find_yaml_files(tmp_path) + assert all("molecule" not in f.parts for f in files)