feat(ISV-7356): certProject check fallback to base branch on operator removal

ansible/roles/operator-pipeline/templates/openshift/pipelines/operator-hosted-pipeline.yml

@@ -536,6 +536,9 @@ spec:
         - name: source
           workspace: repository
           subPath: src
+        - name: base
+          workspace: repository
+          subPath: base
 
     - name: get-organization
       runAfter:

ansible/roles/operator-pipeline/templates/openshift/pipelines/operator-release-pipeline.yml

@@ -293,6 +293,9 @@ spec:
         - name: source
           workspace: repository
           subPath: src
+        - name: base
+          workspace: repository
+          subPath: base
 
     - name: get-organization
       runAfter:

ansible/roles/operator-pipeline/templates/openshift/tasks/cert-project-check.yml

@@ -20,63 +20,27 @@ spec:
       description: Identifier of certification project from Red Hat Connect
   workspaces:
     - name: source
+    - name: base
+      description: Clone of the repo at head of the base branch
+      optional: true
   steps:
     - name: certification-project-check
       image: "$(params.pipeline_image)"
       workingDir: $(workspaces.source.path)
       script: |
         #! /usr/bin/env bash
         set -xe
-        echo "Checking availability of cert project identifier"
 
-        if [ "$(params.cert_project_required)" != "true" ]; then
-          echo "Cert project ID is not required."
-          echo -n "" | tee $(results.certification_project_id.path)
-          exit 0
+        EXTRA_ARGS=""
+        if [[ "$(workspaces.base.bound)" == "true" ]]; then
+          EXTRA_ARGS+=" --repo-base-path $(workspaces.base.path)"
         fi
 
-        if [[ -z "$(params.affected_operators)" && -z "$(params.affected_catalog_operators)" ]]; then
-          echo "No operator is affected."
-          exit 1
-        fi
-
-        # Create dictionary of all affected operators to serve as a set
-        declare -A ALL_AFFECTED_OPERATORS
-
-        # Add operators from affected_operators
-        if [ -n "$(params.affected_operators)" ]; then
-          IFS=',' read -ra affected_ops_array <<< "$(params.affected_operators)"
-          for op in "${affected_ops_array[@]}"; do
-            ALL_AFFECTED_OPERATORS["$op"]=1
-          done
-        fi
-
-        # Add operators from affected_catalog_operators
-        if [ -n "$(params.affected_catalog_operators)" ]; then
-          IFS=',' read -ra affected_catalog_ops_array <<< "$(params.affected_catalog_operators)"
-          for op in "${affected_catalog_ops_array[@]}"; do
-            operator_name=$(echo "$op" | cut -d'/' -f2) # Parse operator name from its catalog path
-            if [ -n "$operator_name" ]; then
-              ALL_AFFECTED_OPERATORS["$operator_name"]=1
-            fi
-          done
-        fi
-
-
-        for operator_name in "${!ALL_AFFECTED_OPERATORS[@]}"; do
-          file_path="operators/$operator_name/ci.yaml"
-
-          if [ ! -f "$file_path" ]; then
-            echo "File '$file_path' not found."
-            exit 1
-          fi
-
-          CERT_PROJECT_ID=$(cat "$file_path" | yq -r '.cert_project_id | select (.!=null)')
-
-          if [ -z "$CERT_PROJECT_ID" ]; then
-            echo "Certification project ID is missing in '$file_path' file (cert_project_id)"
-            exit 1
-          fi
-        done
-
-        echo -n "$CERT_PROJECT_ID" | tee "$(results.certification_project_id.path)"
+        certification-project-check \
+          --repo-head-path "$(workspaces.source.path)" \
+          --affected-operators "$(params.affected_operators)" \
+          --affected-catalog-operators "$(params.affected_catalog_operators)" \
+          --cert-project-required "$(params.cert_project_required)" \
+          --output-file "$(results.certification_project_id.path)" \
+          --verbose \
+          $EXTRA_ARGS

docs/README.md

@@ -39,10 +39,12 @@ Depending on your use case, you may:
 
 To remove the operator completely from the catalog:
 
-- Delete the your operator directory from the `operators/` folder.
+- Delete your operator directory from the `operators/` folder.
 - Remove all catalog files related to your operator from the `catalogs/` directory.
 - Open a **single** pull request that includes these changes. Follow our contribution guidelines on how to [open a PR](users/contributing-via-pr.md).
 
+Community and certified (ISV) operators follow the same removal steps above. You do not need to keep a `ci.yaml` stub in the pull request for certified operators. When the operator directory is removed in the PR, the ISV pipeline reads `cert_project_id` from the `ci.yaml` on the target branch to authorize and process the removal.
+
 For reference, here’s an [example PR](https://github.com/redhat-openshift-ecosystem/community-operators-prod/pull/5955/files) demonstrating these steps.
 
 ### Remove the Operator from Specific Catalog Version(s)

docs/developer-guide.md

@@ -346,7 +346,8 @@ python3 -m pip install ansible-lint
 To run unit tests and code style checkers:
 
 ```bash
-tox
+poetry run tox -p  # all tests in paralel
+poetry run tox -e bandit  # single check 
 ```
 
 ### Local development

operatorcert/entrypoints/certification_project_check.py

@@ -0,0 +1,194 @@
+"""Check certification project identifiers for affected operators."""
+
+import argparse
+import logging
+import sys
+from pathlib import Path
+
+from operatorcert.logger import setup_logger
+from operatorcert.operator_repo.utils import load_yaml
+from operatorcert.utils import SplitArgs
+
+LOGGER = logging.getLogger("operator-cert")
+
+
+def setup_argparser() -> argparse.ArgumentParser:
+    """
+    Setup argument parser
+
+    Returns:
+        argparse.ArgumentParser: Argument parser
+    """
+    parser = argparse.ArgumentParser(
+        description="Check certification project identifiers for affected operators."
+    )
+    parser.add_argument(
+        "--repo-head-path",
+        required=True,
+        type=Path,
+        help="Path to the PR head repository clone",
+    )
+    parser.add_argument(
+        "--repo-base-path",
+        type=Path,
+        help="Path to the base branch repository clone",
+    )
+    parser.add_argument(
+        "--affected-operators",
+        default=[],
+        action=SplitArgs,
+        help="Comma separated list of affected operators",
+    )
+    parser.add_argument(
+        "--affected-catalog-operators",
+        default=[],
+        action=SplitArgs,
+        help="Comma separated list of affected catalog operators",
+    )
+    parser.add_argument(
+        "--cert-project-required",
+        default="true",
+        help="Whether a cert project ID must be present",
+    )
+    parser.add_argument(
+        "--output-file",
+        help="Path to a file where the certification project ID will be stored",
+    )
+    parser.add_argument("--verbose", action="store_true", help="Verbose output")
+
+    return parser
+
+
+def collect_affected_operators(
+    affected_operators: list[str], affected_catalog_operators: list[str]
+) -> set[str]:
+    """
+    Build a set of operator names affected by the pull request.
+
+    Args:
+        affected_operators (list[str]): Operator names from detect-changes
+        affected_catalog_operators (list[str]): Catalog operators in catalog/version format
+
+    Returns:
+        set[str]: Unique affected operator names
+    """
+    operators = set(affected_operators)
+    for catalog_operator in affected_catalog_operators:
+        if not catalog_operator or "/" not in catalog_operator:
+            continue
+        _, operator_name = catalog_operator.split("/", 1)
+        if operator_name.strip():
+            operators.add(operator_name.strip())
+    return operators
+
+
+def resolve_operator_ci_yaml_path(
+    source_root: Path,
+    base_root: Path | None,
+    operator_name: str,
+) -> Path:
+    """
+    Resolve the ci.yaml path for an operator, falling back to the base branch.
+
+    Args:
+        source_root (Path): PR head repository root
+        base_root (Path | None): Base branch repository root
+        operator_name (str): Operator name
+
+    Returns:
+        Path: Resolved ci.yaml path
+
+    Raises:
+        FileNotFoundError: When ci.yaml cannot be found in head or base
+    """
+    head_path = source_root / "operators" / operator_name / "ci.yaml"
+    if head_path.is_file():
+        return head_path
+
+    if base_root is not None:
+        base_path = base_root / "operators" / operator_name / "ci.yaml"
+        if base_path.is_file():
+            LOGGER.info(
+                "Using ci.yaml from base branch for operator '%s' (removed in PR).",
+                operator_name,
+            )
+            return base_path
+
+    raise FileNotFoundError(
+        f"ci.yaml not found for operator '{operator_name}' in PR head or base branch"
+    )
+
+
+def check_certification_projects(
+    source_root: Path,
+    base_root: Path | None,
+    operator_names: set[str],
+) -> str:
+    """
+    Validate cert_project_id for all affected operators.
+
+    Args:
+        source_root (Path): PR head repository root
+        base_root (Path | None): Base branch repository root
+        operator_names (set[str]): Operator names to check
+
+    Returns:
+        str: Certification project ID from the last validated operator
+
+    Raises:
+        ValueError: When no operators are affected or cert_project_id is missing
+        FileNotFoundError: When ci.yaml cannot be resolved
+    """
+    if not operator_names:
+        raise ValueError("No operator is affected.")
+
+    cert_project_id = ""
+    for operator_name in sorted(operator_names):
+        ci_path = resolve_operator_ci_yaml_path(source_root, base_root, operator_name)
+        config = load_yaml(ci_path)
+        project_id = (config or {}).get("cert_project_id")
+        if not project_id:
+            raise ValueError(
+                f"Certification project ID is missing in '{ci_path}' (cert_project_id)"
+            )
+        cert_project_id = project_id
+
+    return cert_project_id
+
+
+def main() -> None:
+    """
+    Main function of the script
+    """
+    parser = setup_argparser()
+    args = parser.parse_args()
+
+    log_level = "DEBUG" if args.verbose else "INFO"
+    setup_logger(level=log_level)
+
+    if args.cert_project_required != "true":
+        LOGGER.info("Cert project ID is not required.")
+        if args.output_file:
+            Path(args.output_file).write_text("", encoding="utf-8")
+        return
+
+    operator_names = collect_affected_operators(
+        args.affected_operators, args.affected_catalog_operators
+    )
+
+    try:
+        cert_project_id = check_certification_projects(
+            args.repo_head_path,
+            args.repo_base_path,
+            operator_names,
+        )
+    except (FileNotFoundError, ValueError) as exc:
+        LOGGER.error("%s", exc)
+        sys.exit(1)
+    LOGGER.info("Certification project ID: %s", cert_project_id)
+    if args.output_file:
+        Path(args.output_file).write_text(cert_project_id, encoding="utf-8")
+
+
+if __name__ == "__main__":  # pragma: no cover
+    main()

operatorcert/entrypoints/check_permissions.py

@@ -562,9 +562,9 @@ def check_permissions(
     operators = operators.union(
         extract_operators_from_catalog(head_repo, affected_catalog_operators)
     )
-    removed_catalog_operators = changes.get("removed_catalog_operators", [])
+    deleted_catalog_operators = changes.get("deleted_catalog_operators", [])
     operators = operators.union(
-        extract_operators_from_catalog(base_repo, removed_catalog_operators)
+        extract_operators_from_catalog(base_repo, deleted_catalog_operators)
     )
 
     is_approved = []

operatorcert/utils.py

@@ -130,7 +130,14 @@ class SplitArgs(argparse.Action):
     def __call__(
         self, parser: Any, namespace: Any, values: Any, option_string: Any = None
     ) -> None:
-        setattr(namespace, self.dest, values.split(","))
+        if not values or values.strip() == "":
+            setattr(namespace, self.dest, [])
+        else:
+            setattr(
+                namespace,
+                self.dest,
+                [v.strip() for v in values.split(",") if v.strip()],
+            )
 
 
 def run_command(

pyproject.toml

@@ -46,6 +46,7 @@ build-scratch-catalog = "operatorcert.entrypoints.build_scratch_catalog:main"
 bulk-retrigger = "operatorcert.entrypoints.bulk_retrigger:main"
 bundle-dockerfile = "operatorcert.entrypoints.bundle_dockerfile:main"
 catalog-browser = "operatorcert.catalog.catalog_cli:main"
+certification-project-check = "operatorcert.entrypoints.certification_project_check:main"
 check-permissions = "operatorcert.entrypoints.check_permissions:main"
 create-container-image = "operatorcert.entrypoints.create_container_image:main"
 create-github-gist = "operatorcert.entrypoints.create_github_gist:main"