test

Signed-off-by: Sean Conroy <sconroy@redhat.com>

Author: seanconroy2021

.github/workflows/pr-assigner.yaml

@@ -14,10 +14,11 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Assign reviewers using shared action
-        uses: konflux-ci/release-service-automations/pr-assigner@5da461efb1b29e523d01989ee09758e85099bda8  # main
+        uses: seanconroy2021/release-service-automations/pr-assigner@auto-pr
         with:
           event-type: ${{ github.event.action }}
           pr-number: ${{ github.event.pull_request.number }}
           removed-assignee: ${{ github.event.action == 'unassigned' && github.event.assignee.login || '' }}
           github-token: ${{ secrets.PR_ASSIGNER_PAT_TOKEN }}
           slack-webhook: ${{ secrets.PR_ASSIGNER_SLACK_WEBHOOK }}
+          pto-calendar-url: ${{ secrets.PTO_CALENDAR_URL }}

scripts/validate-taskrunspecs.py

@@ -0,0 +1,381 @@
+#!/usr/bin/env python3
+"""
+Validate taskRunSpecs in a ReleasePlanAdmission against the pipeline it references.
+
+When a pipeline task is renamed or removed, any RPA with taskRunSpecs referencing
+the old name will silently break at runtime. This script catches that drift by
+resolving the pipeline at a given revision and checking that every pipelineTaskName
+and step name in the RPA actually exists.
+
+Steps:
+    1. Parse the RPA YAML to extract revision, pathInRepo, and taskRunSpecs
+    2. Resolve the pipeline definition at the target revision
+    3. For each taskRunSpec entry, verify the pipelineTaskName exists in the pipeline
+    4. For each stepSpec, resolve the task and verify the step name exists
+    5. Report mismatches and exit non-zero on errors
+"""
+
+import argparse
+import logging
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+from typing import Dict, List, Optional, Tuple
+
+import yaml
+
+LOGGER = logging.getLogger("validate-taskrunspecs")
+
+DEFAULT_LOG_FMT = "%(asctime)s [%(levelname)-8s] %(message)s"
+DEFAULT_DATE_FMT = "%Y-%m-%d %H:%M:%S %z"
+
+DEFAULT_REPO_URL = "https://github.com/konflux-ci/release-service-catalog.git"
+
+
+def parse_arguments() -> argparse.Namespace:  # pragma: no cover
+    """Parse CLI arguments.
+
+    Returns:
+        Parsed argument namespace.
+    """
+    parser = argparse.ArgumentParser(
+        description="Validate taskRunSpecs in an RPA against the pipeline it references.",
+    )
+    parser.add_argument(
+        "--rpa",
+        required=True,
+        help="Path to a ReleasePlanAdmission YAML file",
+    )
+    parser.add_argument(
+        "--revision",
+        help="Override the revision extracted from the RPA",
+    )
+    parser.add_argument(
+        "--local",
+        action="store_true",
+        help="Resolve files from the local working tree instead of cloning",
+    )
+    parser.add_argument(
+        "--repo-url",
+        default=DEFAULT_REPO_URL,
+        help="Git repo URL (ignored when --local is set)",
+    )
+    parser.add_argument(
+        "-d",
+        "--debug",
+        action="store_true",
+        help="Enable debug logging",
+    )
+    return parser.parse_args()
+
+
+def load_yaml(path: str) -> dict:
+    """Load a YAML file and return the parsed document.
+
+    Args:
+        path: Filesystem path to the YAML file.
+
+    Returns:
+        Parsed YAML as a dictionary.
+
+    Raises:
+        FileNotFoundError: If the file does not exist.
+        yaml.YAMLError: If the file is not valid YAML.
+    """
+    with open(path) as f:
+        return yaml.safe_load(f)
+
+
+def extract_rpa_fields(rpa: dict) -> Tuple[str, str, List[dict]]:
+    """Extract the revision, pipeline path, and taskRunSpecs from an RPA.
+
+    Args:
+        rpa: Parsed RPA document.
+
+    Returns:
+        Tuple of (revision, pathInRepo, taskRunSpecs list).
+    """
+    pipeline = rpa.get("spec", {}).get("pipeline", {})
+    ref_params = pipeline.get("pipelineRef", {}).get("params", [])
+    params = {p["name"]: p["value"] for p in ref_params}
+
+    revision = params.get("revision", "")
+    path_in_repo = params.get("pathInRepo", "")
+    task_run_specs = pipeline.get("taskRunSpecs", [])
+
+    return revision, path_in_repo, task_run_specs
+
+
+def clone_repo(url: str, revision: str) -> str:
+    """Shallow-clone a git repo at the given revision into a temp directory.
+
+    Tries a direct --branch clone first. Falls back to a default clone followed
+    by fetch + checkout if the revision is a commit SHA rather than a branch/tag.
+
+    Args:
+        url: Git remote URL.
+        revision: Branch name, tag, or commit SHA.
+
+    Returns:
+        Path to the cloned directory.
+
+    Raises:
+        RuntimeError: If the clone fails entirely.
+    """
+    tmpdir = tempfile.mkdtemp(prefix="validate-trs-")
+
+    result = subprocess.run(
+        [
+            "git",
+            "clone",
+            "--depth",
+            "1",
+            "--branch",
+            revision,
+            "--single-branch",
+            "-q",
+            url,
+            tmpdir,
+        ],
+        capture_output=True,
+        text=True,
+    )
+    if result.returncode == 0:
+        LOGGER.debug(f"Cloned {url} at {revision}")
+        return tmpdir
+
+    LOGGER.debug(
+        f"Branch clone failed, falling back to default clone: {result.stderr.strip()}"
+    )
+    shutil.rmtree(tmpdir, ignore_errors=True)
+    tmpdir = tempfile.mkdtemp(prefix="validate-trs-")
+
+    subprocess.run(
+        ["git", "clone", "--depth", "1", "-q", url, tmpdir],
+        check=True,
+        capture_output=True,
+    )
+    subprocess.run(
+        ["git", "fetch", "--depth", "1", "origin", revision],
+        cwd=tmpdir,
+        check=False,
+        capture_output=True,
+    )
+    result = subprocess.run(
+        ["git", "checkout", revision],
+        cwd=tmpdir,
+        check=False,
+        capture_output=True,
+        text=True,
+    )
+    if result.returncode != 0:
+        raise RuntimeError(
+            f"Could not checkout revision '{revision}': {result.stderr.strip()}"
+        )
+
+    return tmpdir
+
+
+def resolve_file(
+    path: str, revision: str, local: bool, clone_dir: Optional[str]
+) -> Optional[str]:
+    """Read a file either from the local tree or from a cloned repo.
+
+    Args:
+        path: Relative path inside the repo.
+        revision: Git revision (unused when local=True).
+        local: If True, read from the current working directory.
+        clone_dir: Path to the cloned repo (used when local=False).
+
+    Returns:
+        File contents as a string, or None if the file does not exist.
+    """
+    if local:
+        if not os.path.isfile(path):
+            LOGGER.debug(f"Local file not found: {path}")
+            return None
+        with open(path) as f:
+            return f.read()
+
+    target = os.path.join(clone_dir, path)
+    if not os.path.isfile(target):
+        LOGGER.debug(f"File not found in clone: {path}")
+        return None
+    with open(target) as f:
+        return f.read()
+
+
+def extract_pipeline_tasks(pipeline_yaml: str) -> Dict[str, str]:
+    """Parse a pipeline definition and return a map of task name to pathInRepo.
+
+    Args:
+        pipeline_yaml: Raw YAML content of the pipeline file.
+
+    Returns:
+        Dict mapping each pipeline task name to its pathInRepo value.
+        Tasks without a pathInRepo get an empty string.
+    """
+    doc = yaml.safe_load(pipeline_yaml)
+    tasks: Dict[str, str] = {}
+    task_blocks = doc.get("spec", {}).get("tasks", [])
+    finally_blocks = doc.get("spec", {}).get("finally", [])
+
+    for block in task_blocks + finally_blocks:
+        name = block.get("name", "")
+        ref = block.get("taskRef", {})
+        if not isinstance(ref, dict):
+            tasks[name] = ""
+            continue
+
+        params = {p["name"]: p["value"] for p in ref.get("params", [])}
+        tasks[name] = params.get("pathInRepo", "")
+
+    return tasks
+
+
+def extract_task_steps(task_yaml: str) -> List[str]:
+    """Parse a task definition and return the list of step names.
+
+    Args:
+        task_yaml: Raw YAML content of the task file.
+
+    Returns:
+        List of step names in order.
+    """
+    doc = yaml.safe_load(task_yaml)
+    return [step["name"] for step in doc.get("spec", {}).get("steps", [])]
+
+
+def validate(
+    rpa_path: str, revision_override: Optional[str], local: bool, repo_url: str
+) -> int:
+    """Run the full validation of taskRunSpecs against the pipeline.
+
+    Args:
+        rpa_path: Path to the RPA YAML file.
+        revision_override: If set, use this instead of the revision in the RPA.
+        local: Resolve files from the local working tree.
+        repo_url: Git repo URL for cloning.
+
+    Returns:
+        0 on success, 1 if any errors were found.
+    """
+    rpa = load_yaml(rpa_path)
+    revision, pipeline_path, task_run_specs = extract_rpa_fields(rpa)
+
+    if revision_override:
+        revision = revision_override
+
+    if not revision or not pipeline_path:
+        LOGGER.error("Could not extract revision or pathInRepo from the RPA")
+        return 1
+
+    has_template_vars = "${" in revision or "${" in pipeline_path
+    if has_template_vars and not local and not revision_override:
+        LOGGER.error(
+            f"RPA contains unresolved template variables (revision='{revision}', "
+            f"path='{pipeline_path}'). Use --local or --revision to override."
+        )
+        return 1
+
+    LOGGER.info(f"Revision: {revision}")
+    LOGGER.info(f"Pipeline: {pipeline_path}")
+
+    clone_dir = None
+    if not local:
+        LOGGER.info(f"Cloning {repo_url} at {revision}")
+        clone_dir = clone_repo(repo_url, revision)
+
+    pipeline_content = resolve_file(pipeline_path, revision, local, clone_dir)
+    if pipeline_content is None:
+        LOGGER.error(f"Pipeline not found: {pipeline_path}")
+        return 1
+
+    pipeline_tasks = extract_pipeline_tasks(pipeline_content)
+    LOGGER.info(f"Pipeline tasks: {', '.join(sorted(pipeline_tasks))}")
+
+    if not task_run_specs:
+        LOGGER.info("No taskRunSpecs in this RPA, nothing to validate")
+        return 0
+
+    LOGGER.info(f"Checking {len(task_run_specs)} taskRunSpec entries")
+
+    errors = 0
+    warnings = 0
+    available_names = ", ".join(sorted(pipeline_tasks))
+
+    for entry in task_run_specs:
+        task_name = entry.get("pipelineTaskName", "")
+
+        if task_name not in pipeline_tasks:
+            LOGGER.error(
+                f"pipelineTaskName '{task_name}' does not exist in pipeline "
+                f"(available: {available_names})"
+            )
+            errors += 1
+            continue
+
+        step_specs = entry.get("stepSpecs", [])
+        if not step_specs:
+            LOGGER.info(f"  {task_name}: ok")
+            continue
+
+        task_path = pipeline_tasks[task_name]
+        if not task_path:
+            LOGGER.warning(f"  {task_name}: task path unknown, cannot verify steps")
+            warnings += 1
+            continue
+
+        task_content = resolve_file(task_path, revision, local, clone_dir)
+        if task_content is None:
+            LOGGER.warning(f"  {task_name}: task file not found at {task_path}")
+            warnings += 1
+            continue
+
+        task_steps = extract_task_steps(task_content)
+        step_errors = 0
+
+        for spec in step_specs:
+            step_name = spec.get("name", "")
+            if step_name not in task_steps:
+                LOGGER.error(
+                    f"  {task_name}: step '{step_name}' does not exist "
+                    f"(available: {', '.join(task_steps)})"
+                )
+                step_errors += 1
+                errors += 1
+
+        if step_errors == 0:
+            LOGGER.info(f"  {task_name}: ok ({len(step_specs)} steps verified)")
+
+    if errors:
+        LOGGER.error(
+            f"Validation failed: {errors} error(s), {warnings} warning(s). "
+            "Update the RPA to match the pipeline at the target revision."
+        )
+    else:
+        LOGGER.info(f"Validation passed: {warnings} warning(s)")
+
+    return 1 if errors else 0
+
+
+def main() -> None:  # pragma: no cover
+    args = parse_arguments()
+
+    log_level = logging.DEBUG if args.debug else logging.INFO
+    stream_handler = logging.StreamHandler(sys.stdout)
+    stream_handler.setLevel(log_level)
+    logging.basicConfig(
+        level=log_level,
+        format=DEFAULT_LOG_FMT,
+        datefmt=DEFAULT_DATE_FMT,
+        handlers=[stream_handler],
+    )
+
+    sys.exit(validate(args.rpa, args.revision, args.local, args.repo_url))
+
+
+if __name__ == "__main__":  # pragma: no cover
+    main()