From 48596627d53efbfb35444dc4a4442e346086ea9f Mon Sep 17 00:00:00 2001 From: emil User Date: Wed, 12 Aug 2026 17:37:49 +0000 Subject: [PATCH] DEVX-155: refactor: extract wait_for_checks, consolidate ansible_checks, deprecate ci/discover_runners --- AGENTS.md | 23 +- ...check-consolidation-and-wait-for-checks.md | 108 ++++++ docs/mapping.json | 4 +- docs/tech/architecture.md | 27 +- docs/user/cli-commands.md | 31 ++ src/devx/ci/discover_runners.py | 201 ++--------- src/devx/ci/wait_for_checks.py | 209 ++++++++++++ src/devx/cli.py | 7 + src/devx/molecule/discover_runners.py | 32 +- src/devx/tools/ansible_checks/__init__.py | 34 ++ src/devx/tools/ansible_checks/_shared.py | 178 ++++++++++ src/devx/tools/ansible_checks/jinja_expr.py | 226 +++++++++++++ src/devx/tools/ansible_checks/no_log.py | 128 +++++++ .../ansible_checks/no_state_absent_on_db.py | 100 ++++++ src/devx/tools/ansible_checks/patterns.py | 226 +++++++++++++ .../tools/ansible_checks/set_fact_to_json.py | 134 ++++++++ src/devx/tools/check_ansible_no_log.py | 193 +---------- .../check_ansible_no_state_absent_on_db.py | 138 +------- src/devx/tools/check_ansible_patterns.py | 316 ++---------------- .../tools/check_ansible_set_fact_to_json.py | 159 +-------- src/devx/tools/check_jinja_expr.py | 257 +------------- src/devx/translations.json | 112 +++++++ tests/unit/test_ansible_checks_shared.py | 231 +++++++++++++ tests/unit/test_cli.py | 7 + tests/unit/test_discover_runners.py | 83 +++-- ...st_tools_check_ansible_set_fact_to_json.py | 12 + tests/unit/test_wait_for_checks.py | 272 +++++++++++++++ 27 files changed, 2251 insertions(+), 1197 deletions(-) create mode 100644 docs/decisions/0002-ansible-check-consolidation-and-wait-for-checks.md create mode 100644 src/devx/ci/wait_for_checks.py create mode 100644 src/devx/tools/ansible_checks/__init__.py create mode 100644 src/devx/tools/ansible_checks/_shared.py create mode 100644 src/devx/tools/ansible_checks/jinja_expr.py create mode 100644 src/devx/tools/ansible_checks/no_log.py create mode 100644 src/devx/tools/ansible_checks/no_state_absent_on_db.py create mode 100644 src/devx/tools/ansible_checks/patterns.py create mode 100644 src/devx/tools/ansible_checks/set_fact_to_json.py create mode 100644 tests/unit/test_ansible_checks_shared.py create mode 100644 tests/unit/test_wait_for_checks.py diff --git a/AGENTS.md b/AGENTS.md index 7179915..0aa5dde 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -98,7 +98,9 @@ src/devx/ │ ├── record_deployed_tag.py # Record deployed tag to Gitea repo variable │ ├── cancel_superseded_runs.py # Cancel in-flight CI runs for the same PR branch │ ├── check_workflow_artifact_deps.py # Verify artifact download jobs depend on upload jobs -│ └── check_workflow_tofu_init.py # Verify tofu-state jobs have a tofu-init step +│ ├── check_workflow_tofu_init.py # Verify tofu-state jobs have a tofu-init step +│ ├── discover_runners.py # Deprecated wrapper → molecule/discover_runners +│ └── wait_for_checks.py # Poll Gitea Actions for job completion (replaces inline shell polling) ├── tools/ # Developer tooling modules (run locally or by CI) │ ├── setup.py # Environment setup (venv, deps, hooks) │ ├── install_tools.py # Install actionlint, git-cliff, act_runner, tea, hadolint, vale @@ -122,12 +124,19 @@ src/devx/ │ ├── pr_label.py # Add labels to PRs (idempotent) │ ├── pre_push_check.py # Validate Vikunja task existence before push │ ├── check_docker_init.py # Check Docker Compose services with healthchecks have init: true -│ ├── check_ansible_set_fact_to_json.py # Check set_fact tasks don't misuse to_json +│ ├── check_ansible_set_fact_to_json.py # Thin wrapper → ansible_checks/set_fact_to_json │ ├── check_alert_rules.py # Validate Prometheus alert rules with promtool -│ ├── check_ansible_no_log.py # Check Ansible tasks for missing no_log on secrets -│ ├── check_ansible_patterns.py # Detect dangerous failure-masking patterns -│ ├── check_jinja_expr.py # Validate Jinja2 expressions in Ansible files -│ ├── check_ansible_no_state_absent_on_db.py # Prevent state:absent on DB paths +│ ├── check_ansible_no_log.py # Thin wrapper → ansible_checks/no_log +│ ├── check_ansible_patterns.py # Thin wrapper → ansible_checks/patterns +│ ├── check_jinja_expr.py # Thin wrapper → ansible_checks/jinja_expr +│ ├── check_ansible_no_state_absent_on_db.py # Thin wrapper → ansible_checks/no_state_absent_on_db +│ ├── ansible_checks/ # Composable Ansible check subpackage (canonical implementations) +│ │ ├── _shared.py # AnsibleFileFinder, AnsibleYAMLParser, ViolationReporter +│ │ ├── no_log.py # Check missing no_log on secret-handling tasks +│ │ ├── patterns.py # Detect dangerous failure-masking patterns +│ │ ├── set_fact_to_json.py # Check set_fact tasks don't misuse to_json +│ │ ├── no_state_absent_on_db.py # Prevent state:absent on DB paths +│ │ └── jinja_expr.py # Validate Jinja2 expressions in Ansible files │ └── _shared.py # Shared tool utilities ├── opentofu.py # OpenTofu output helpers (get_tofu_output, get_tofu_vm_ip, get_tofu_vm_field) ├── utils/ # Shared utilities (reusable across projects) @@ -143,7 +152,7 @@ src/devx/ │ ├── ui.py # say() — unified click.echo + logging output │ └── jinja.py # Jinja2 environment helpers + Ansible-compatible filters └── molecule/ # Optional molecule testing helpers (for Ansible projects) - ├── discover_runners.py # Dynamic Gitea runner discovery + ├── discover_runners.py # Dynamic Gitea runner discovery (canonical; ci/discover_runners is a deprecated wrapper) ├── distribute_molecule.py # Distribute molecule scenarios across runners (LPT scheduling, --roles-root for multi-role) ├── molecule_ci_guard.py # Run molecule with cross-runner fail-fast (--roles-root) ├── molecule_all.py # Run all molecule scenarios locally diff --git a/docs/decisions/0002-ansible-check-consolidation-and-wait-for-checks.md b/docs/decisions/0002-ansible-check-consolidation-and-wait-for-checks.md new file mode 100644 index 0000000..9ef8009 --- /dev/null +++ b/docs/decisions/0002-ansible-check-consolidation-and-wait-for-checks.md @@ -0,0 +1,108 @@ +# ADR-0002: Ansible Check Tool Consolidation and wait_for_checks Extraction + +Date: 2026-08-12 +Status: Accepted + +## Context + +The devx package had two categories of code duplication and inline +workflow logic that were hard to test and maintain: + +### 1. Ansible Check Tools — Duplicated Boilerplate + +Five Ansible check tools (`check_ansible_no_log`, +`check_ansible_patterns`, `check_ansible_set_fact_to_json`, +`check_ansible_no_state_absent_on_db`, `check_jinja_expr`) each +implemented their own file discovery, YAML parsing, task iteration, and +violation reporting logic. While the check logic differed, the +supporting infrastructure was copy-pasted across all five modules: + +- `find_task_files()` — glob YAML files, skip molecule +- YAML multi-document parsing with error handling +- Task iteration (bare lists, play dicts with `tasks`/`pre_tasks`/`post_tasks`/`handlers`, nested `block` tasks) +- Violation formatting (`path:line — message`) + +This made it difficult to add new checks (each new tool repeated the +boilerplate) and risky to change shared behavior (fixes had to be +applied to all five modules independently). + +### 2. Inline Job Polling in Workflow YAML + +The `grm` repository's `ci.yml` workflow contained ~25 lines of inline +shell + Python polling logic to wait for the `molecule-tests` job to +complete before the auto-merge step. This logic: + +- Was not testable (embedded in workflow YAML) +- Duplicated the Gitea API client pattern already used elsewhere +- Had no timeout handling, no error reporting, no retry logic +- Could not be reused by other repositories + +### 3. Duplicate discover_runners Modules + +`devx.ci.discover_runners` and `devx.molecule.discover_runners` were +near-identical modules. The `ci/` version had better error logging +(warnings on non-200 responses, 403 suppression for instance-level +queries), while the `molecule/` version silently swallowed errors. +Both were imported by different workflows, making it unclear which was +canonical. + +## Decision + +### 1. Composable `ansible_checks/` Subpackage + +Consolidate the five Ansible check tools into a +`devx.tools.ansible_checks/` subpackage with shared utilities: + +- `_shared.py` — `AnsibleFileFinder`, `AnsibleYAMLParser`, + `ViolationReporter` classes providing composable helpers +- `no_log.py`, `patterns.py`, `set_fact_to_json.py`, + `no_state_absent_on_db.py`, `jinja_expr.py` — canonical check + implementations using the shared utilities + +The old modules (`check_ansible_*.py`, `check_jinja_expr.py`) remain as +**thin backward-compat wrappers** that re-export the canonical +implementation and preserve the CLI entry point. This avoids breaking +existing Makefile targets and workflow references. + +**Composition over inheritance**: each check module picks the helpers it +needs. Tools that don't parse YAML (for example line-based scanners) can skip +`AnsibleYAMLParser` entirely. + +### 2. Extracted `wait_for_checks` Module + +Extract the inline polling logic into `devx.ci.wait_for_checks`: + +- Polls the Gitea API for job completion status +- Configurable job name prefix, timeout, poll interval +- Exit codes: 0 (success), 1 (failure), 2 (timeout), 3 (API error) +- `--require-success/--no-require-success` flag for flexibility +- 100% test coverage with mocked API responses + +This replaces the inline shell polling in `grm` `ci.yml` with a +reusable, testable Python module. + +### 3. Deprecated `ci/discover_runners` Wrapper + +Merge the `ci/discover_runners` implementation (with its better error +logging) into `molecule/discover_runners` as the canonical version. +Make `ci/discover_runners` a deprecated wrapper that: + +- Re-exports all public symbols from `molecule.discover_runners` +- Emits a `DeprecationWarning` when run as `__main__` +- Preserves backward compatibility for existing workflow references + +New code should import from `devx.molecule.discover_runners` directly. + +## Consequences + +- **New checks are easier to write**: import `_shared` helpers, implement + only the check-specific logic +- **Shared behavior can be fixed in one place**: file discovery, YAML + parsing, violation formatting +- **Workflow polling is testable**: `wait_for_checks` has 26 unit tests + covering success, failure, timeout, and API error scenarios +- **Backward compatibility preserved**: all existing Makefile targets, + workflow references, and test imports continue to work via wrappers +- **Migration path is gradual**: new code uses the subpackage; old code + can migrate at its own pace; wrappers can be removed in a future + release once all references are updated diff --git a/docs/mapping.json b/docs/mapping.json index 81dffd0..8f8846f 100644 --- a/docs/mapping.json +++ b/docs/mapping.json @@ -3,5 +3,7 @@ "user/getting-started.md": "Getting-Started", "user/cli-commands.md": "CLI-Commands", "tech/architecture.md": "Architecture", - "tech/ci-cd-workflow.md": "CI-CD-Workflow" + "tech/ci-cd-workflow.md": "CI-CD-Workflow", + "decisions/0001-test-isolation-pytest-plugin-and-shift-left-quality-gates.md": "ADR-0001-Test-Isolation", + "decisions/0002-ansible-check-consolidation-and-wait-for-checks.md": "ADR-0002-Ansible-Check-Consolidation" } diff --git a/docs/tech/architecture.md b/docs/tech/architecture.md index 3cc658c..6a469f2 100644 --- a/docs/tech/architecture.md +++ b/docs/tech/architecture.md @@ -33,7 +33,8 @@ src/devx/ │ ├── notify_failure.py # Create Gitea issues on CI failures │ ├── distribute_files.py # Distribute files across parallel runners │ ├── integration_guard.py # Run pytest with cross-runner fail-fast -│ ├── discover_runners.py # Dynamic Gitea runner discovery +│ ├── discover_runners.py # Deprecated wrapper → molecule/discover_runners +│ ├── wait_for_checks.py # Poll Gitea Actions for job completion │ ├── check_translations.py # Translation completeness check │ └── doc_coverage.py # Documentation coverage check ├── tools/ # Developer tooling modules (run locally or by CI) @@ -48,7 +49,7 @@ src/devx/ │ └── install_checkmake.py # Install checkmake (Makefile linter) └── molecule/ # Optional molecule testing helpers (Ansible projects) ├── __init__.py - ├── discover_runners.py # Dynamic Gitea runner discovery + ├── discover_runners.py # Dynamic Gitea runner discovery (canonical) ├── distribute_molecule.py # Distribute scenarios across runners ├── molecule_ci_guard.py # Run molecule with cross-runner fail-fast ├── molecule_all.py # Run all molecule scenarios locally @@ -285,13 +286,24 @@ Click commands from `cli.py` and verifies each has documentation in `architecture.md` and CI scripts in `ci-cd-workflow.md`. Supports `--fail-on-missing` to enforce 100% coverage. -### `discover_runners.py` +### `discover_runners.py` (deprecated wrapper) + +> **Deprecated:** Use `devx.molecule.discover_runners` instead. This +> module is a thin wrapper that re-exports the canonical implementation. Discovers available Gitea Actions runners at three levels: repository, organization, and instance (administrator). Falls back to the `MOLECULE_RUNNERS` repo variable or `DEFAULT_MAX_RUNNERS` (3). Outputs runner count or a JSON index array for use as a dynamic matrix in Gitea Actions. +### `wait_for_checks.py` + +Polls the Gitea Actions API for job completion status. Used by auto-merge +jobs that need to wait for parallel jobs (for example molecule-tests) before +proceeding. Replaces inline shell polling in workflow YAML with a +reusable, testable Python module. Exit codes: 0 (success), 1 (job +failure), 2 (timeout), 3 (API error or no matching jobs). + ### `distribute_files.py` Distributes files matching a glob pattern across N parallel runners @@ -396,8 +408,13 @@ Intended for local development; CI uses the parallel matrix instead. ### `molecule/discover_runners.py` -Discovers available Gitea Actions runners for molecule tests. Same logic as -`devx.ci.discover_runners` but intended for molecule-specific workflows. +Discovers available Gitea Actions runners for molecule tests. This is the +canonical implementation; `devx.ci.discover_runners` is a deprecated wrapper +that re-exports from this module. Queries runners at repository, +organization, and instance (administrator) levels, with warnings logged +to stderr on non-200 responses (except 403 on instance-level, which is +expected without admin scope). Falls back to `MOLECULE_RUNNERS` env var +or `DEFAULT_MAX_RUNNERS` (3). ### `start_docker.py` diff --git a/docs/user/cli-commands.md b/docs/user/cli-commands.md index 915c77c..c4ff5e5 100644 --- a/docs/user/cli-commands.md +++ b/docs/user/cli-commands.md @@ -83,6 +83,11 @@ devx ci detect-release-commit ### `devx ci discover-runners` +> **Deprecated:** Use `devx molecule discover-runners` instead. This +> command is a thin wrapper that re-exports the canonical implementation +> from `devx.molecule.discover_runners`. It will be removed in a future +> release. + Discover available Gitea Actions runners for dynamic job distribution. Queries the Gitea API for registered runners at repository, organization, and instance (administrator) levels. Falls back to `MOLECULE_RUNNERS` repo variable or @@ -315,6 +320,32 @@ devx ci validate-commit-msg commit-msg.txt --branch master Options: - `--branch ` — override branch detection (for CI use) +### `devx ci wait-for-checks` + +Wait for Gitea Actions jobs to complete by polling the API. Used by +auto-merge jobs that need to wait for parallel jobs (for example molecule-tests) +before proceeding. Replaces inline shell polling in workflow YAML with +a reusable, testable Python module. + +Exit codes: +- `0` — all matching jobs completed successfully +- `1` — one or more matching jobs failed (when `--require-success` is set) +- `2` — timeout reached before all jobs completed +- `3` — API error or no matching jobs found + +```bash +devx ci wait-for-checks --job-name molecule-tests --repo oblachno-oss/grm +devx ci wait-for-checks --job-name molecule-tests --timeout 1200 --poll-interval 10 +devx ci wait-for-checks --job-name molecule-tests --no-require-success +``` + +Options: +- `--job-name ` — job name prefix to match (required) +- `--repo ` — repository (default: `$GITHUB_REPOSITORY`) +- `--timeout ` — max wait time (default: 1200 = 20 min) +- `--poll-interval ` — seconds between polls (default: 10) +- `--require-success / --no-require-success` — exit 1 if a job failed (default: yes) + ### `devx ci cancel-superseded-runs` Cancel in-flight CI runs for the same PR branch when a new push triggers diff --git a/src/devx/ci/discover_runners.py b/src/devx/ci/discover_runners.py index c31005d..9b2b63c 100644 --- a/src/devx/ci/discover_runners.py +++ b/src/devx/ci/discover_runners.py @@ -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()) diff --git a/src/devx/ci/wait_for_checks.py b/src/devx/ci/wait_for_checks.py new file mode 100644 index 0000000..1971d24 --- /dev/null +++ b/src/devx/ci/wait_for_checks.py @@ -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() diff --git a/src/devx/cli.py b/src/devx/cli.py index 94a0c94..ab95f05 100644 --- a/src/devx/cli.py +++ b/src/devx/cli.py @@ -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: diff --git a/src/devx/molecule/discover_runners.py b/src/devx/molecule/discover_runners.py index a938e6f..deec1c8 100644 --- a/src/devx/molecule/discover_runners.py +++ b/src/devx/molecule/discover_runners.py @@ -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 diff --git a/src/devx/tools/ansible_checks/__init__.py b/src/devx/tools/ansible_checks/__init__.py new file mode 100644 index 0000000..102dccc --- /dev/null +++ b/src/devx/tools/ansible_checks/__init__.py @@ -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", +] diff --git a/src/devx/tools/ansible_checks/_shared.py b/src/devx/tools/ansible_checks/_shared.py new file mode 100644 index 0000000..c65c92c --- /dev/null +++ b/src/devx/tools/ansible_checks/_shared.py @@ -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)) diff --git a/src/devx/tools/ansible_checks/jinja_expr.py b/src/devx/tools/ansible_checks/jinja_expr.py new file mode 100644 index 0000000..e49d6e0 --- /dev/null +++ b/src/devx/tools/ansible_checks/jinja_expr.py @@ -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 diff --git a/src/devx/tools/ansible_checks/no_log.py b/src/devx/tools/ansible_checks/no_log.py new file mode 100644 index 0000000..e67f2db --- /dev/null +++ b/src/devx/tools/ansible_checks/no_log.py @@ -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", "") + 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 diff --git a/src/devx/tools/ansible_checks/no_state_absent_on_db.py b/src/devx/tools/ansible_checks/no_state_absent_on_db.py new file mode 100644 index 0000000..c53e131 --- /dev/null +++ b/src/devx/tools/ansible_checks/no_state_absent_on_db.py @@ -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 diff --git a/src/devx/tools/ansible_checks/patterns.py b/src/devx/tools/ansible_checks/patterns.py new file mode 100644 index 0000000..cff9535 --- /dev/null +++ b/src/devx/tools/ansible_checks/patterns.py @@ -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", "") + 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 diff --git a/src/devx/tools/ansible_checks/set_fact_to_json.py b/src/devx/tools/ansible_checks/set_fact_to_json.py new file mode 100644 index 0000000..7ea1587 --- /dev/null +++ b/src/devx/tools/ansible_checks/set_fact_to_json.py @@ -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 diff --git a/src/devx/tools/check_ansible_no_log.py b/src/devx/tools/check_ansible_no_log.py index c2af05e..609e163 100644 --- a/src/devx/tools/check_ansible_no_log.py +++ b/src/devx/tools/check_ansible_no_log.py @@ -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", "") - 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") diff --git a/src/devx/tools/check_ansible_no_state_absent_on_db.py b/src/devx/tools/check_ansible_no_state_absent_on_db.py index 07a3aa1..7f0a3d4 100644 --- a/src/devx/tools/check_ansible_no_state_absent_on_db.py +++ b/src/devx/tools/check_ansible_no_state_absent_on_db.py @@ -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:") diff --git a/src/devx/tools/check_ansible_patterns.py b/src/devx/tools/check_ansible_patterns.py index e07294d..f2d4e09 100644 --- a/src/devx/tools/check_ansible_patterns.py +++ b/src/devx/tools/check_ansible_patterns.py @@ -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", "") - - # 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:") diff --git a/src/devx/tools/check_ansible_set_fact_to_json.py b/src/devx/tools/check_ansible_set_fact_to_json.py index 6139e79..754e053 100644 --- a/src/devx/tools/check_ansible_set_fact_to_json.py +++ b/src/devx/tools/check_ansible_set_fact_to_json.py @@ -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:") diff --git a/src/devx/tools/check_jinja_expr.py b/src/devx/tools/check_jinja_expr.py index 40b8990..d3e3244 100644 --- a/src/devx/tools/check_jinja_expr.py +++ b/src/devx/tools/check_jinja_expr.py @@ -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:") diff --git a/src/devx/translations.json b/src/devx/translations.json index 005f469..e45aed2 100644 --- a/src/devx/translations.json +++ b/src/devx/translations.json @@ -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", diff --git a/tests/unit/test_ansible_checks_shared.py b/tests/unit/test_ansible_checks_shared.py new file mode 100644 index 0000000..5304c1d --- /dev/null +++ b/tests/unit/test_ansible_checks_shared.py @@ -0,0 +1,231 @@ +"""Unit tests for devx.tools.ansible_checks._shared.""" + +from pathlib import Path + +import pytest + +from devx.tools.ansible_checks._shared import ( + DEFAULT_ANSIBLE_DIRS, + AnsibleFileFinder, + AnsibleYAMLParser, + ViolationReporter, +) + + +class TestAnsibleFileFinder: + def test_find_task_files_single_yaml(self, tmp_path: Path) -> None: + f = tmp_path / "test.yml" + f.write_text("tasks: []") + assert AnsibleFileFinder.find_task_files(f) == [f] + + def test_find_task_files_single_non_yaml(self, tmp_path: Path) -> None: + f = tmp_path / "test.txt" + f.write_text("hello") + assert AnsibleFileFinder.find_task_files(f) == [] + + def test_find_task_files_dir(self, tmp_path: Path) -> None: + (tmp_path / "a.yml").write_text("tasks: []") + (tmp_path / "b.yaml").write_text("tasks: []") + (tmp_path / "c.txt").write_text("hello") + result = AnsibleFileFinder.find_task_files(tmp_path) + assert len(result) == 2 + assert all(f.suffix in (".yml", ".yaml") for f in result) + + def test_find_task_files_skip_molecule(self, tmp_path: Path) -> None: + (tmp_path / "a.yml").write_text("tasks: []") + mol = tmp_path / "molecule" / "default" + mol.mkdir(parents=True) + (mol / "main.yml").write_text("tasks: []") + result = AnsibleFileFinder.find_task_files(tmp_path, skip_molecule=True) + assert len(result) == 1 + assert "molecule" not in result[0].parts + + def test_find_task_files_include_molecule(self, tmp_path: Path) -> None: + (tmp_path / "a.yml").write_text("tasks: []") + mol = tmp_path / "molecule" / "default" + mol.mkdir(parents=True) + (mol / "main.yml").write_text("tasks: []") + result = AnsibleFileFinder.find_task_files(tmp_path, skip_molecule=False) + assert len(result) == 2 + + def test_find_task_files_nonexistent(self, tmp_path: Path) -> None: + assert AnsibleFileFinder.find_task_files(tmp_path / "nonexistent") == [] + + def test_find_yaml_files_single_file(self, tmp_path: Path) -> None: + f = tmp_path / "test.txt" + f.write_text("hello") + # find_yaml_files accepts any single file (no suffix check) + assert AnsibleFileFinder.find_yaml_files(f) == [f] + + def test_find_yaml_files_dir(self, tmp_path: Path) -> None: + (tmp_path / "a.yml").write_text("tasks: []") + (tmp_path / "sub").mkdir() + (tmp_path / "sub" / "b.yaml").write_text("tasks: []") + result = AnsibleFileFinder.find_yaml_files(tmp_path) + assert len(result) == 2 + + def test_find_yaml_files_skip_molecule(self, tmp_path: Path) -> None: + (tmp_path / "a.yml").write_text("tasks: []") + mol = tmp_path / "molecule" / "default" + mol.mkdir(parents=True) + (mol / "main.yml").write_text("tasks: []") + result = AnsibleFileFinder.find_yaml_files(tmp_path, skip_molecule=True) + assert len(result) == 1 + + def test_find_yaml_files_include_molecule(self, tmp_path: Path) -> None: + (tmp_path / "a.yml").write_text("tasks: []") + mol = tmp_path / "molecule" / "default" + mol.mkdir(parents=True) + (mol / "main.yml").write_text("tasks: []") + result = AnsibleFileFinder.find_yaml_files(tmp_path, skip_molecule=False) + assert len(result) == 2 + + def test_find_task_and_playbook_files(self, tmp_path: Path) -> None: + role = tmp_path / "roles" / "myrole" + (role / "tasks").mkdir(parents=True) + (role / "tasks" / "main.yml").write_text("tasks: []") + pb = tmp_path / "playbooks" + pb.mkdir() + (pb / "deploy.yml").write_text("tasks: []") + (tmp_path / "random.yml").write_text("tasks: []") + result = AnsibleFileFinder.find_task_and_playbook_files(tmp_path) + # Should find tasks/main.yml and playbooks/deploy.yml, not random.yml + names = [f.name for f in result] + assert "main.yml" in names + assert "deploy.yml" in names + assert "random.yml" not in names + + def test_find_task_and_playbook_files_skip_molecule(self, tmp_path: Path) -> None: + role = tmp_path / "roles" / "myrole" + (role / "tasks").mkdir(parents=True) + (role / "tasks" / "main.yml").write_text("tasks: []") + mol = role / "molecule" / "default" / "tasks" + mol.mkdir(parents=True) + (mol / "main.yml").write_text("tasks: []") + result = AnsibleFileFinder.find_task_and_playbook_files(tmp_path, skip_molecule=True) + assert len(result) == 1 + assert "molecule" not in result[0].parts + + +class TestAnsibleYAMLParser: + def test_parse_file_valid(self) -> None: + content = "---\n- name: test\n shell: echo hi\n" + docs = AnsibleYAMLParser.parse_file(content) + assert len(docs) == 1 + assert isinstance(docs[0], list) + + def test_parse_file_multi_doc(self) -> None: + content = "---\n- a\n---\n- b\n" + docs = AnsibleYAMLParser.parse_file(content) + assert len(docs) == 2 + + def test_parse_file_empty_docs_filtered(self) -> None: + content = "---\n- a\n---\n\n" + docs = AnsibleYAMLParser.parse_file(content) + assert len(docs) == 1 + + def test_parse_file_yaml_error(self) -> None: + content = "{{ invalid: [" + docs = AnsibleYAMLParser.parse_file(content) + assert docs == [] + + def test_iter_tasks_bare_list(self) -> None: + doc = [{"name": "task1", "shell": "echo hi"}, {"name": "task2", "shell": "echo bye"}] + tasks = list(AnsibleYAMLParser.iter_tasks(doc)) + assert len(tasks) == 2 + assert tasks[0][0]["name"] == "task1" + assert tasks[0][1] == 1 + assert tasks[1][1] == 2 + + def test_iter_tasks_play_dict(self) -> None: + doc = {"hosts": "all", "tasks": [{"name": "task1", "shell": "echo hi"}]} + tasks = list(AnsibleYAMLParser.iter_tasks(doc)) + assert len(tasks) == 1 + assert tasks[0][0]["name"] == "task1" + + def test_iter_tasks_play_with_pre_post_handlers(self) -> None: + doc = { + "hosts": "all", + "pre_tasks": [{"name": "pre", "shell": "echo pre"}], + "tasks": [{"name": "main", "shell": "echo main"}], + "post_tasks": [{"name": "post", "shell": "echo post"}], + "handlers": [{"name": "handler", "shell": "echo handler"}], + } + tasks = list(AnsibleYAMLParser.iter_tasks(doc)) + assert len(tasks) == 4 + names = [t[0]["name"] for t in tasks] + # Order: tasks, pre_tasks, post_tasks, handlers (as defined in _iter_play_sections) + assert names == ["main", "pre", "post", "handler"] + + def test_iter_tasks_block(self) -> None: + doc = [{"name": "outer", "block": [{"name": "inner", "shell": "echo hi"}]}] + tasks = list(AnsibleYAMLParser.iter_tasks(doc)) + # outer is not a play (no task sections) → yielded as bare task + # inner is yielded from block + assert len(tasks) == 2 + assert tasks[0][0]["name"] == "outer" + assert tasks[1][0]["name"] == "inner" + + def test_iter_tasks_block_in_play_section(self) -> None: + """Block tasks within a play's tasks section are yielded.""" + doc = { + "hosts": "all", + "tasks": [ + {"name": "outer", "block": [{"name": "inner", "shell": "echo hi"}]}, + ], + } + tasks = list(AnsibleYAMLParser.iter_tasks(doc)) + assert len(tasks) == 2 + assert tasks[0][0]["name"] == "outer" + assert tasks[1][0]["name"] == "inner" + + def test_iter_tasks_play_list(self) -> None: + doc = [{"hosts": "all", "tasks": [{"name": "task1", "shell": "echo hi"}]}] + tasks = list(AnsibleYAMLParser.iter_tasks(doc)) + assert len(tasks) == 1 + assert tasks[0][0]["name"] == "task1" + + def test_iter_tasks_non_dict_items_skipped(self) -> None: + doc = ["string", 42, {"name": "task1", "shell": "echo hi"}] + tasks = list(AnsibleYAMLParser.iter_tasks(doc)) + assert len(tasks) == 1 + + +class TestViolationReporter: + def test_format_violation_with_line(self, tmp_path: Path) -> None: + result = ViolationReporter.format_violation(tmp_path / "foo.yml", tmp_path, 42, "bad") + assert result == "foo.yml:42 — bad" + + def test_format_violation_without_line(self, tmp_path: Path) -> None: + result = ViolationReporter.format_violation(tmp_path / "foo.yml", tmp_path, None, "bad") + assert result == "foo.yml — bad" + + def test_format_violation_not_relative(self, tmp_path: Path) -> None: + other = Path("/other/path") + result = ViolationReporter.format_violation(other, tmp_path, 1, "bad") + assert str(other) in result + assert "bad" in result + + def test_report_no_violations(self, capsys: pytest.CaptureFixture[str]) -> None: + ViolationReporter.report([], "test-tool") + captured = capsys.readouterr() + assert "OK" in captured.out + assert "test-tool" in captured.out + + def test_report_with_violations(self, capsys: pytest.CaptureFixture[str]) -> None: + with pytest.raises(SystemExit) as exc_info: + ViolationReporter.report(["v1", "v2"], "test-tool") + assert exc_info.value.code == 1 + captured = capsys.readouterr() + assert "FAIL" in captured.out + assert "v1" in captured.out + assert "v2" in captured.out + + +class TestDefaultAnsibleDirs: + def test_is_tuple(self) -> None: + assert isinstance(DEFAULT_ANSIBLE_DIRS, tuple) + + def test_contains_expected(self) -> None: + assert "ansible/roles" in DEFAULT_ANSIBLE_DIRS + assert "ansible/playbooks" in DEFAULT_ANSIBLE_DIRS diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index ab5de0a..4d6331a 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -150,6 +150,13 @@ class TestCiCommands: assert result.exit_code == 0 mock_run.assert_called_once_with("devx.ci.validate_commit_msg", ["msg"]) + @patch("devx.cli._run_module") + def test_ci_wait_for_checks(self, mock_run: MagicMock) -> None: + runner = CliRunner() + result = runner.invoke(cli, ["ci", "wait-for-checks", "--", "--job-name", "molecule-tests"]) + assert result.exit_code == 0 + mock_run.assert_called_once_with("devx.ci.wait_for_checks", ["--job-name", "molecule-tests"]) + class TestToolsCommands: @patch("devx.cli._run_module") diff --git a/tests/unit/test_discover_runners.py b/tests/unit/test_discover_runners.py index 3fd4b38..4b1b8c2 100644 --- a/tests/unit/test_discover_runners.py +++ b/tests/unit/test_discover_runners.py @@ -1,4 +1,9 @@ -"""Unit tests for scripts/ci/discover_runners.py.""" +"""Unit tests for devx.ci.discover_runners (deprecated wrapper). + +The wrapper re-exports from devx.molecule.discover_runners; these tests +verify backward compatibility by importing through the wrapper and +patching the canonical implementation's requests module. +""" import json from pathlib import Path @@ -32,7 +37,7 @@ class TestGenerateIndices: class TestQueryRunners: - @patch("devx.ci.discover_runners.requests.get") + @patch("devx.molecule.discover_runners.requests.get") def test_returns_total_from_all_levels(self, mock_get: MagicMock) -> None: """Runners from repo, org, and admin levels are summed.""" responses = [ @@ -44,7 +49,7 @@ class TestQueryRunners: result = query_runners("https://api.example.com", "token", "owner", "repo") assert result == 6 - @patch("devx.ci.discover_runners.requests.get") + @patch("devx.molecule.discover_runners.requests.get") def test_skips_non_200(self, mock_get: MagicMock) -> None: """Non-200 responses (e.g., 403 for admin) are skipped.""" responses = [ @@ -56,7 +61,7 @@ class TestQueryRunners: result = query_runners("https://api.example.com", "token", "owner", "repo") assert result == 3 - @patch("devx.ci.discover_runners.requests.get") + @patch("devx.molecule.discover_runners.requests.get") def test_handles_request_exception(self, mock_get: MagicMock) -> None: """Network errors are caught and don't crash.""" mock_get.side_effect = [ @@ -67,7 +72,7 @@ class TestQueryRunners: result = query_runners("https://api.example.com", "token", "owner", "repo") assert result == 3 - @patch("devx.ci.discover_runners.requests.get") + @patch("devx.molecule.discover_runners.requests.get") def test_all_failures_return_zero(self, mock_get: MagicMock) -> None: """When all API calls fail, returns 0.""" mock_get.side_effect = [ @@ -78,7 +83,7 @@ class TestQueryRunners: result = query_runners("https://api.example.com", "token", "owner", "repo") assert result == 0 - @patch("devx.ci.discover_runners.requests.get") + @patch("devx.molecule.discover_runners.requests.get") def test_value_error_on_repo_level(self, mock_get: MagicMock) -> None: """JSON parse error on repo level is caught.""" responses = [ @@ -90,7 +95,7 @@ class TestQueryRunners: result = query_runners("https://api.example.com", "token", "owner", "repo") assert result == 3 - @patch("devx.ci.discover_runners.requests.get") + @patch("devx.molecule.discover_runners.requests.get") def test_value_error_on_org_level(self, mock_get: MagicMock) -> None: """JSON parse error on org level is caught.""" responses = [ @@ -102,7 +107,7 @@ class TestQueryRunners: result = query_runners("https://api.example.com", "token", "owner", "repo") assert result == 3 - @patch("devx.ci.discover_runners.requests.get") + @patch("devx.molecule.discover_runners.requests.get") def test_value_error_on_admin_level(self, mock_get: MagicMock) -> None: """JSON parse error on admin level is caught.""" responses = [ @@ -114,14 +119,14 @@ class TestQueryRunners: result = query_runners("https://api.example.com", "token", "owner", "repo") assert result == 3 - @patch("devx.ci.discover_runners.requests.get") + @patch("devx.molecule.discover_runners.requests.get") def test_request_exception_on_all_levels(self, mock_get: MagicMock) -> None: """Network errors on all levels return 0.""" mock_get.side_effect = __import__("requests").RequestException("network error") result = query_runners("https://api.example.com", "token", "owner", "repo") assert result == 0 - @patch("devx.ci.discover_runners.requests.get") + @patch("devx.molecule.discover_runners.requests.get") def test_query_runners_403_no_warning(self, mock_get: MagicMock, capsys: pytest.CaptureFixture[str]) -> None: """403 on instance-level runners should not produce a warning (expected without admin scope).""" responses = [ @@ -135,7 +140,7 @@ class TestQueryRunners: captured = capsys.readouterr() assert "instance-level" not in captured.err - @patch("devx.ci.discover_runners.requests.get") + @patch("devx.molecule.discover_runners.requests.get") def test_instance_level_non_403_warns(self, mock_get: MagicMock, capsys: pytest.CaptureFixture[str]) -> None: """Non-200, non-403 status on instance-level runners should produce a warning.""" responses = [ @@ -152,37 +157,37 @@ class TestQueryRunners: class TestGetRunnerCount: - @patch("devx.ci.discover_runners.query_runners", return_value=5) + @patch("devx.molecule.discover_runners.query_runners", return_value=5) def test_uses_api_count_when_positive(self, mock_query: MagicMock) -> None: result = get_runner_count("https://api.example.com", "token", "owner", "repo") assert result == 5 - @patch("devx.ci.discover_runners.query_runners", return_value=0) + @patch("devx.molecule.discover_runners.query_runners", return_value=0) @patch.dict("os.environ", {"MOLECULE_RUNNERS": "4"}) def test_falls_back_to_env_var(self, mock_query: MagicMock) -> None: result = get_runner_count("https://api.example.com", "token", "owner", "repo") assert result == 4 - @patch("devx.ci.discover_runners.query_runners", return_value=0) + @patch("devx.molecule.discover_runners.query_runners", return_value=0) @patch.dict("os.environ", {"MOLECULE_RUNNERS": "invalid"}) def test_falls_back_to_default_on_invalid_env(self, mock_query: MagicMock) -> None: result = get_runner_count("https://api.example.com", "token", "owner", "repo") assert result == DEFAULT_MAX_RUNNERS - @patch("devx.ci.discover_runners.query_runners", return_value=0) + @patch("devx.molecule.discover_runners.query_runners", return_value=0) @patch.dict("os.environ", {}, clear=True) def test_falls_back_to_default_when_no_env(self, mock_query: MagicMock) -> None: result = get_runner_count("https://api.example.com", "token", "owner", "repo") assert result == DEFAULT_MAX_RUNNERS - @patch("devx.ci.discover_runners.query_runners", return_value=0) + @patch("devx.molecule.discover_runners.query_runners", return_value=0) @patch.dict("os.environ", {"MOLECULE_RUNNERS": "0"}) def test_env_var_zero_falls_back_to_default(self, mock_query: MagicMock) -> None: """MOLECULE_RUNNERS=0 is invalid, falls back to default.""" result = get_runner_count("https://api.example.com", "token", "owner", "repo") assert result == DEFAULT_MAX_RUNNERS - @patch("devx.ci.discover_runners.query_runners", return_value=0) + @patch("devx.molecule.discover_runners.query_runners", return_value=0) @patch.dict("os.environ", {}, clear=True) def test_no_token_uses_env_var(self, mock_query: MagicMock) -> None: """When no token, skips API and uses env/default.""" @@ -192,7 +197,7 @@ class TestGetRunnerCount: class TestMain: - @patch("devx.ci.discover_runners.get_runner_count", return_value=3) + @patch("devx.molecule.discover_runners.get_runner_count", return_value=3) def test_default_output(self, mock_count: MagicMock) -> None: runner = CliRunner() result = runner.invoke(main, []) @@ -200,28 +205,28 @@ class TestMain: assert "count=3" in result.output assert 'indices=["1", "2", "3"]' in result.output - @patch("devx.ci.discover_runners.get_runner_count", return_value=5) + @patch("devx.molecule.discover_runners.get_runner_count", return_value=5) def test_count_only(self, mock_count: MagicMock) -> None: runner = CliRunner() result = runner.invoke(main, ["--count"]) assert result.exit_code == 0 assert result.output.strip() == "5" - @patch("devx.ci.discover_runners.get_runner_count", return_value=4) + @patch("devx.molecule.discover_runners.get_runner_count", return_value=4) def test_indices_only(self, mock_count: MagicMock) -> None: runner = CliRunner() result = runner.invoke(main, ["--indices"]) assert result.exit_code == 0 assert json.loads(result.output.strip()) == ["1", "2", "3", "4"] - @patch("devx.ci.discover_runners.get_runner_count", return_value=1) + @patch("devx.molecule.discover_runners.get_runner_count", return_value=1) def test_single_runner(self, mock_count: MagicMock) -> None: runner = CliRunner() result = runner.invoke(main, ["--indices"]) assert result.exit_code == 0 assert json.loads(result.output.strip()) == ["1"] - @patch("devx.ci.discover_runners.get_runner_count", return_value=3) + @patch("devx.molecule.discover_runners.get_runner_count", return_value=3) def test_github_output(self, mock_count: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: gh_file = tmp_path / "output.txt" monkeypatch.setenv("GITHUB_OUTPUT", str(gh_file)) @@ -232,14 +237,14 @@ class TestMain: assert "runner-count=3" in content assert "runner-indices=" in content - @patch("devx.ci.discover_runners.get_runner_count", return_value=3) + @patch("devx.molecule.discover_runners.get_runner_count", return_value=3) def test_github_output_no_env(self, mock_count: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delenv("GITHUB_OUTPUT", raising=False) runner = CliRunner() result = runner.invoke(main, ["--github-output"]) assert result.exit_code != 0 - @patch("devx.ci.discover_runners.get_runner_count", return_value=2) + @patch("devx.molecule.discover_runners.get_runner_count", return_value=2) def test_explicit_owner_and_repo(self, mock_count: MagicMock) -> None: """When --owner and --repo are provided, env vars are not used.""" runner = CliRunner() @@ -251,8 +256,8 @@ class TestMain: assert "myorg" in args assert "myrepo" in args - @patch("devx.ci.discover_runners.get_ci_token", side_effect=click.ClickException("no token")) - @patch("devx.ci.discover_runners.get_runner_count", return_value=3) + @patch("devx.molecule.discover_runners.get_ci_token", side_effect=click.ClickException("no token")) + @patch("devx.molecule.discover_runners.get_runner_count", return_value=3) def test_missing_token_runs_without_api(self, mock_count: MagicMock, mock_token: MagicMock) -> None: """When no token is available, runner discovery falls back to env/default.""" runner = CliRunner() @@ -261,3 +266,29 @@ class TestMain: assert result.output.strip() == "3" args, _ = mock_count.call_args assert args[1] is None # token passed as None when missing + + +class TestDeprecationWrapper: + def test_re_exports_canonical_symbols(self) -> None: + """The wrapper re-exports the canonical implementation's symbols.""" + from devx.ci import discover_runners as ci_mod + from devx.molecule import discover_runners as mol_mod + + assert ci_mod.query_runners is mol_mod.query_runners + assert ci_mod.get_runner_count is mol_mod.get_runner_count + assert ci_mod.generate_indices is mol_mod.generate_indices + assert ci_mod.main is mol_mod.main + assert ci_mod.DEFAULT_MAX_RUNNERS is mol_mod.DEFAULT_MAX_RUNNERS + + def test_emit_deprecation_warning(self) -> None: + """_emit_deprecation_warning issues a DeprecationWarning.""" + import warnings + + from devx.ci.discover_runners import _emit_deprecation_warning + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + _emit_deprecation_warning() + assert len(caught) == 1 + assert issubclass(caught[0].category, DeprecationWarning) + assert "deprecated" in str(caught[0].message) diff --git a/tests/unit/test_tools_check_ansible_set_fact_to_json.py b/tests/unit/test_tools_check_ansible_set_fact_to_json.py index 024d619..fe9a223 100644 --- a/tests/unit/test_tools_check_ansible_set_fact_to_json.py +++ b/tests/unit/test_tools_check_ansible_set_fact_to_json.py @@ -268,6 +268,18 @@ class TestCheckFile: errors = _check_file(f, tmp_path) assert len(errors) == 2 + def test_check_file_os_error(self, tmp_path: Path, monkeypatch) -> None: + """OSError reading a file returns empty errors (not a crash).""" + f = tmp_path / "playbook.yml" + f.write_text("- name: ok\n set_fact:\n x: 1\n") + + def _raise(*args, **kwargs): + raise OSError("disk error") + + monkeypatch.setattr(Path, "read_text", _raise) + errors = _check_file(f, tmp_path) + assert errors == [] + class TestMain: def test_passes_when_clean(self, tmp_path: Path): diff --git a/tests/unit/test_wait_for_checks.py b/tests/unit/test_wait_for_checks.py new file mode 100644 index 0000000..e67a1d8 --- /dev/null +++ b/tests/unit/test_wait_for_checks.py @@ -0,0 +1,272 @@ +"""Unit tests for devx.ci.wait_for_checks.""" + +from unittest.mock import MagicMock, patch + +import pytest +from click.testing import CliRunner + +from devx.ci.wait_for_checks import ( + main, + poll_until_complete, + query_job_status, +) + + +def _mock_response(status_code: int = 200, json_data: object | None = None) -> MagicMock: + m = MagicMock() + m.status_code = status_code + if json_data is None: + m.json.side_effect = ValueError("no json") + else: + m.json.return_value = json_data + return m + + +class TestQueryJobStatus: + @patch("devx.ci.wait_for_checks.requests.get") + def test_returns_matching_jobs(self, mock_get: MagicMock) -> None: + """Jobs whose name starts with the prefix are returned.""" + mock_get.side_effect = [ + _mock_response(200, [{"id": 1}, {"id": 2}]), + _mock_response(200, [{"name": "molecule-tests (1)", "status": "completed", "conclusion": "success"}]), + _mock_response(200, [{"name": "other-job", "status": "completed", "conclusion": "success"}]), + ] + result = query_job_status("https://api", "tok", "o/r", "molecule-tests") + assert len(result) == 1 + assert result[0]["name"] == "molecule-tests (1)" + assert result[0]["status"] == "completed" + assert result[0]["conclusion"] == "success" + + @patch("devx.ci.wait_for_checks.requests.get") + def test_no_matching_jobs(self, mock_get: MagicMock) -> None: + """When no job names match the prefix, returns empty list.""" + mock_get.side_effect = [ + _mock_response(200, [{"id": 1}]), + _mock_response(200, [{"name": "other-job", "status": "completed", "conclusion": "success"}]), + ] + result = query_job_status("https://api", "tok", "o/r", "molecule-tests") + assert result == [] + + @patch("devx.ci.wait_for_checks.requests.get") + def test_api_error_returns_empty(self, mock_get: MagicMock) -> None: + """Network errors on the runs endpoint return an empty list.""" + import requests + + mock_get.side_effect = requests.ConnectionError("down") + result = query_job_status("https://api", "tok", "o/r", "molecule-tests") + assert result == [] + + @patch("devx.ci.wait_for_checks.requests.get") + def test_non_200_returns_empty(self, mock_get: MagicMock) -> None: + """Non-200 on runs endpoint returns empty list.""" + mock_get.side_effect = [_mock_response(500, {"message": "err"})] + result = query_job_status("https://api", "tok", "o/r", "molecule-tests") + assert result == [] + + @patch("devx.ci.wait_for_checks.requests.get") + def test_jobs_as_dict_with_jobs_key(self, mock_get: MagicMock) -> None: + """Jobs endpoint returning {'jobs': [...]} dict is handled.""" + mock_get.side_effect = [ + _mock_response(200, [{"id": 1}]), + _mock_response( + 200, {"jobs": [{"name": "molecule-tests (1)", "status": "in_progress", "conclusion": None}]} + ), + ] + result = query_job_status("https://api", "tok", "o/r", "molecule-tests") + assert len(result) == 1 + assert result[0]["status"] == "in_progress" + + @patch("devx.ci.wait_for_checks.requests.get") + def test_runs_as_dict_with_runs_key(self, mock_get: MagicMock) -> None: + """Runs endpoint returning {'runs': [...]} dict is handled.""" + mock_get.side_effect = [ + _mock_response(200, {"runs": [{"id": 1}]}), + _mock_response(200, [{"name": "molecule-tests (1)", "status": "completed", "conclusion": "success"}]), + ] + result = query_job_status("https://api", "tok", "o/r", "molecule-tests") + assert len(result) == 1 + + @patch("devx.ci.wait_for_checks.requests.get") + def test_jobs_endpoint_error_skips_run(self, mock_get: MagicMock) -> None: + """A failed jobs query for one run doesn't abort the whole call.""" + mock_get.side_effect = [ + _mock_response(200, [{"id": 1}, {"id": 2}]), + _mock_response(500, {"message": "err"}), + _mock_response(200, [{"name": "molecule-tests (1)", "status": "completed", "conclusion": "success"}]), + ] + result = query_job_status("https://api", "tok", "o/r", "molecule-tests") + assert len(result) == 1 + + @patch("devx.ci.wait_for_checks.requests.get") + def test_run_without_id_skipped(self, mock_get: MagicMock) -> None: + """Runs missing an 'id' field are skipped.""" + mock_get.side_effect = [ + _mock_response(200, [{"foo": "bar"}, {"id": 1}]), + _mock_response(200, [{"name": "molecule-tests (1)", "status": "completed", "conclusion": "success"}]), + ] + result = query_job_status("https://api", "tok", "o/r", "molecule-tests") + assert len(result) == 1 + + @patch("devx.ci.wait_for_checks.requests.get") + def test_jobs_query_exception_skips_run(self, mock_get: MagicMock) -> None: + """A ConnectionError on the jobs endpoint for one run is skipped.""" + import requests + + mock_get.side_effect = [ + _mock_response(200, [{"id": 1}, {"id": 2}]), + requests.ConnectionError("down"), + _mock_response(200, [{"name": "molecule-tests (1)", "status": "completed", "conclusion": "success"}]), + ] + result = query_job_status("https://api", "tok", "o/r", "molecule-tests") + assert len(result) == 1 + + @patch("devx.ci.wait_for_checks.requests.get") + def test_jobs_json_value_error_skips_run(self, mock_get: MagicMock) -> None: + """A ValueError (bad JSON) on the jobs endpoint is skipped.""" + mock_get.side_effect = [ + _mock_response(200, [{"id": 1}]), + _mock_response(200), # json raises ValueError by default + ] + result = query_job_status("https://api", "tok", "o/r", "molecule-tests") + assert result == [] + + +class TestPollUntilComplete: + @patch("devx.ci.wait_for_checks.time.sleep") + @patch("devx.ci.wait_for_checks.time.monotonic") + @patch("devx.ci.wait_for_checks.query_job_status") + def test_all_jobs_succeed(self, mock_query: MagicMock, mock_mono: MagicMock, mock_sleep: MagicMock) -> None: + """All jobs completed with success → returns 0.""" + mock_query.return_value = [{"name": "molecule-tests (1)", "status": "completed", "conclusion": "success"}] + mock_mono.side_effect = [0.0, 0.0] + code = poll_until_complete("https://api", "tok", "o/r", "molecule-tests", timeout=100, interval=10) + assert code == 0 + + @patch("devx.ci.wait_for_checks.time.sleep") + @patch("devx.ci.wait_for_checks.time.monotonic") + @patch("devx.ci.wait_for_checks.query_job_status") + def test_job_fails(self, mock_query: MagicMock, mock_mono: MagicMock, mock_sleep: MagicMock) -> None: + """A job with non-success conclusion → returns 1.""" + mock_query.return_value = [{"name": "molecule-tests (1)", "status": "completed", "conclusion": "failure"}] + mock_mono.side_effect = [0.0, 0.0] + code = poll_until_complete("https://api", "tok", "o/r", "molecule-tests", timeout=100, interval=10) + assert code == 1 + + @patch("devx.ci.wait_for_checks.time.sleep") + @patch("devx.ci.wait_for_checks.time.monotonic") + @patch("devx.ci.wait_for_checks.query_job_status") + def test_no_require_success(self, mock_query: MagicMock, mock_mono: MagicMock, mock_sleep: MagicMock) -> None: + """With require_success=False, a failed job returns 0.""" + mock_query.return_value = [{"name": "molecule-tests (1)", "status": "completed", "conclusion": "failure"}] + mock_mono.side_effect = [0.0, 0.0] + code = poll_until_complete( + "https://api", "tok", "o/r", "molecule-tests", timeout=100, interval=10, require_success=False + ) + assert code == 0 + + @patch("devx.ci.wait_for_checks.time.sleep") + @patch("devx.ci.wait_for_checks.time.monotonic") + @patch("devx.ci.wait_for_checks.query_job_status") + def test_timeout(self, mock_query: MagicMock, mock_mono: MagicMock, mock_sleep: MagicMock) -> None: + """Jobs never complete → returns 2 after timeout.""" + mock_query.return_value = [{"name": "molecule-tests (1)", "status": "in_progress", "conclusion": None}] + # monotonic calls: deadline=0, while-check=0 (enter), sleep-calc=0, while-check=200 (exit) + mock_mono.side_effect = [0.0, 0.0, 0.0, 200.0] + code = poll_until_complete("https://api", "tok", "o/r", "molecule-tests", timeout=100, interval=10) + assert code == 2 + + @patch("devx.ci.wait_for_checks.time.sleep") + @patch("devx.ci.wait_for_checks.time.monotonic") + @patch("devx.ci.wait_for_checks.query_job_status") + def test_no_jobs_found_timeout(self, mock_query: MagicMock, mock_mono: MagicMock, mock_sleep: MagicMock) -> None: + """No matching jobs at all → returns 3.""" + mock_query.return_value = [] + # monotonic calls: deadline=0, while-check=0 (enter), sleep-calc=0, while-check=200 (exit) + mock_mono.side_effect = [0.0, 0.0, 0.0, 200.0] + code = poll_until_complete("https://api", "tok", "o/r", "molecule-tests", timeout=100, interval=10) + assert code == 3 + + @patch("devx.ci.wait_for_checks.time.sleep") + @patch("devx.ci.wait_for_checks.time.monotonic") + @patch("devx.ci.wait_for_checks.query_job_status") + def test_in_progress_then_success(self, mock_query: MagicMock, mock_mono: MagicMock, mock_sleep: MagicMock) -> None: + """First poll in_progress, second poll success → returns 0.""" + mock_query.side_effect = [ + [{"name": "molecule-tests (1)", "status": "in_progress", "conclusion": None}], + [{"name": "molecule-tests (1)", "status": "completed", "conclusion": "success"}], + ] + # monotonic: deadline=0, while=0 (enter), sleep-calc=0, while=5 (enter), success→return + mock_mono.side_effect = [0.0, 0.0, 0.0, 5.0] + code = poll_until_complete("https://api", "tok", "o/r", "molecule-tests", timeout=100, interval=10) + assert code == 0 + + +class TestMain: + @patch("devx.ci.wait_for_checks.get_ci_token", return_value="tok") + @patch("devx.ci.wait_for_checks.poll_until_complete", return_value=0) + def test_success_exit_code(self, mock_poll: MagicMock, mock_token: MagicMock) -> None: + runner = CliRunner() + result = runner.invoke(main, ["--job-name", "molecule-tests", "--repo", "o/r"]) + assert result.exit_code == 0 + + @patch("devx.ci.wait_for_checks.get_ci_token", return_value="tok") + @patch("devx.ci.wait_for_checks.poll_until_complete", return_value=1) + def test_failure_exit_code(self, mock_poll: MagicMock, mock_token: MagicMock) -> None: + runner = CliRunner() + result = runner.invoke(main, ["--job-name", "molecule-tests", "--repo", "o/r"]) + assert result.exit_code == 1 + + @patch("devx.ci.wait_for_checks.get_ci_token", return_value="tok") + @patch("devx.ci.wait_for_checks.poll_until_complete", return_value=2) + def test_timeout_exit_code(self, mock_poll: MagicMock, mock_token: MagicMock) -> None: + runner = CliRunner() + result = runner.invoke(main, ["--job-name", "molecule-tests", "--repo", "o/r"]) + assert result.exit_code == 2 + + @patch("devx.ci.wait_for_checks.get_ci_token", return_value="tok") + @patch("devx.ci.wait_for_checks.poll_until_complete", return_value=3) + def test_api_error_exit_code(self, mock_poll: MagicMock, mock_token: MagicMock) -> None: + runner = CliRunner() + result = runner.invoke(main, ["--job-name", "molecule-tests", "--repo", "o/r"]) + assert result.exit_code == 3 + + def test_missing_job_name(self) -> None: + runner = CliRunner() + result = runner.invoke(main, ["--repo", "o/r"]) + assert result.exit_code != 0 + + def test_missing_repo(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("GITHUB_REPOSITORY", raising=False) + runner = CliRunner() + result = runner.invoke(main, ["--job-name", "molecule-tests"]) + assert result.exit_code != 0 + + @patch("devx.ci.wait_for_checks.get_ci_token", return_value="tok") + @patch("devx.ci.wait_for_checks.poll_until_complete", return_value=0) + def test_repo_from_env(self, mock_poll: MagicMock, mock_token: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("GITHUB_REPOSITORY", "oblachno-oss/grm") + runner = CliRunner() + result = runner.invoke(main, ["--job-name", "molecule-tests"]) + assert result.exit_code == 0 + args, kwargs = mock_poll.call_args + assert args[2] == "oblachno-oss/grm" + + @patch("devx.ci.wait_for_checks.get_ci_token", return_value="tok") + @patch("devx.ci.wait_for_checks.poll_until_complete", return_value=0) + def test_invalid_timeout(self, mock_poll: MagicMock, mock_token: MagicMock) -> None: + runner = CliRunner() + result = runner.invoke(main, ["--job-name", "molecule-tests", "--repo", "o/r", "--timeout", "0"]) + assert result.exit_code != 0 + + @patch("devx.ci.wait_for_checks.get_ci_token", return_value="tok") + @patch("devx.ci.wait_for_checks.poll_until_complete", return_value=0) + def test_invalid_interval(self, mock_poll: MagicMock, mock_token: MagicMock) -> None: + runner = CliRunner() + result = runner.invoke(main, ["--job-name", "molecule-tests", "--repo", "o/r", "--poll-interval", "0"]) + assert result.exit_code != 0 + + @patch("devx.ci.wait_for_checks.get_ci_token", side_effect=__import__("click").ClickException("no token")) + def test_no_token_exit_3(self, mock_token: MagicMock) -> None: + runner = CliRunner() + result = runner.invoke(main, ["--job-name", "molecule-tests", "--repo", "o/r"]) + assert result.exit_code == 3