devx/src/devx/ci/check_workflow_artifact_deps.py

"""Check that workflow jobs downloading artifacts depend on the uploading job.

This prevents the class of bug where a job downloads an artifact produced by
another job but does not declare that job in its ``needs`` list. When both
jobs run in parallel, the download fails because the artifact hasn't been
uploaded yet.

The check scans all workflow YAML files for:
  - ``gitea-upload-artifact`` / ``actions/upload-artifact`` steps
  - ``gitea-download-artifact`` / ``actions/download-artifact`` steps

For each download, it finds the job(s) that upload an artifact with a
matching name and verifies that at least one uploading job is in the
downloading job's ``needs`` list.

Artifact names with ``${{ ... }}`` expressions are matched literally
(both sides use the same expression, so they resolve to the same value
at runtime).

Usage::

    python -m devx.ci.check_workflow_artifact_deps
    python -m devx.ci.check_workflow_artifact_deps --workflow .gitea/workflows/ci.yml

Exit code 0 if all artifact dependencies are satisfied, 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"

UPLOAD_ACTIONS = ("upload-artifact",)
DOWNLOAD_ACTIONS = ("download-artifact",)


def _is_artifact_action(uses: str, action_types: tuple[str, ...]) -> bool:
    """Check if a step's ``uses`` field references an artifact action."""
    if not uses:
        return False
    uses_lower = uses.lower()
    return any(action in uses_lower for action in action_types)


def _extract_artifact_info(workflow: dict) -> tuple[dict[str, list[str]], list[tuple[str, str, str]]]:
    """Extract artifact upload and download info from a workflow.

    Returns:
        uploads: Mapping of artifact_name → list of job names that upload it.
        downloads: List of (job_name, artifact_name, step_name) tuples.
    """
    uploads: dict[str, list[str]] = {}
    downloads: list[tuple[str, str, str]] = []

    jobs = workflow.get("jobs", {})
    for job_name, job_def in jobs.items():
        for step in job_def.get("steps", []):
            uses = step.get("uses", "")
            with_data = step.get("with", {})
            artifact_name = with_data.get("name", "")
            step_name = step.get("name", "")

            if _is_artifact_action(uses, UPLOAD_ACTIONS):
                if artifact_name:
                    uploads.setdefault(artifact_name, []).append(job_name)
            elif _is_artifact_action(uses, DOWNLOAD_ACTIONS) and artifact_name:
                downloads.append((job_name, artifact_name, step_name))

    return uploads, downloads


def _check_workflow(filepath: Path) -> list[str]:
    """Check a single workflow file for missing artifact dependencies.

    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}"]

    if not isinstance(workflow, dict):
        return [f"{filepath}: not a valid workflow (expected dict)"]

    uploads, downloads = _extract_artifact_info(workflow)
    jobs = workflow.get("jobs", {})

    for dl_job, artifact_name, step_name in downloads:
        uploading_jobs = uploads.get(artifact_name, [])
        if not uploading_jobs:
            # Artifact not uploaded in this workflow — may come from an
            # external source (e.g., S3). Skip.
            continue

        dl_job_def = jobs.get(dl_job, {})
        needs_raw = dl_job_def.get("needs", [])
        needs = {needs_raw} if isinstance(needs_raw, str) else set(needs_raw or [])

        # Check if any uploading job is in the download job's needs
        if not any(uploader in needs for uploader in uploading_jobs):
            # Check if the download step has continue-on-error: true
            # (valid guard when the uploading job may be skipped due to
            # Gitea Actions' needs skip behavior — the download will
            # fail gracefully if the artifact doesn't exist).
            dl_steps = dl_job_def.get("steps", [])
            step_def = next((s for s in dl_steps if s.get("name", "") == step_name), {})
            if step_def.get("continue-on-error") is True:
                continue

            uploaders_str = ", ".join(sorted(uploading_jobs))
            errors.append(
                f"{filepath.name}::{dl_job}: step '{step_name}' downloads "
                f"artifact '{artifact_name}' produced by job(s) "
                f"[{uploaders_str}] but none are in its 'needs' list "
                f"(current needs: {sorted(needs) or 'none'}). "
                f"Add the uploading job to 'needs' or guard the download "
                f"with an if: condition checking the upload job's result."
            )

    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/).",
)
def main(workflow: Path | None, workflows_dir: Path | None) -> None:
    """Check that artifact download jobs depend on upload jobs."""
    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)
        all_errors.extend(errors)

    if all_errors:
        click.echo("[check-workflow-artifact-deps] FAIL: missing artifact dependencies found:")
        for err in all_errors:
            click.echo(f"  - {err}")
        sys.exit(1)
    else:
        click.echo("[check-workflow-artifact-deps] OK: all artifact downloads have upload jobs in needs.")


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