oblachno-oss/devx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

DEVX-155: refactor: extract wait_for_checks, consolidate ansible_checks, deprecate ci/discover_runners

Browse Source
This commit is contained in:
emil User
2026-08-12 17:37:49 +00:00
parent b6f93cc2a8
commit 48596627d5
27 changed files with 2251 additions and 1197 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/devx/ci/discover_runners.py
+28 -173
View File
@@ -1,19 +1,18 @@
#!/usr/bin/env python3
"""Discover available Gitea Actions runners for dynamic job distribution.
Queries the Gitea API for registered runners at three levels:
1. Repository level: GET /repos/{owner}/{repo}/actions/runners
2. Organization level: GET /orgs/{org}/actions/runners
3. Instance (admin) level: GET /admin/actions/runners
.. deprecated:: Phase 1c
Use :mod:`devx.molecule.discover_runners` instead. This module is a
thin wrapper that re-exports the canonical implementation from
:mod:`devx.molecule.discover_runners` for backward compatibility
with existing workflow references and Makefile targets.
Falls back to the ``MOLECULE_RUNNERS`` repo variable or environment
variable, then to ``DEFAULT_MAX_RUNNERS`` (3).
Outputs:
- ``--count``: prints the number of available runners
- ``--indices``: prints a JSON array [0, 1, ..., N-1] for use as a
dynamic matrix in Gitea Actions
- (default): prints both as ``count=N`` and ``indices=[0,1,...]``
The canonical implementation lives in
:mod:`devx.molecule.discover_runners` because runner discovery is
primarily used by the molecule test distribution pipeline. CI
workflows that still reference ``python -m devx.ci.discover_runners``
will continue to work via this wrapper, but new code should import
from :mod:`devx.molecule.discover_runners` directly.
Usage:
python3 -m devx.ci.discover_runners --owner oblachno-oss --repo devx
@@ -23,172 +22,28 @@ Usage:
from __future__ import annotations
import json
import os
import sys
import warnings
import click
import requests
from devx.config import GITEA_API_URL, REPO_NAME, REPO_OWNER
from devx.i18n import _
from devx.tokens import get_ci_token
DEFAULT_MAX_RUNNERS = 3
def query_runners(api_url: str, token: str, owner: str, repo: str) -> int:
"""Query the Gitea API for registered runners at all levels.
Returns the total count of active runners. If the API call fails
(e.g., no admin access for instance-level runners), falls back to
what we can see. Fallbacks are logged to stderr for debugging.
"""
headers = {"Authorization": f"token {token}"}
total = 0
# 1. Repository-level runners
try:
r = requests.get(
f"{api_url}/repos/{owner}/{repo}/actions/runners",
headers=headers,
timeout=10,
)
if r.status_code == 200:
data = r.json()
total += data.get("total_count", 0)
else:
click.echo(_("Warning: repo-level runners query returned HTTP {status}", status=r.status_code), err=True)
except (requests.RequestException, ValueError) as e:
click.echo(_("Warning: repo-level runners query failed: {error}", error=e), err=True)
# 2. Organization-level runners
try:
r = requests.get(
f"{api_url}/orgs/{owner}/actions/runners",
headers=headers,
timeout=10,
)
if r.status_code == 200:
data = r.json()
total += data.get("total_count", 0)
else:
click.echo(_("Warning: org-level runners query returned HTTP {status}", status=r.status_code), err=True)
except (requests.RequestException, ValueError) as e:
click.echo(_("Warning: org-level runners query failed: {error}", error=e), err=True)
# 3. Instance-level runners (requires admin scope)
try:
r = requests.get(
f"{api_url}/admin/actions/runners",
headers=headers,
timeout=10,
)
if r.status_code == 200:
data = r.json()
total += data.get("total_count", 0)
elif r.status_code != 403: # 403 is expected without admin scope
click.echo(
_("Warning: instance-level runners query returned HTTP {status}", status=r.status_code),
err=True,
)
except (requests.RequestException, ValueError) as e:
click.echo(_("Warning: instance-level runners query failed: {error}", error=e), err=True)
return total
def get_runner_count(api_url: str, token: str | None, owner: str, repo: str) -> int:
"""Determine the number of available runners.
Tries the Gitea API first, then falls back to env vars, then default.
"""
# Try API query if we have a token
if token:
api_count = query_runners(api_url, token, owner, repo)
if api_count > 0:
return api_count
# Fall back to MOLECULE_RUNNERS env var (set by CI from repo variable)
env_count = os.environ.get("MOLECULE_RUNNERS")
if env_count:
try:
count = int(env_count)
if count > 0:
return count
except ValueError:
pass
# Fall back to default
return DEFAULT_MAX_RUNNERS
def generate_indices(count: int) -> list[str]:
"""Generate a list of runner indices ["1", "2", ..., "N"].
Uses 1-based string indices because Gitea Actions renders
integer 0 and string "0" as empty in ${{ matrix.runner-index }}
expressions, causing --runner-index to be passed without a value.
The distribute_molecule.py script converts these back to 0-based
internally.
"""
return [str(i + 1) for i in range(count)]
@click.command()
@click.option("--owner", default=None, help="Repository owner (for API query).")
@click.option("--repo", default=None, help="Repository name (for API query).")
@click.option("--count", "output_count", is_flag=True, help="Output only the count.")
@click.option("--indices", "output_indices", is_flag=True, help="Output only the JSON indices array.")
@click.option(
"--github-output",
"github_output",
is_flag=True,
default=False,
help="Write results to $GITHUB_OUTPUT file (for CI workflow steps).",
from devx.molecule.discover_runners import ( # noqa: F401 — re-exported for backward compat
DEFAULT_MAX_RUNNERS,
generate_indices,
get_runner_count,
main,
query_runners,
)
def main(
owner: str | None,
repo: str | None,
output_count: bool,
output_indices: bool,
github_output: bool,
) -> None:
try:
token = get_ci_token()
except click.ClickException:
token = None
if owner is None:
owner = os.environ.get("DEVX_REPO_OWNER", "") or REPO_OWNER
if repo is None:
repo = os.environ.get("DEVX_REPO_NAME", "") or REPO_NAME
_DEPRECATION_MSG = (
"devx.ci.discover_runners is deprecated; use devx.molecule.discover_runners instead. "
"This wrapper will be removed in a future release."
)
count = get_runner_count(GITEA_API_URL, token, owner, repo)
indices = generate_indices(count)
if github_output:
gh_output = os.environ.get("GITHUB_OUTPUT")
if not gh_output:
raise click.ClickException("GITHUB_OUTPUT environment variable is not set")
with open(gh_output, "a", encoding="utf-8") as f: # noqa: PTH123
f.write(f"runner-count={count}\n")
f.write(f"runner-indices={json.dumps(indices)}\n")
click.echo(_("Runner count: {count}", count=count))
click.echo(_("Runner indices: {indices}", indices=indices))
return
if output_count:
click.echo(str(count))
return
if output_indices:
click.echo(json.dumps(indices))
return
# Default: output both as key=value pairs for CI consumption
click.echo(_("count={count}", count=count))
click.echo(_("indices={indices}", indices=json.dumps(indices)))
def _emit_deprecation_warning() -> None:
"""Emit a DeprecationWarning when this module is imported for CLI use."""
warnings.warn(_DEPRECATION_MSG, DeprecationWarning, stacklevel=2)
if __name__ == "__main__": # pragma: no cover
main()
_emit_deprecation_warning()
sys.exit(main())
src/devx/ci/wait_for_checks.py
+209
View File
@@ -0,0 +1,209 @@
#!/usr/bin/env python3
"""Wait for Gitea Actions jobs to complete.
Polls the Gitea API for job completion status. Used by auto-merge
jobs that need to wait for molecule-tests or other parallel jobs
before proceeding.
Exits:
0 — all matching jobs completed successfully
1 — one or more matching jobs failed
2 — timeout reached before all jobs completed
3 — API error or job not found
Usage:
python3 -m devx.ci.wait_for_checks \\
--job-name molecule-tests \\
--repo oblachno-oss/grm \\
--timeout 1200 \\
--poll-interval 10
"""
from __future__ import annotations
import os
import sys
import time
import click
import requests
from devx.config import GITEA_API_URL
from devx.i18n import _
from devx.tokens import get_ci_token
def query_job_status(api_url: str, token: str, repo: str, job_name_prefix: str) -> list[dict]:
"""Query the Gitea API for the status of jobs matching *job_name_prefix*.
Fetches the most recent pull_request runs (up to 3) and inspects
their jobs. Returns a list of ``{"name": str, "status": str,
"conclusion": str | None}`` dicts for jobs whose name starts with
*job_name_prefix*. On API errors, logs a warning to stderr and
returns an empty list — callers treat this as "no information yet"
and retry on the next poll.
"""
headers = {"Authorization": f"token {token}"}
matches: list[dict] = []
try:
r = requests.get(
f"{api_url}/repos/{repo}/actions/runs",
headers=headers,
params={"limit": 5, "event": "pull_request"},
timeout=10,
)
if r.status_code != 200:
click.echo(
_("Warning: actions runs query returned HTTP {status}", status=r.status_code),
err=True,
)
return []
runs = r.json()
if isinstance(runs, dict):
runs = runs.get("runs", [])
except (requests.RequestException, ValueError) as e:
click.echo(_("Warning: actions runs query failed: {error}", error=e), err=True)
return []
for run in runs[:3]:
run_id = run.get("id")
if run_id is None:
continue
try:
jr = requests.get(
f"{api_url}/repos/{repo}/actions/runs/{run_id}/jobs",
headers=headers,
timeout=10,
)
if jr.status_code != 200:
click.echo(
_(
"Warning: jobs query for run {run_id} returned HTTP {status}",
run_id=run_id,
status=jr.status_code,
),
err=True,
)
continue
jobs = jr.json()
if isinstance(jobs, dict):
jobs = jobs.get("jobs", [])
except (requests.RequestException, ValueError) as e:
click.echo(
_("Warning: jobs query for run {run_id} failed: {error}", run_id=run_id, error=e),
err=True,
)
continue
for job in jobs:
name = job.get("name", "")
if name.startswith(job_name_prefix):
matches.append(
{
"name": name,
"status": job.get("status", "unknown"),
"conclusion": job.get("conclusion"),
}
)
return matches
def poll_until_complete(
api_url: str,
token: str,
repo: str,
job_name: str,
timeout: int,
interval: int,
require_success: bool = True,
) -> int:
"""Poll *query_job_status* until all matching jobs complete or *timeout*.
Returns:
0 — all matching jobs completed successfully (or any completed, when
*require_success* is False)
1 — at least one matching job completed with a non-success conclusion
(only when *require_success* is True)
2 — *timeout* reached before all matching jobs completed
3 — no matching jobs found at all within *timeout*
"""
deadline = time.monotonic() + timeout
found_any = False
while time.monotonic() < deadline:
jobs = query_job_status(api_url, token, repo, job_name)
if jobs:
found_any = True
all_completed = all(j["status"] == "completed" for j in jobs)
if all_completed:
if require_success and any(j["conclusion"] != "success" for j in jobs):
click.echo(
_("Job(s) completed with non-success conclusion: {jobs}", jobs=jobs),
err=True,
)
return 1
click.echo(_("All matching jobs completed successfully: {jobs}", jobs=jobs))
return 0
# Not all completed (or no jobs yet) — sleep and retry.
time.sleep(min(interval, max(0, deadline - time.monotonic())))
if not found_any:
click.echo(_("No matching jobs found for prefix '{prefix}' within timeout.", prefix=job_name), err=True)
return 3
click.echo(_("Timeout reached waiting for jobs matching '{prefix}'.", prefix=job_name), err=True)
return 2
@click.command()
@click.option("--job-name", required=True, help="Job name prefix to match (e.g. 'molecule-tests').")
@click.option(
"--repo",
default=None,
help="Repository as owner/name (default: $GITHUB_REPOSITORY env var).",
)
@click.option("--timeout", type=int, default=1200, help="Max seconds to wait (default: 1200 = 20 min).")
@click.option("--poll-interval", "interval", type=int, default=10, help="Seconds between polls (default: 10).")
@click.option(
"--require-success/--no-require-success",
default=True,
help="Exit 1 if a matched job failed (default: yes).",
)
def main(job_name: str, repo: str | None, timeout: int, interval: int, require_success: bool) -> None:
"""Wait for Gitea Actions jobs matching --job-name to complete."""
if repo is None:
repo = os.environ.get("GITHUB_REPOSITORY", "")
if not repo or "/" not in repo:
raise click.ClickException(_("--repo is required (or set GITHUB_REPOSITORY=owner/name)"))
if timeout <= 0:
raise click.ClickException(_("--timeout must be positive"))
if interval <= 0:
raise click.ClickException(_("--poll-interval must be positive"))
try:
token = get_ci_token()
except click.ClickException as e:
click.echo(str(e), err=True)
sys.exit(3)
click.echo(
_(
"Waiting for jobs matching '{prefix}' in {repo} (timeout={timeout}s, interval={interval}s)",
prefix=job_name,
repo=repo,
timeout=timeout,
interval=interval,
)
)
code = poll_until_complete(
GITEA_API_URL,
token,
repo,
job_name,
timeout,
interval,
require_success=require_success,
)
sys.exit(code)
if __name__ == "__main__": # pragma: no cover
main()
src/devx/cli.py
+7
View File
@@ -158,6 +158,13 @@ def ci_validate_commit_msg(args: tuple[str, ...]) -> None:
_run_module("devx.ci.validate_commit_msg", list(args))
@ci.command("wait-for-checks")
@click.argument("args", nargs=-1)
def ci_wait_for_checks(args: tuple[str, ...]) -> None:
"""Wait for Gitea Actions jobs to complete (polls API)."""
_run_module("devx.ci.wait_for_checks", list(args))
@ci.command("distribute-files")
@click.argument("args", nargs=-1)
def ci_distribute_files(args: tuple[str, ...]) -> None:
src/devx/molecule/discover_runners.py
+21 -11
View File
@@ -30,6 +30,7 @@ import click
import requests
from devx.config import GITEA_API_URL, REPO_NAME, REPO_OWNER
from devx.i18n import _
from devx.tokens import get_ci_token
DEFAULT_MAX_RUNNERS = 3
@@ -40,7 +41,7 @@ def query_runners(api_url: str, token: str, owner: str, repo: str) -> int:
Returns the total count of active runners. If the API call fails
(e.g., no admin access for instance-level runners), falls back to
what we can see.
what we can see. Fallbacks are logged to stderr for debugging.
"""
headers = {"Authorization": f"token {token}"}
total = 0
@@ -55,8 +56,10 @@ def query_runners(api_url: str, token: str, owner: str, repo: str) -> int:
if r.status_code == 200:
data = r.json()
total += data.get("total_count", 0)
except (requests.RequestException, ValueError):
pass
else:
click.echo(_("Warning: repo-level runners query returned HTTP {status}", status=r.status_code), err=True)
except (requests.RequestException, ValueError) as e:
click.echo(_("Warning: repo-level runners query failed: {error}", error=e), err=True)
# 2. Organization-level runners
try:
@@ -68,8 +71,10 @@ def query_runners(api_url: str, token: str, owner: str, repo: str) -> int:
if r.status_code == 200:
data = r.json()
total += data.get("total_count", 0)
except (requests.RequestException, ValueError):
pass
else:
click.echo(_("Warning: org-level runners query returned HTTP {status}", status=r.status_code), err=True)
except (requests.RequestException, ValueError) as e:
click.echo(_("Warning: org-level runners query failed: {error}", error=e), err=True)
# 3. Instance-level runners (requires admin scope)
try:
@@ -81,8 +86,13 @@ def query_runners(api_url: str, token: str, owner: str, repo: str) -> int:
if r.status_code == 200:
data = r.json()
total += data.get("total_count", 0)
except (requests.RequestException, ValueError):
pass
elif r.status_code != 403: # 403 is expected without admin scope
click.echo(
_("Warning: instance-level runners query returned HTTP {status}", status=r.status_code),
err=True,
)
except (requests.RequestException, ValueError) as e:
click.echo(_("Warning: instance-level runners query failed: {error}", error=e), err=True)
return total
@@ -163,8 +173,8 @@ def main(
with open(gh_output, "a", encoding="utf-8") as f: # noqa: PTH123
f.write(f"runner-count={count}\n")
f.write(f"runner-indices={json.dumps(indices)}\n")
click.echo(f"Runner count: {count}")
click.echo(f"Runner indices: {indices}")
click.echo(_("Runner count: {count}", count=count))
click.echo(_("Runner indices: {indices}", indices=indices))
return
if output_count:
@@ -176,8 +186,8 @@ def main(
return
# Default: output both as key=value pairs for CI consumption
click.echo(f"count={count}")
click.echo(f"indices={json.dumps(indices)}")
click.echo(_("count={count}", count=count))
click.echo(_("indices={indices}", indices=json.dumps(indices)))
if __name__ == "__main__": # pragma: no cover
src/devx/tools/ansible_checks/__init__.py
+34
View File
@@ -0,0 +1,34 @@
"""Ansible check tools — composable validators for Ansible playbooks and roles.
Each check module exports a ``check_*`` function that returns a list of
violation strings. The shared utilities in :mod:`devx.tools.ansible_checks._shared`
handle file discovery, YAML parsing, and violation reporting.
The old entry points (``devx.tools.check_ansible_*``, ``devx.tools.check_jinja_expr``)
remain as thin wrappers for backward compatibility with existing Makefile
targets and workflow references.
"""
from devx.tools.ansible_checks._shared import (
DEFAULT_ANSIBLE_DIRS,
AnsibleFileFinder,
AnsibleYAMLParser,
ViolationReporter,
)
from devx.tools.ansible_checks.jinja_expr import check_jinja_expr
from devx.tools.ansible_checks.no_log import check_no_log
from devx.tools.ansible_checks.no_state_absent_on_db import check_no_state_absent_on_db
from devx.tools.ansible_checks.patterns import check_patterns
from devx.tools.ansible_checks.set_fact_to_json import check_set_fact_to_json
__all__ = [
"DEFAULT_ANSIBLE_DIRS",
"AnsibleFileFinder",
"AnsibleYAMLParser",
"ViolationReporter",
"check_jinja_expr",
"check_no_log",
"check_no_state_absent_on_db",
"check_patterns",
"check_set_fact_to_json",
]
src/devx/tools/ansible_checks/_shared.py
+178
View File
@@ -0,0 +1,178 @@
"""Shared utilities for Ansible check tools.
Provides composable helpers for file discovery, YAML parsing, and
violation reporting used by the modules in :mod:`devx.tools.ansible_checks`.
Composition over inheritance: each check module picks the helpers it
needs. Tools that don't parse YAML (e.g. line-based scanners) can skip
:class:`AnsibleYAMLParser` entirely.
"""
from __future__ import annotations
import sys
from collections.abc import Iterator
from pathlib import Path
from typing import Final
import click
import yaml
from devx.i18n import _
#: Default Ansible directories scanned by checks that accept ``--ansible-dir``.
#: Immutable tuple (not a list) to avoid module-level mutable globals.
DEFAULT_ANSIBLE_DIRS: Final[tuple[str, ...]] = ("ansible/roles", "ansible/playbooks")
class AnsibleFileFinder:
"""File discovery helpers for Ansible YAML files."""
@staticmethod
def find_task_files(base: Path, skip_molecule: bool = True) -> list[Path]:
"""Find all YAML files under *base*, recursively.
If *base* is a single YAML file, returns ``[base]``. If *base* is
not a file or directory, returns ``[]``. When *skip_molecule* is
True, files with ``molecule`` in their path parts are excluded.
"""
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 skip_molecule and "molecule" in f.parts:
continue
files.append(f)
return files
@staticmethod
def find_yaml_files(base: Path, skip_molecule: bool = True) -> list[Path]:
"""Find YAML files under *base* using ``glob`` (non-recursive rglob).
Unlike :meth:`find_task_files`, this uses ``base.glob("**/*.yml")``
and does not check the suffix when *base* is a single file (any
file is accepted). Used by the Jinja expression checker which
scans all YAML files including defaults/handlers.
"""
if base.is_file():
return [base]
files: list[Path] = []
for pattern in ("**/*.yml", "**/*.yaml"):
files.extend(base.glob(pattern))
if skip_molecule:
return [f for f in files if "molecule" not in f.parts]
return files
@staticmethod
def find_task_and_playbook_files(base: Path, skip_molecule: bool = True) -> list[Path]:
"""Find task files (``tasks/*.yml``) and playbook files (``playbooks/*.yml``).
Used by the no_log checker which scans role task files and
top-level playbook files. When *skip_molecule* is True, molecule
scenario files are excluded.
"""
task_files = list(base.rglob("tasks/*.yml")) + list(base.rglob("tasks/*.yaml"))
task_files += list(base.glob("playbooks/*.yml")) + list(base.glob("playbooks/*.yaml"))
if skip_molecule:
task_files = [f for f in task_files if "molecule" not in f.parts]
return sorted(task_files)
class AnsibleYAMLParser:
"""YAML parsing helpers for Ansible files."""
@staticmethod
def parse_file(content: str) -> list[dict]:
"""Parse multi-document YAML from *content*.
Returns a list of non-None documents. On ``YAMLError`` or
``OSError``, returns an empty list (the caller skips the file).
"""
try:
docs = list(yaml.safe_load_all(content))
except (yaml.YAMLError, OSError):
return []
return [d for d in docs if d]
@staticmethod
def iter_tasks(doc: dict | list) -> Iterator[tuple[dict, int]]:
"""Yield ``(task_dict, line_number)`` tuples from a YAML document.
Handles:
- Bare task lists (role tasks files): ``[task1, task2, ...]``
- Play dicts with ``hosts`` key: iterates ``tasks``,
``pre_tasks``, ``post_tasks``, ``handlers`` sections
- Nested ``block`` tasks
The line number is the 1-based index within the task section
(not the file line number — callers use it for display only).
"""
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")):
yield from AnsibleYAMLParser._iter_play_sections(item)
else:
yield item, i + 1
block = item.get("block")
if isinstance(block, list):
for j, bt in enumerate(block):
if isinstance(bt, dict):
yield bt, i + j + 1
elif isinstance(doc, dict):
yield from AnsibleYAMLParser._iter_play_sections(doc)
@staticmethod
def _iter_play_sections(doc: dict) -> Iterator[tuple[dict, int]]:
"""Yield tasks from play sections (tasks, pre_tasks, post_tasks, handlers)."""
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):
yield task, i + 1
block = task.get("block")
if isinstance(block, list):
for j, bt in enumerate(block):
if isinstance(bt, dict):
yield bt, i + j + 1
class ViolationReporter:
"""Standardized violation formatting and reporting."""
@staticmethod
def format_violation(
filepath: Path,
repo_root: Path,
line_num: int | None,
message: str,
) -> str:
"""Format a violation as ``"{relative_path}:{line_num} — message"``.
Falls back to the full path if *filepath* is not relative to
*repo_root*. When *line_num* is None, omits the line number.
"""
try:
display_path = filepath.relative_to(repo_root)
except ValueError:
display_path = filepath
if line_num is not None:
return f"{display_path}:{line_num} — {message}"
return f"{display_path} — {message}"
@staticmethod
def report(violations: list[str], tool_name: str) -> None:
"""Print violations and exit with the appropriate code.
Prints ``[{tool_name}] FAIL`` or ``[{tool_name}] OK`` and exits
1 if violations are non-empty, 0 otherwise.
"""
if violations:
click.echo(_("[{tool}] FAIL: {count} violation(s) found.", tool=tool_name, count=len(violations)))
for v in violations:
click.echo(f" - {v}")
sys.exit(1)
click.echo(_("[{tool}] OK: no violations found.", tool=tool_name))
src/devx/tools/ansible_checks/jinja_expr.py
+226
View File
@@ -0,0 +1,226 @@
"""Validate Jinja2 expressions in Ansible files by rendering them.
Extracted from :mod:`devx.tools.check_jinja_expr` as part of the
Ansible check tool consolidation. The old module remains as a thin
wrapper for backward compatibility.
"""
from __future__ import annotations
import re
from pathlib import Path
from jinja2 import Environment
from jinja2.exceptions import TemplateSyntaxError, UndefinedError
from devx.tools.ansible_checks._shared import AnsibleFileFinder, ViolationReporter
REPO_ROOT = Path.cwd()
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
}
EXPR_PATTERN = re.compile(r"\{\{(.*?)\}\}", re.DOTALL)
def _default_ansible_dirs() -> list[Path]:
"""Return the default directories to scan for Ansible files."""
return [
REPO_ROOT / "ansible" / "playbooks",
REPO_ROOT / "ansible" / "roles",
]
def _find_yaml_files(path: Path) -> list[Path]:
"""Find Ansible YAML files (tasks, playbooks, handlers) in a path."""
return AnsibleFileFinder.find_yaml_files(path, skip_molecule=True)
def _extract_expressions(content: str) -> list[str]:
"""Extract Jinja expressions from file content."""
expressions = []
for match in EXPR_PATTERN.finditer(content):
raw = match.group(1)
if "\n" in raw:
continue
expr = raw.strip()
if not expr or expr.startswith("%") or len(expr) <= 1:
continue
if expr.startswith(".") or "println" in expr:
continue
if ".State." in expr or ".NetworkSettings." in expr:
continue
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
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:
return True, f"Skipped (undefined: {e})"
except Exception as e:
error_msg = str(e)
if "Invalid value for epoch" in error_msg:
return False, f"strftime filter argument error: {error_msg}"
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:
display_path = ViolationReporter.format_violation(filepath, repo_root, None, "")
display_path = display_path.removesuffix(" — ")
violations.append(f"{display_path}: `{{{{ {expr} }}}}` — {msg}")
return violations
def check_jinja_expr(path: Path | None, ansible_dirs: list[Path] | None = None) -> list[str]:
"""Validate Jinja2 expressions in Ansible files.
Args:
path: Specific file or directory to check. If None, *ansible_dirs*
is used.
ansible_dirs: Directories to scan when *path* is None.
Returns:
List of violation messages (empty if all renderable expressions pass).
"""
if path:
files = _find_yaml_files(path)
else:
files: list[Path] = []
for d in ansible_dirs or _default_ansible_dirs():
files.extend(_find_yaml_files(d))
all_violations: list[str] = []
for f in files:
all_violations.extend(_check_file(f, REPO_ROOT))
return all_violations
src/devx/tools/ansible_checks/no_log.py
+128
View File
@@ -0,0 +1,128 @@
"""Check Ansible tasks for missing no_log on secret-handling tasks.
Extracted from :mod:`devx.tools.check_ansible_no_log` as part of the
Ansible check tool consolidation. The old module remains as a thin
wrapper for backward compatibility.
"""
from __future__ import annotations
import re
from pathlib import Path
from devx.tools.ansible_checks._shared import AnsibleFileFinder, AnsibleYAMLParser
# Patterns that indicate a task is handling secrets.
SECRET_PATTERNS = [
re.compile(r"\{\{[^}]*_secrets\.", re.IGNORECASE),
re.compile(r"\{\{[^}]*password", re.IGNORECASE),
re.compile(r"\{\{[^}]*_secret\b", re.IGNORECASE),
re.compile(r"\{\{[^}]*api_key", re.IGNORECASE),
re.compile(r"\{\{[^}]*(?:vault_token|auth_token|access_token|bot_token)", re.IGNORECASE),
]
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",
}
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)
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."""
violations: list[str] = []
if _has_no_log(task):
return violations
has_secrets = False
for key, value in task.items():
if key in NON_VALUE_KEYS:
continue
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_no_log(path: Path, ansible_dirs: list[Path] | None = None) -> list[str]:
"""Check all Ansible task files for missing no_log on secret-handling tasks.
Args:
path: The base directory to scan (or a specific file).
ansible_dirs: Unused — kept for API symmetry with other checks.
The no_log checker scans *path* directly.
Returns:
List of violation messages (empty if all OK).
"""
all_violations: list[str] = []
task_files = AnsibleFileFinder.find_task_and_playbook_files(path)
for task_file in task_files:
try:
content = task_file.read_text(encoding="utf-8")
except (OSError, UnicodeDecodeError):
continue
docs = AnsibleYAMLParser.parse_file(content)
for doc in docs:
for task, task_num in AnsibleYAMLParser.iter_tasks(doc):
all_violations.extend(_check_task(task, task_file, task_num))
return all_violations
# Backward-compat alias for the old public function name.
check_directory = check_no_log
src/devx/tools/ansible_checks/no_state_absent_on_db.py
+100
View File
@@ -0,0 +1,100 @@
"""Check Ansible tasks for ``state: absent`` on database data directories.
Extracted from :mod:`devx.tools.check_ansible_no_state_absent_on_db` as
part of the Ansible check tool consolidation. The old module remains as
a thin wrapper for backward compatibility.
"""
from __future__ import annotations
import re
from pathlib import Path
from devx.tools.ansible_checks._shared import AnsibleFileFinder, ViolationReporter
REPO_ROOT = Path.cwd()
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_PATTERNS = (
re.compile(r"state:\s*absent", re.IGNORECASE),
re.compile(r"rm\s+-rf.*\bdb\b", re.IGNORECASE),
)
ALLOWED_CONTEXT_KEYWORDS = (
"upgrade-postgres",
"PG_VERSION",
"pg_version",
)
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."""
return AnsibleFileFinder.find_task_files(base, skip_molecule=True)
def _check_file(filepath: Path, repo_root: Path) -> list[str]:
"""Check a YAML file for state: absent on DB data directory paths."""
try:
content = filepath.read_text(encoding="utf-8")
except (OSError, UnicodeDecodeError):
return []
if not any(p.search(content) for p in DB_PATH_PATTERNS):
return []
display_path = ViolationReporter.format_violation(filepath, repo_root, None, "")
display_path = display_path.removesuffix(" — ")
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
context_start = max(0, i - 5)
context_end = min(len(lines), i + 6)
context = "\n".join(lines[context_start:context_end])
if any(kw in context for kw in ALLOWED_CONTEXT_KEYWORDS):
continue
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
def check_no_state_absent_on_db(path: Path | None, ansible_dirs: list[Path] | None = None) -> list[str]:
"""Check that no Ansible task uses state: absent on a DB data directory.
Args:
path: Specific file or directory to check. If None, *ansible_dirs*
is used.
ansible_dirs: Directories to scan when *path* is None.
Returns:
List of violation messages (empty if clean).
"""
if path:
files = _find_task_files(path)
else:
files: list[Path] = []
for d in ansible_dirs or []:
files.extend(_find_task_files(d))
all_violations: list[str] = []
for f in files:
all_violations.extend(_check_file(f, REPO_ROOT))
return all_violations
src/devx/tools/ansible_checks/patterns.py
+226
View File
@@ -0,0 +1,226 @@
"""Check Ansible tasks for dangerous patterns that mask failures.
Extracted from :mod:`devx.tools.check_ansible_patterns` as part of the
Ansible check tool consolidation. The old module remains as a thin
wrapper for backward compatibility.
"""
from __future__ import annotations
import re
from pathlib import Path
from devx.tools.ansible_checks._shared import AnsibleFileFinder, AnsibleYAMLParser
REPO_ROOT = Path.cwd()
# 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_COMMAND_PREFIXES = (
"docker rm",
"docker stop",
"docker rmi",
"docker network rm",
"docker volume rm",
"pkill",
"kill",
"journalctl --vacuum",
"apt-get clean",
"apt-get autoremove",
"docker image prune",
"docker container prune",
"docker volume prune",
"docker builder prune",
"find / -name",
"chmod",
"rm -f",
"docker network connect",
"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",
)
CRITICAL_TASK_KEYWORDS = (
"password",
"secret",
"provision",
"oidc",
)
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."""
name_lower = task_name.lower()
if any(re.search(kw, name_lower) for kw in LEGITIMATE_TASK_NAME_KEYWORDS):
return True
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."""
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>")
if ALLOW_MARKER in task_name:
return violations
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."
)
failed_when = task.get("failed_when")
if failed_when is False:
name_lower = task_name.lower()
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 []
if not (
OR_TRUE_PATTERN.search(content) or "failed_when: false" in content or REDIRECT_DEVNULL_PATTERN.search(content)
):
return []
has_allow_marker = ALLOW_MARKER in content
docs = AnsibleYAMLParser.parse_file(content)
violations: list[str] = []
for doc in docs:
for task, task_num in AnsibleYAMLParser.iter_tasks(doc):
violations.extend(_check_task(task, filepath, task_num, repo_root))
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 task, task_num in AnsibleYAMLParser._iter_play_sections(doc):
errors.extend(_check_task(task, filepath, task_num, repo_root))
def _find_task_files(base: Path) -> list[Path]:
"""Find all YAML task files under a base directory, skipping molecule."""
return AnsibleFileFinder.find_task_files(base, skip_molecule=True)
def check_patterns(path: Path | None, ansible_dirs: list[Path] | None = None) -> list[str]:
"""Check Ansible tasks for dangerous failure-masking patterns.
Args:
path: Specific file or directory to check. If None, *ansible_dirs*
is used.
ansible_dirs: Directories to scan when *path* is None.
Returns:
List of violation messages (empty if clean).
"""
if path:
files = _find_task_files(path)
else:
files: list[Path] = []
for d in ansible_dirs or []:
files.extend(_find_task_files(d))
all_violations: list[str] = []
for f in files:
all_violations.extend(_check_file(f, REPO_ROOT))
return all_violations
src/devx/tools/ansible_checks/set_fact_to_json.py
+134
View File
@@ -0,0 +1,134 @@
"""Check that Ansible ``set_fact`` tasks don't misuse ``| to_json``.
Extracted from :mod:`devx.tools.check_ansible_set_fact_to_json` as part
of the Ansible check tool consolidation. The old module remains as a
thin wrapper for backward compatibility.
"""
from __future__ import annotations
from pathlib import Path
import yaml
from devx.tools.ansible_checks._shared import AnsibleFileFinder, ViolationReporter
REPO_ROOT = Path.cwd()
TO_JSON_FILTERS = ("| to_json", "| to_nice_json", "|to_json", "|to_nice_json")
def _find_task_files(base: Path) -> list[Path]:
"""Find all YAML task files under a base directory."""
return AnsibleFileFinder.find_task_files(base, skip_molecule=False)
def _check_file(filepath: Path, repo_root: Path) -> list[str]:
"""Check a single YAML file for set_fact + to_json misuse."""
errors: list[str] = []
try:
content = filepath.read_text(encoding="utf-8")
except (OSError, UnicodeDecodeError):
return errors
try:
docs = list(yaml.safe_load_all(content))
except yaml.YAMLError as exc:
return [f"{filepath}: cannot parse YAML: {exc}"]
for doc in docs:
if isinstance(doc, list):
for item in doc:
if isinstance(item, dict):
if any(k in item for k in ("tasks", "pre_tasks", "post_tasks", "handlers", "roles")):
_check_tasks(item, filepath, errors, repo_root)
else:
_check_task(item, filepath, errors, repo_root)
block = item.get("block")
if isinstance(block, list):
_check_task_list(block, filepath, errors, repo_root)
elif isinstance(doc, dict):
_check_tasks(doc, filepath, errors, repo_root)
return errors
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."""
tasks = doc.get("tasks")
if isinstance(tasks, list):
_check_task_list(tasks, filepath, errors, repo_root)
for role_key in ("pre_tasks", "post_tasks", "handlers"):
section = doc.get(role_key)
if isinstance(section, list):
_check_task_list(section, filepath, errors, repo_root)
roles = doc.get("roles")
if isinstance(roles, list):
for role_entry in roles:
if isinstance(role_entry, dict):
role_tasks = role_entry.get("tasks")
if isinstance(role_tasks, list):
_check_task_list(role_tasks, filepath, errors, repo_root)
def _check_task_list(tasks: list, filepath: Path, errors: list[str], repo_root: Path) -> None:
"""Check a list of task definitions for set_fact + to_json."""
for task in tasks:
if not isinstance(task, dict):
continue
_check_task(task, filepath, errors, repo_root)
block = task.get("block")
if isinstance(block, list):
_check_task_list(block, filepath, errors, repo_root)
def _check_task(task: dict, filepath: Path, errors: list[str], repo_root: Path) -> None:
"""Check a single task for set_fact + to_json misuse."""
has_set_fact = False
for key in task:
if key in {"set_fact", "ansible.builtin.set_fact"}:
has_set_fact = True
break
if not has_set_fact:
return
set_fact_body = task.get("set_fact") or task.get("ansible.builtin.set_fact")
if not isinstance(set_fact_body, dict):
return
task_name = task.get("name", "(unnamed)")
for fact_name, fact_value in set_fact_body.items():
if fact_name in ("cacheable",):
continue
value_str = str(fact_value)
for filter_pattern in TO_JSON_FILTERS:
if filter_pattern in value_str:
display_path = ViolationReporter.format_violation(filepath, repo_root, None, "")
display_path = display_path.removesuffix(" — ")
errors.append(
f"{display_path}: task '{task_name}' "
f"sets fact '{fact_name}' with '{filter_pattern.strip()}' "
f"— this converts native Python types to JSON strings. "
f"Remove the filter to preserve the native type, or use "
f"'| from_json' in the consuming task if the string "
f"representation is intentional."
)
break
def check_set_fact_to_json(path: Path | None, ansible_dirs: list[Path] | None = None) -> list[str]:
"""Check that set_fact tasks don't misuse to_json.
Args:
path: Specific file or directory to check. If None, *ansible_dirs*
is used.
ansible_dirs: Directories to scan when *path* is None.
Returns:
List of error messages (empty if all OK).
"""
if path:
files = _find_task_files(path)
else:
files: list[Path] = []
for d in ansible_dirs or []:
files.extend(_find_task_files(d))
all_errors: list[str] = []
for f in files:
all_errors.extend(_check_file(f, REPO_ROOT))
return all_errors
src/devx/tools/check_ansible_no_log.py
+16 -177
View File
@@ -1,17 +1,9 @@
"""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.
Thin wrapper around :mod:`devx.tools.ansible_checks.no_log` for
backward compatibility. The check logic lives in the subpackage; this
module preserves the CLI entry point and re-exports the internal
helpers so existing tests and imports continue to work.
Usage::
@@ -24,178 +16,25 @@ 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
from devx.tools.ansible_checks.no_log import (
NON_VALUE_KEYS, # noqa: F401
SECRET_PATTERNS, # noqa: F401 — re-exported for backward compat
TASK_VALUE_KEYS, # noqa: F401
_check_task, # noqa: F401
_contains_secret, # noqa: F401
_has_no_log, # noqa: F401
check_directory, # noqa: F401
check_no_log, # noqa: F401
)
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(
@@ -216,7 +55,7 @@ def main(path: Path | None, ansible_dir: Path | None) -> None:
click.echo(f"Error: {target} is not a directory", err=True)
sys.exit(2)
violations = check_directory(target)
violations = check_no_log(target)
if violations:
click.echo(f"Found {len(violations)} task(s) handling secrets without no_log:\n")
src/devx/tools/check_ansible_no_state_absent_on_db.py
+17 -121
View File
@@ -1,18 +1,10 @@
"""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
Thin wrapper around
:mod:`devx.tools.ansible_checks.no_state_absent_on_db` for backward
compatibility. The check logic lives in the subpackage; this module
preserves the CLI entry point and re-exports the internal helpers so
existing tests and imports continue to work.
Usage::
@@ -24,114 +16,27 @@ 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()
from devx.tools.ansible_checks.no_state_absent_on_db import (
ALLOW_MARKER, # noqa: F401 — re-exported for backward compat
ALLOWED_CONTEXT_KEYWORDS, # noqa: F401
DB_PATH_PATTERNS, # noqa: F401
DESTRUCTIVE_PATTERNS, # noqa: F401
REPO_ROOT,
_check_file, # noqa: F401
_find_task_files, # noqa: F401
check_no_state_absent_on_db, # noqa: F401
)
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(
@@ -150,16 +55,7 @@ def _check_file(filepath: Path, repo_root: Path) -> list[str]:
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))
all_violations = check_no_state_absent_on_db(path) if path else check_no_state_absent_on_db(None, dirs)
if all_violations:
click.echo("[check-ansible-no-state-absent-on-db] FAIL: destructive operations on DB paths:")
src/devx/tools/check_ansible_patterns.py
+25 -291
View File
@@ -1,18 +1,9 @@
"""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.
Thin wrapper around :mod:`devx.tools.ansible_checks.patterns` for
backward compatibility. The check logic lives in the subpackage; this
module preserves the CLI entry point and re-exports the internal
helpers so existing tests and imports continue to work.
Usage::
@@ -24,284 +15,36 @@ 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()
from devx.tools.ansible_checks.patterns import (
ALLOW_MARKER, # noqa: F401 — re-exported for backward compat
COMMAND_VALUE_KEYS, # noqa: F401
CRITICAL_TASK_KEYWORDS, # noqa: F401
LEGITIMATE_COMMAND_PREFIXES, # noqa: F401
LEGITIMATE_FAILED_WHEN_KEYWORDS, # noqa: F401
LEGITIMATE_TASK_NAME_KEYWORDS, # noqa: F401
OR_TRUE_PATTERN, # noqa: F401
REDIRECT_DEVNULL_PATTERN, # noqa: F401
REPO_ROOT,
SHELL_MODULE_KEYS, # noqa: F401
_check_file, # noqa: F401
_check_task, # noqa: F401
_check_tasks, # noqa: F401
_find_task_files, # noqa: F401
_is_legitimate_devnull, # noqa: F401
_is_legitimate_or_true, # noqa: F401
check_patterns, # noqa: F401
)
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(
@@ -320,16 +63,7 @@ def _find_task_files(base: Path) -> list[Path]:
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))
all_violations = check_patterns(path) if path else check_patterns(None, dirs)
if all_violations:
click.echo("[check-ansible-patterns] FAIL: dangerous failure-masking patterns found:")
src/devx/tools/check_ansible_set_fact_to_json.py
+16 -143
View File
@@ -1,19 +1,9 @@
"""Check that Ansible ``set_fact`` tasks don't misuse ``| to_json``.
This prevents the class of bug where ``set_fact`` tasks use
``{{ targets | to_json }}`` to store Python lists, but ``to_json``
converts native types to JSON strings. Ansible then stored the result
as a string, so iterating over the fact yielded individual characters
instead of list items, causing ``object of type 'str' has no attribute
'ip'`` errors.
The check scans all Ansible task files (playbooks and role tasks) for
``set_fact`` tasks where any value uses ``| to_json`` or ``| to_nice_json``
and flags them as potential bugs.
``| to_json`` is legitimate in Jinja2 templates (e.g., rendering JSON
config files) but almost never correct in ``set_fact`` — the fact should
store the native Python type so downstream tasks can iterate/index it.
Thin wrapper around :mod:`devx.tools.ansible_checks.set_fact_to_json`
for backward compatibility. The check logic lives in the subpackage;
this module preserves the CLI entry point and re-exports the internal
helpers so existing tests and imports continue to work.
Usage::
@@ -29,130 +19,23 @@ import sys
from pathlib import Path
import click
import yaml
REPO_ROOT = Path.cwd()
from devx.tools.ansible_checks.set_fact_to_json import (
REPO_ROOT,
TO_JSON_FILTERS, # noqa: F401 — re-exported for backward compat
_check_file, # noqa: F401
_check_task, # noqa: F401
_check_task_list, # noqa: F401
_check_tasks, # noqa: F401
_find_task_files, # noqa: F401
check_set_fact_to_json, # noqa: F401
)
DEFAULT_ANSIBLE_DIRS: list[Path] = [
REPO_ROOT / "ansible" / "playbooks",
REPO_ROOT / "ansible" / "roles",
]
TO_JSON_FILTERS = ("| to_json", "| to_nice_json", "|to_json", "|to_nice_json")
def _find_task_files(base: Path) -> list[Path]:
"""Find all YAML task files under a base directory."""
if base.is_file() and base.suffix in (".yml", ".yaml"):
return [base]
if not base.is_dir():
return []
return sorted(base.rglob("*.yml")) + sorted(base.rglob("*.yaml"))
def _check_file(filepath: Path, repo_root: Path) -> list[str]:
"""Check a single YAML file for set_fact + to_json misuse.
Returns a list of error messages (empty if all OK).
"""
errors: list[str] = []
content = filepath.read_text(encoding="utf-8")
# Multi-document YAML (--- separators) is common in playbooks
try:
docs = list(yaml.safe_load_all(content))
except yaml.YAMLError as exc:
return [f"{filepath}: cannot parse YAML: {exc}"]
for doc in docs:
if isinstance(doc, list):
# Could be a playbook (list of plays) or a role tasks file (list of tasks)
for item in doc:
if isinstance(item, dict):
if any(k in item for k in ("tasks", "pre_tasks", "post_tasks", "handlers", "roles")):
# It's a play
_check_tasks(item, filepath, errors, repo_root)
else:
# It's a bare task (role tasks file)
_check_task(item, filepath, errors, repo_root)
block = item.get("block")
if isinstance(block, list):
_check_task_list(block, filepath, errors, repo_root)
elif isinstance(doc, dict):
# Role tasks file or single play — _check_tasks handles all task sections
_check_tasks(doc, filepath, errors, repo_root)
return errors
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."""
tasks = doc.get("tasks")
if isinstance(tasks, list):
_check_task_list(tasks, filepath, errors, repo_root)
for role_key in ("pre_tasks", "post_tasks", "handlers"):
section = doc.get(role_key)
if isinstance(section, list):
_check_task_list(section, filepath, errors, repo_root)
# Check tasks in roles imported via `roles:` key
roles = doc.get("roles")
if isinstance(roles, list):
for role_entry in roles:
if isinstance(role_entry, dict):
role_tasks = role_entry.get("tasks")
if isinstance(role_tasks, list):
_check_task_list(role_tasks, filepath, errors, repo_root)
def _check_task_list(tasks: list, filepath: Path, errors: list[str], repo_root: Path) -> None:
"""Check a list of task definitions for set_fact + to_json."""
for task in tasks:
if not isinstance(task, dict):
continue
_check_task(task, filepath, errors, repo_root)
# Check nested block tasks
block = task.get("block")
if isinstance(block, list):
_check_task_list(block, filepath, errors, repo_root)
def _check_task(task: dict, filepath: Path, errors: list[str], repo_root: Path) -> None:
"""Check a single task for set_fact + to_json misuse."""
# Detect set_fact — could be a module name key or ansible.builtin.set_fact
has_set_fact = False
for key in task:
if key in {"set_fact", "ansible.builtin.set_fact"}:
has_set_fact = True
break
if not has_set_fact:
return
set_fact_body = task.get("set_fact") or task.get("ansible.builtin.set_fact")
if not isinstance(set_fact_body, dict):
return
task_name = task.get("name", "(unnamed)")
for fact_name, fact_value in set_fact_body.items():
if fact_name in ("cacheable",):
continue
value_str = str(fact_value)
for filter_pattern in TO_JSON_FILTERS:
if filter_pattern in value_str:
try:
display_path = filepath.relative_to(repo_root)
except ValueError:
display_path = filepath
errors.append(
f"{display_path}: task '{task_name}' "
f"sets fact '{fact_name}' with '{filter_pattern.strip()}' "
f"— this converts native Python types to JSON strings. "
f"Remove the filter to preserve the native type, or use "
f"'| from_json' in the consuming task if the string "
f"representation is intentional."
)
break # One error per fact is enough
@click.command()
@click.option(
@@ -171,17 +54,7 @@ def _check_task(task: dict, filepath: Path, errors: list[str], repo_root: Path)
def main(path: Path | None, ansible_dirs: tuple[Path, ...]) -> None:
"""Check that set_fact tasks don't misuse to_json."""
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_errors: list[str] = []
for f in files:
errors = _check_file(f, REPO_ROOT)
all_errors.extend(errors)
all_errors = check_set_fact_to_json(path) if path else check_set_fact_to_json(None, dirs)
if all_errors:
click.echo("[check-ansible-set-fact-to-json] FAIL: set_fact with to_json found:")
src/devx/tools/check_jinja_expr.py
+15 -242
View File
@@ -1,15 +1,9 @@
"""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).
Thin wrapper around :mod:`devx.tools.ansible_checks.jinja_expr` for
backward compatibility. The check logic lives in the subpackage; this
module preserves the CLI entry point and re-exports the internal
helpers so existing tests and imports continue to work.
Usage::
@@ -21,233 +15,21 @@ 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
from devx.tools.ansible_checks.jinja_expr import (
EXPR_PATTERN, # noqa: F401 — re-exported for backward compat
MOCK_CONTEXT, # noqa: F401
_check_file, # noqa: F401
_default_ansible_dirs, # noqa: F401
_extract_expressions, # noqa: F401
_find_yaml_files, # noqa: F401
_render_expression, # noqa: F401
check_jinja_expr, # noqa: F401
)
@click.command()
@@ -267,16 +49,7 @@ def _check_file(filepath: Path, repo_root: Path) -> list[str]:
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))
all_violations = check_jinja_expr(path) if path else check_jinja_expr(None, dirs)
if all_violations:
click.echo("[check-jinja-expr] FAIL: invalid Jinja expressions found:")
src/devx/translations.json
+112
View File
@@ -845,6 +845,14 @@
"zh": "--checklist-confirmed is required for APPROVE events.",
"Cleaning up: running molecule destroy for {scenario}": "Cleaning up: running molecule destroy for {scenario}"
},
"--poll-interval must be positive": {
"bg": "--poll-interval must be positive",
"de": "--poll-interval must be positive",
"en": "--poll-interval must be positive",
"pl": "--poll-interval must be positive",
"ru": "--poll-interval must be positive",
"zh": "--poll-interval must be positive"
},
"--push requires --registry": {
"bg": "--push requires --registry",
"de": "--push requires --registry",
@@ -854,6 +862,14 @@
"zh": "--push requires --registry",
"Cleaning up: running molecule destroy for {scenario}": "Cleaning up: running molecule destroy for {scenario}"
},
"--repo is required (or set GITHUB_REPOSITORY=owner/name)": {
"bg": "--repo is required (or set GITHUB_REPOSITORY=owner/name)",
"de": "--repo is required (or set GITHUB_REPOSITORY=owner/name)",
"en": "--repo is required (or set GITHUB_REPOSITORY=owner/name)",
"pl": "--repo is required (or set GITHUB_REPOSITORY=owner/name)",
"ru": "--repo is required (or set GITHUB_REPOSITORY=owner/name)",
"zh": "--repo is required (or set GITHUB_REPOSITORY=owner/name)"
},
"--skip-build: skipping package build and PyPI publish.": {
"bg": "--skip-build: skipping package build and PyPI publish.",
"de": "--skip-build: skipping package build and PyPI publish.",
@@ -863,6 +879,14 @@
"zh": "--skip-build: skipping package build and PyPI publish.",
"Cleaning up: running molecule destroy for {scenario}": "Cleaning up: running molecule destroy for {scenario}"
},
"--timeout must be positive": {
"bg": "--timeout must be positive",
"de": "--timeout must be positive",
"en": "--timeout must be positive",
"pl": "--timeout must be positive",
"ru": "--timeout must be positive",
"zh": "--timeout must be positive"
},
"=== Release Alignment Verification ===\n": {
"bg": "=== Release Alignment Verification ===\n",
"de": "=== Release Alignment Verification ===\n",
@@ -908,6 +932,14 @@
"zh": "Additional directory to scan (default: scripts, tests). Can be repeated.",
"Cleaning up: running molecule destroy for {scenario}": "Cleaning up: running molecule destroy for {scenario}"
},
"All matching jobs completed successfully: {jobs}": {
"bg": "All matching jobs completed successfully: {jobs}",
"de": "All matching jobs completed successfully: {jobs}",
"en": "All matching jobs completed successfully: {jobs}",
"pl": "All matching jobs completed successfully: {jobs}",
"ru": "All matching jobs completed successfully: {jobs}",
"zh": "All matching jobs completed successfully: {jobs}"
},
"All molecule tests passed.": {
"bg": "All molecule tests passed.",
"de": "All molecule tests passed.",
@@ -2167,6 +2199,14 @@
"zh": "输入必须是 JSON 数组，得到 {type}",
"Cleaning up: running molecule destroy for {scenario}": "Cleaning up: running molecule destroy for {scenario}"
},
"Job(s) completed with non-success conclusion: {jobs}": {
"bg": "Job(s) completed with non-success conclusion: {jobs}",
"de": "Job(s) completed with non-success conclusion: {jobs}",
"en": "Job(s) completed with non-success conclusion: {jobs}",
"pl": "Job(s) completed with non-success conclusion: {jobs}",
"ru": "Job(s) completed with non-success conclusion: {jobs}",
"zh": "Job(s) completed with non-success conclusion: {jobs}"
},
"Label '{label}' already on PR #{pr}.": {
"bg": "Label '{label}' already on PR #{pr}.",
"de": "Label '{label}' already on PR #{pr}.",
@@ -2419,6 +2459,14 @@
"zh": "No jobs found for run #{run_id}.",
"Cleaning up: running molecule destroy for {scenario}": "Cleaning up: running molecule destroy for {scenario}"
},
"No matching jobs found for prefix '{prefix}' within timeout.": {
"bg": "No matching jobs found for prefix '{prefix}' within timeout.",
"de": "No matching jobs found for prefix '{prefix}' within timeout.",
"en": "No matching jobs found for prefix '{prefix}' within timeout.",
"pl": "No matching jobs found for prefix '{prefix}' within timeout.",
"ru": "No matching jobs found for prefix '{prefix}' within timeout.",
"zh": "No matching jobs found for prefix '{prefix}' within timeout."
},
"No open PR found for branch '{branch}'.": {
"bg": "No open PR found for branch '{branch}'.",
"de": "No open PR found for branch '{branch}'.",
@@ -3454,6 +3502,14 @@
"zh": "Timeout reached after {timeout}s.",
"Cleaning up: running molecule destroy for {scenario}": "Cleaning up: running molecule destroy for {scenario}"
},
"Timeout reached waiting for jobs matching '{prefix}'.": {
"bg": "Timeout reached waiting for jobs matching '{prefix}'.",
"de": "Timeout reached waiting for jobs matching '{prefix}'.",
"en": "Timeout reached waiting for jobs matching '{prefix}'.",
"pl": "Timeout reached waiting for jobs matching '{prefix}'.",
"ru": "Timeout reached waiting for jobs matching '{prefix}'.",
"zh": "Timeout reached waiting for jobs matching '{prefix}'."
},
"Transitive-subprocess advisories (runtime audit is authoritative):": {
"bg": "Transitive-subprocess advisories (runtime audit is authoritative):",
"de": "Transitive-subprocess advisories (runtime audit is authoritative):",
@@ -3706,6 +3762,30 @@
"zh": "Waiting for CI checks to complete (timeout: {timeout}s)...",
"Cleaning up: running molecule destroy for {scenario}": "Cleaning up: running molecule destroy for {scenario}"
},
"Waiting for jobs matching '{prefix}' in {repo} (timeout={timeout}s, interval={interval}s)": {
"bg": "Waiting for jobs matching '{prefix}' in {repo} (timeout={timeout}s, interval={interval}s)",
"de": "Waiting for jobs matching '{prefix}' in {repo} (timeout={timeout}s, interval={interval}s)",
"en": "Waiting for jobs matching '{prefix}' in {repo} (timeout={timeout}s, interval={interval}s)",
"pl": "Waiting for jobs matching '{prefix}' in {repo} (timeout={timeout}s, interval={interval}s)",
"ru": "Waiting for jobs matching '{prefix}' in {repo} (timeout={timeout}s, interval={interval}s)",
"zh": "Waiting for jobs matching '{prefix}' in {repo} (timeout={timeout}s, interval={interval}s)"
},
"Warning: actions runs query failed: {error}": {
"bg": "Warning: actions runs query failed: {error}",
"de": "Warning: actions runs query failed: {error}",
"en": "Warning: actions runs query failed: {error}",
"pl": "Warning: actions runs query failed: {error}",
"ru": "Warning: actions runs query failed: {error}",
"zh": "Warning: actions runs query failed: {error}"
},
"Warning: actions runs query returned HTTP {status}": {
"bg": "Warning: actions runs query returned HTTP {status}",
"de": "Warning: actions runs query returned HTTP {status}",
"en": "Warning: actions runs query returned HTTP {status}",
"pl": "Warning: actions runs query returned HTTP {status}",
"ru": "Warning: actions runs query returned HTTP {status}",
"zh": "Warning: actions runs query returned HTTP {status}"
},
"Warning: could not fetch tags from origin.": {
"bg": "Warning: could not fetch tags from origin.",
"de": "Warning: could not fetch tags from origin.",
@@ -3733,6 +3813,22 @@
"zh": "Warning: instance-level runners query returned HTTP {status}",
"Cleaning up: running molecule destroy for {scenario}": "Cleaning up: running molecule destroy for {scenario}"
},
"Warning: jobs query for run {run_id} failed: {error}": {
"bg": "Warning: jobs query for run {run_id} failed: {error}",
"de": "Warning: jobs query for run {run_id} failed: {error}",
"en": "Warning: jobs query for run {run_id} failed: {error}",
"pl": "Warning: jobs query for run {run_id} failed: {error}",
"ru": "Warning: jobs query for run {run_id} failed: {error}",
"zh": "Warning: jobs query for run {run_id} failed: {error}"
},
"Warning: jobs query for run {run_id} returned HTTP {status}": {
"bg": "Warning: jobs query for run {run_id} returned HTTP {status}",
"de": "Warning: jobs query for run {run_id} returned HTTP {status}",
"en": "Warning: jobs query for run {run_id} returned HTTP {status}",
"pl": "Warning: jobs query for run {run_id} returned HTTP {status}",
"ru": "Warning: jobs query for run {run_id} returned HTTP {status}",
"zh": "Warning: jobs query for run {run_id} returned HTTP {status}"
},
"Warning: org-level runners query failed: {error}": {
"bg": "Warning: org-level runners query failed: {error}",
"de": "Warning: org-level runners query failed: {error}",
@@ -4048,6 +4144,22 @@
"zh": "[tool.devx] 缺少必需的键: {keys}",
"Cleaning up: running molecule destroy for {scenario}": "Cleaning up: running molecule destroy for {scenario}"
},
"[{tool}] FAIL: {count} violation(s) found.": {
"bg": "[{tool}] FAIL: {count} violation(s) found.",
"de": "[{tool}] FAIL: {count} violation(s) found.",
"en": "[{tool}] FAIL: {count} violation(s) found.",
"pl": "[{tool}] FAIL: {count} violation(s) found.",
"ru": "[{tool}] FAIL: {count} violation(s) found.",
"zh": "[{tool}] FAIL: {count} violation(s) found."
},
"[{tool}] OK: no violations found.": {
"bg": "[{tool}] OK: no violations found.",
"de": "[{tool}] OK: no violations found.",
"en": "[{tool}] OK: no violations found.",
"pl": "[{tool}] OK: no violations found.",
"ru": "[{tool}] OK: no violations found.",
"zh": "[{tool}] OK: no violations found."
},
"active": {
"bg": "активен",
"de": "aktiv",