devx/src/devx/ci/check_workflow_tofu_init.py

"""Check that workflow jobs using tofu state have a tofu-init step.

This prevents the class of bug where a job runs ``tofu output`` or calls
a script that uses tofu state without first running ``tofu init``,
causing "Required plugins are not installed" errors.

The check scans all workflow YAML files for jobs that:
  - Call scripts that use ``tofu output`` (configurable via --state-scripts)
  - Call ``tofu output`` directly
  - Call ``tofu plan`` or ``tofu apply`` directly

For each such job, it verifies the same job has a ``tofu-init`` step,
either:
  - Directly via ``tofu init`` in a step's run command
  - Via ``create_staging_deployment.py --phase tofu-init``
  - Via ``create_production_deployment.py --phase tofu-init``

Usage::

    python -m devx.ci.check_workflow_tofu_init
    python -m devx.ci.check_workflow_tofu_init --workflow .gitea/workflows/deploy.yml

Exit code 0 if all jobs have tofu-init, 1 otherwise.
"""

from __future__ import annotations

import sys
from pathlib import Path

import click
import yaml

REPO_ROOT = Path.cwd()
WORKFLOWS_DIR = REPO_ROOT / ".gitea" / "workflows"

# Scripts that call `tofu output`, `tofu plan`, or `tofu apply` internally.
# If a job calls any of these, it must have a tofu-init step.
# NOTE: destroy_orphans.py reads terraform.tfstate directly from disk
# (does not invoke `tofu output`), so it does NOT need tofu-init.
DEFAULT_TOFU_STATE_SCRIPTS: set[str] = {
    "preflight_deploy.py",
}

# Commands that directly use tofu state (must be preceded by tofu init).
TOFU_STATE_COMMANDS = ("tofu output", "tofu plan", "tofu apply", "tofu show")

# Commands that initialize tofu (counted as tofu-init steps).
TOFU_INIT_COMMANDS = (
    "tofu init",
    "--phase tofu-init",
    "tofu-init",
)


def _check_workflow(filepath: Path, state_scripts: set[str]) -> list[str]:
    """Check a single workflow file for missing tofu-init steps.

    Returns a list of error messages (empty if all OK).
    """
    errors: list[str] = []
    content = filepath.read_text(encoding="utf-8")
    try:
        workflow = yaml.safe_load(content)
    except yaml.YAMLError as exc:
        return [f"{filepath}: cannot parse YAML: {exc}"]

    jobs = workflow.get("jobs", {})
    for job_name, job_def in jobs.items():
        steps = job_def.get("steps", [])
        if not steps:
            continue

        uses_tofu_state = False
        has_tofu_init = False

        for step in steps:
            run_cmd = step.get("run", "")
            if not run_cmd:
                continue
            # Check if this step uses tofu state
            for script in state_scripts:
                if script in run_cmd:
                    uses_tofu_state = True
            for cmd in TOFU_STATE_COMMANDS:
                if cmd in run_cmd:
                    uses_tofu_state = True
            # Check if this step initializes tofu
            for cmd in TOFU_INIT_COMMANDS:
                if cmd in run_cmd:
                    has_tofu_init = True

        if uses_tofu_state and not has_tofu_init:
            errors.append(
                f"{filepath.name}::{job_name}: uses tofu state "
                f"(tofu output/plan/apply or {state_scripts}) "
                f"but has no tofu-init step. Add a step running "
                f"'create_*_deployment.py --phase tofu-init' before "
                f"the first tofu state access."
            )

    return errors


@click.command()
@click.option(
    "--workflow",
    type=click.Path(exists=True, path_type=Path),
    help="Check a specific workflow file (default: all in .gitea/workflows/).",
)
@click.option(
    "--workflows-dir",
    type=click.Path(exists=True, path_type=Path),
    default=None,
    help="Override the workflows directory (default: .gitea/workflows/).",
)
@click.option(
    "--state-script",
    "state_scripts",
    multiple=True,
    default=None,
    help="Add a script name that uses tofu state (can be repeated). Overrides the default list if any are specified.",
)
def main(workflow: Path | None, workflows_dir: Path | None, state_scripts: tuple[str, ...]) -> None:
    """Check that workflow jobs using tofu state have a tofu-init step."""
    scripts = set(state_scripts) if state_scripts else DEFAULT_TOFU_STATE_SCRIPTS
    wdir = workflows_dir or WORKFLOWS_DIR
    files = [workflow] if workflow else sorted(wdir.glob("*.yml"))

    all_errors: list[str] = []
    for f in files:
        errors = _check_workflow(f, scripts)
        all_errors.extend(errors)

    if all_errors:
        click.echo("[check-workflow-tofu-init] FAIL: missing tofu-init steps found:")
        for err in all_errors:
            click.echo(f"  - {err}")
        sys.exit(1)
    else:
        click.echo("[check-workflow-tofu-init] OK: all tofu-state jobs have tofu-init.")


if __name__ == "__main__":  # pragma: no cover
    main()