chore: update badge URLs to commit a506e113 [skip ci]

release: v0.50.0 [skip ci]
DEVX-154: feat: add 5 standalone lint scripts from infra
2026-08-09 22:09:27 +00:00 · 2026-08-09 22:08:35 +00:00 · 2026-08-09 22:07:38 +00:00 · 2026-08-09 09:39:51 +00:00
16 changed files with 2595 additions and 20 deletions
@@ -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)
@@ -142,6 +146,7 @@ src/devx/
    ├── discover_runners.py    # Dynamic Gitea runner discovery
    ├── distribute_molecule.py # Distribute molecule scenarios across runners (LPT scheduling, --roles-root for multi-role)
    ├── 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
 ```
@@ -2,6 +2,12 @@

 All notable changes to this project will be documented in this file.

+## [0.50.0] - 2026-08-09
+
+### Features
+
+- Add 5 standalone lint scripts from infra
+
 ## [0.49.0] - 2026-08-09

 ### Features
@@ -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/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/)
+[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/a506e1130cab09747bb881e0fe98093e6bcd8110/coverage.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
+[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/a506e1130cab09747bb881e0fe98093e6bcd8110/tests.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
+[![Docs](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/a506e1130cab09747bb881e0fe98093e6bcd8110/docs.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki)
+[![Code Quality](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/a506e1130cab09747bb881e0fe98093e6bcd8110/quality.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
+[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/a506e1130cab09747bb881e0fe98093e6bcd8110/version.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases)
+[![Python](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/a506e1130cab09747bb881e0fe98093e6bcd8110/python.svg)](https://www.python.org/downloads/)

 ## Why devx?

@@ -87,7 +87,7 @@ extra index and list devx in your dependencies:
 ```toml
 [project]
 dependencies = [
-    "devx>=0.49.0",
+    "devx>=0.50.0",
 ]

 [tool.pip]
@@ -101,8 +101,8 @@ pip install -e .
 ```

 > **Note:** If your project requires a specific devx version, pin it in
-> `dependencies` (for example, `"devx==0.49.0"`) or use a version constraint
-> (for example, `"devx>=0.49.0,<0.50"`).
+> `dependencies` (for example, `"devx==0.50.0"`) or use a version constraint
+> (for example, `"devx>=0.50.0,<0.51"`).

 ### Optional extras

@@ -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/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/)
+[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/a506e1130cab09747bb881e0fe98093e6bcd8110/coverage.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
+[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/a506e1130cab09747bb881e0fe98093e6bcd8110/tests.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
+[![Docs](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/a506e1130cab09747bb881e0fe98093e6bcd8110/docs.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki)
+[![Code Quality](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/a506e1130cab09747bb881e0fe98093e6bcd8110/quality.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
+[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/a506e1130cab09747bb881e0fe98093e6bcd8110/version.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases)
+[![Python](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/a506e1130cab09747bb881e0fe98093e6bcd8110/python.svg)](https://www.python.org/downloads/)

 ## Overview

@@ -74,14 +74,14 @@ Add devx to your `pyproject.toml` dependencies and configure the registry:
 ```toml
 [project]
 dependencies = [
-    "devx>=0.49.0",
+    "devx>=0.50.0",
 ]

 [tool.pip]
 extra-index-url = "https://git.oblachno.oblachno.fyi/api/packages/oblachno-oss/pypi/simple"
 ```

-Pin a specific version if needed: `"devx==0.49.0"` or `"devx>=0.49.0,<0.50"`.
+Pin a specific version if needed: `"devx==0.50.0"` or `"devx>=0.50.0,<0.51"`.

 ### Optional extras

@@ -48,12 +48,12 @@ Add devx to your `pyproject.toml`:
 ```toml
 [project]
 dependencies = [
-    "devx>=0.49.0",
+    "devx>=0.50.0",
 ]

 [project.optional-dependencies]
 dev = [
-    "devx>=0.49.0",
+    "devx>=0.50.0",
 ]
 ```

@@ -1,3 +1,3 @@
 """devx — reusable development and CI/CD tools for oblachno-oss projects."""

-__version__ = "0.49.0"
+__version__ = "0.50.0"
@@ -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/<role>/  →  molecule-<role>
+
+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/<role> (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()
@@ -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", "<unnamed>")
+        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()
@@ -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()
@@ -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", "<unnamed>")
+
+    # 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()
@@ -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={{ <expr> }}'`")
+        sys.exit(1)
+    else:
+        click.echo("[check-jinja-expr] OK: all Jinja expressions render correctly.")
+
+
+if __name__ == "__main__":  # pragma: no cover
+    main()
@@ -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/<role>/ 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")
@@ -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
@@ -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
@@ -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
@@ -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)
Author	SHA1	Message	Date
gitea-actions-bot	0341ee74c9	chore: update badge URLs to commit `a506e113` [skip ci]	2026-08-09 22:09:27 +00:00
devx-ci-bot	6c8b02909f	release: v0.50.0 [skip ci]	2026-08-09 22:08:35 +00:00
emilandemo	eeb291564d	DEVX-154: feat: add 5 standalone lint scripts from infra Post-merge / detect-and-configure (push) Successful in 23s Details Post-merge / release-and-maintain (push) Successful in 1m21s Details Co-authored-by: emil User <emil.simeonov@tutanota.com>	2026-08-09 22:07:38 +00:00
gitea-actions-bot	16fb319a47	chore: update badge URLs to commit `6a22c4cb` [skip ci]	2026-08-09 09:39:51 +00:00