|
| 1 | +"""Check certification project identifiers for affected operators.""" |
| 2 | + |
| 3 | +import argparse |
| 4 | +import logging |
| 5 | +import sys |
| 6 | +from pathlib import Path |
| 7 | + |
| 8 | +from operatorcert.logger import setup_logger |
| 9 | +from operatorcert.operator_repo.utils import load_yaml |
| 10 | +from operatorcert.utils import SplitArgs |
| 11 | + |
| 12 | +LOGGER = logging.getLogger("operator-cert") |
| 13 | + |
| 14 | + |
| 15 | +def setup_argparser() -> argparse.ArgumentParser: |
| 16 | + """ |
| 17 | + Setup argument parser |
| 18 | +
|
| 19 | + Returns: |
| 20 | + argparse.ArgumentParser: Argument parser |
| 21 | + """ |
| 22 | + parser = argparse.ArgumentParser( |
| 23 | + description="Check certification project identifiers for affected operators." |
| 24 | + ) |
| 25 | + parser.add_argument( |
| 26 | + "--repo-head-path", |
| 27 | + required=True, |
| 28 | + type=Path, |
| 29 | + help="Path to the PR head repository clone", |
| 30 | + ) |
| 31 | + parser.add_argument( |
| 32 | + "--repo-base-path", |
| 33 | + type=Path, |
| 34 | + help="Path to the base branch repository clone", |
| 35 | + ) |
| 36 | + parser.add_argument( |
| 37 | + "--affected-operators", |
| 38 | + default=[], |
| 39 | + action=SplitArgs, |
| 40 | + help="Comma separated list of affected operators", |
| 41 | + ) |
| 42 | + parser.add_argument( |
| 43 | + "--affected-catalog-operators", |
| 44 | + default=[], |
| 45 | + action=SplitArgs, |
| 46 | + help="Comma separated list of affected catalog operators", |
| 47 | + ) |
| 48 | + parser.add_argument( |
| 49 | + "--cert-project-required", |
| 50 | + default="true", |
| 51 | + help="Whether a cert project ID must be present", |
| 52 | + ) |
| 53 | + parser.add_argument( |
| 54 | + "--output-file", |
| 55 | + help="Path to a file where the certification project ID will be stored", |
| 56 | + ) |
| 57 | + parser.add_argument("--verbose", action="store_true", help="Verbose output") |
| 58 | + |
| 59 | + return parser |
| 60 | + |
| 61 | + |
| 62 | +def collect_affected_operators( |
| 63 | + affected_operators: list[str], affected_catalog_operators: list[str] |
| 64 | +) -> set[str]: |
| 65 | + """ |
| 66 | + Build a set of operator names affected by the pull request. |
| 67 | +
|
| 68 | + Args: |
| 69 | + affected_operators (list[str]): Operator names from detect-changes |
| 70 | + affected_catalog_operators (list[str]): Catalog operators in catalog/version format |
| 71 | +
|
| 72 | + Returns: |
| 73 | + set[str]: Unique affected operator names |
| 74 | + """ |
| 75 | + operators = set(affected_operators) |
| 76 | + for catalog_operator in affected_catalog_operators: |
| 77 | + if "/" not in catalog_operator: |
| 78 | + continue |
| 79 | + _, operator_name = catalog_operator.split("/", 1) |
| 80 | + if operator_name: |
| 81 | + operators.add(operator_name) |
| 82 | + return operators |
| 83 | + |
| 84 | + |
| 85 | +def resolve_operator_ci_yaml_path( |
| 86 | + source_root: Path, |
| 87 | + base_root: Path | None, |
| 88 | + operator_name: str, |
| 89 | +) -> Path: |
| 90 | + """ |
| 91 | + Resolve the ci.yaml path for an operator, falling back to the base branch. |
| 92 | +
|
| 93 | + Args: |
| 94 | + source_root (Path): PR head repository root |
| 95 | + base_root (Path | None): Base branch repository root |
| 96 | + operator_name (str): Operator name |
| 97 | +
|
| 98 | + Returns: |
| 99 | + Path: Resolved ci.yaml path |
| 100 | +
|
| 101 | + Raises: |
| 102 | + FileNotFoundError: When ci.yaml cannot be found in head or base |
| 103 | + """ |
| 104 | + head_path = source_root / "operators" / operator_name / "ci.yaml" |
| 105 | + if head_path.is_file(): |
| 106 | + return head_path |
| 107 | + |
| 108 | + if base_root is not None: |
| 109 | + base_path = base_root / "operators" / operator_name / "ci.yaml" |
| 110 | + if base_path.is_file(): |
| 111 | + LOGGER.info( |
| 112 | + "Using ci.yaml from base branch for operator '%s' (removed in PR).", |
| 113 | + operator_name, |
| 114 | + ) |
| 115 | + return base_path |
| 116 | + |
| 117 | + raise FileNotFoundError( |
| 118 | + f"ci.yaml not found for operator '{operator_name}' in PR head or base branch" |
| 119 | + ) |
| 120 | + |
| 121 | + |
| 122 | +def check_certification_projects( |
| 123 | + source_root: Path, |
| 124 | + base_root: Path | None, |
| 125 | + operator_names: set[str], |
| 126 | +) -> str: |
| 127 | + """ |
| 128 | + Validate cert_project_id for all affected operators. |
| 129 | +
|
| 130 | + Args: |
| 131 | + source_root (Path): PR head repository root |
| 132 | + base_root (Path | None): Base branch repository root |
| 133 | + operator_names (set[str]): Operator names to check |
| 134 | +
|
| 135 | + Returns: |
| 136 | + str: Certification project ID from the last validated operator |
| 137 | +
|
| 138 | + Raises: |
| 139 | + ValueError: When no operators are affected or cert_project_id is missing |
| 140 | + FileNotFoundError: When ci.yaml cannot be resolved |
| 141 | + """ |
| 142 | + if not operator_names: |
| 143 | + raise ValueError("No operator is affected.") |
| 144 | + |
| 145 | + cert_project_id = "" |
| 146 | + for operator_name in sorted(operator_names): |
| 147 | + ci_path = resolve_operator_ci_yaml_path(source_root, base_root, operator_name) |
| 148 | + config = load_yaml(ci_path) |
| 149 | + project_id = (config or {}).get("cert_project_id") |
| 150 | + if not project_id: |
| 151 | + raise ValueError( |
| 152 | + f"Certification project ID is missing in '{ci_path}' (cert_project_id)" |
| 153 | + ) |
| 154 | + cert_project_id = project_id |
| 155 | + |
| 156 | + return cert_project_id |
| 157 | + |
| 158 | + |
| 159 | +def main() -> None: |
| 160 | + """ |
| 161 | + Main function of the script |
| 162 | + """ |
| 163 | + parser = setup_argparser() |
| 164 | + args = parser.parse_args() |
| 165 | + |
| 166 | + log_level = "DEBUG" if args.verbose else "INFO" |
| 167 | + setup_logger(level=log_level) |
| 168 | + |
| 169 | + if args.cert_project_required != "true": |
| 170 | + LOGGER.info("Cert project ID is not required.") |
| 171 | + if args.output_file: |
| 172 | + Path(args.output_file).write_text("", encoding="utf-8") |
| 173 | + return |
| 174 | + |
| 175 | + operator_names = collect_affected_operators( |
| 176 | + args.affected_operators, args.affected_catalog_operators |
| 177 | + ) |
| 178 | + |
| 179 | + try: |
| 180 | + cert_project_id = check_certification_projects( |
| 181 | + args.repo_head_path, |
| 182 | + args.repo_base_path, |
| 183 | + operator_names, |
| 184 | + ) |
| 185 | + except (FileNotFoundError, ValueError) as exc: |
| 186 | + LOGGER.error("%s", exc) |
| 187 | + sys.exit(1) |
| 188 | + print(cert_project_id) |
| 189 | + if args.output_file: |
| 190 | + Path(args.output_file).write_text(cert_project_id, encoding="utf-8") |
| 191 | + |
| 192 | + |
| 193 | +if __name__ == "__main__": # pragma: no cover |
| 194 | + main() |
0 commit comments