From d36a8617c40f42d00df97374dfae42cbb0dda5e6 Mon Sep 17 00:00:00 2001 From: Jefferson Ramos Date: Thu, 23 Jul 2026 22:54:12 -0300 Subject: [PATCH] Return per-operator verdicts from olm-check MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The olm-check command previously dumped all version entries for every matched package, leaving the LLM to determine whether the specific installed version was actually tracked. Claude handled this correctly but gemini and openai miscounted — they saw data for a package and assumed lifecycle data was available, even when the installed version was not tracked (e.g. web-terminal v1.16 is not in the API). Refactor olm-check to return one result per operator. When a version is provided, the tool normalizes it to major.minor and matches against API version names. Use a single `error` string field — present means failure (self-describing), absent means success — so incoherent states are impossible. Also switch from api_search("OpenShift") to api_all() so operators with non-OpenShift product names (e.g. compliance-operator) are not missed by the name filter. --- cluster-update/product-lifecycle/SKILL.md | 60 ++++-- .../product-lifecycle/scripts/plc_lookup.py | 86 ++++++-- .../scripts/tests/test_plc_lookup.py | 202 +++++++++++------- 3 files changed, 227 insertions(+), 121 deletions(-) diff --git a/cluster-update/product-lifecycle/SKILL.md b/cluster-update/product-lifecycle/SKILL.md index b5d94fc..3bc2adf 100644 --- a/cluster-update/product-lifecycle/SKILL.md +++ b/cluster-update/product-lifecycle/SKILL.md @@ -41,29 +41,52 @@ Be specific with product names to avoid overly broad results. ```bash python3 product-lifecycle/scripts/plc_lookup.py olm-check --ocp 4.21 \ - --operators '[{"package":"cluster-logging"},{"package":"web-terminal"}]' + --operators '[{"package":"cluster-logging","version":"6.5.1"},{"package":"web-terminal"}]' ``` -Looks up each operator by its OLM `package` name. Reports -`lifecycle_unavailable` for operators not tracked by package name. -When `olm-check` reports `lifecycle_unavailable`, retry those operators -with `products` using the operator's product name (e.g. "compliance operator") -before concluding that lifecycle data is unavailable. +Looks up each operator by its OLM `package` name. Each operator in the +JSON array accepts `package` (required) and `version` (optional). + +Returns **one result per operator**. The shape depends on whether a +`version` was provided: + +- **With version (matched):** `package`, `requested_version`, `product`, + `status`, `ocp_compatible`, `phases`. No `error` field. +- **Without version (package exists):** `package`, `product`, + `available_versions`. No `error` field. +- **Error:** `error` field set — explains what went wrong (package not + found, version not tracked). Includes `available_versions` when the + package exists but the version doesn't. + +When a version is provided, the tool normalizes it to major.minor +(e.g. `1.9.0` → `1.9`) and matches against API version names. + +Reports `lifecycle_unavailable` listing every operator whose result has `error` set. + ## Output Format -All commands output JSON. Each product version entry includes: +All commands output JSON. + +### `products` output + +Each product version entry includes `product`, `former_names`, `package`, +`version`, `status`, `ocp_versions`, `ocp_compatible`, and `phases`. + +### `olm-check` output + +One entry per operator: | Field | Description | |---|---| -| `product` | Current product name | -| `former_names` | Previous product names (useful for search fallback) | -| `package` | OLM package name (maps to Subscription `spec.name`) | -| `version` | Version number | -| `status` | Raw API support status (e.g. `"Full Support"`, `"End of life"`) | -| `ocp_versions` | List of compatible OCP versions (empty for non-layered products) | -| `ocp_compatible` | `true`/`false`/`null` — only present when `--ocp` is used | -| `phases` | Array of lifecycle phases with `name`, `start_date`, `end_date` | +| `package` | OLM package name queried | +| `requested_version` | Normalized version checked (when provided) | +| `error` | Why data is unavailable — absent on success, set on failure | +| `product` | Product name (on success) | +| `status` | Raw API support status, e.g. `"Full Support"`, `"End of life"` (on success) | +| `ocp_compatible` | `true`/`false`/`null` — whether the version is compatible with the target OCP (on success) | +| `phases` | Lifecycle phases with `name`, `start_date`, `end_date` (on success) | +| `available_versions` | Versions the API does track (when package exists but version doesn't) | ## When to Use @@ -77,10 +100,7 @@ All commands output JSON. Each product version entry includes: ## Important - `ocp_versions` is only present on **layered product** versions, not on OCP itself. -- Not all operators have lifecycle entries — report "lifecycle data unavailable" - rather than treating missing data as an error. -- An operator has lifecycle data only if the API returns an entry matching its - **installed version**. If the API tracks the package but not the specific - version installed, treat it as "lifecycle data unavailable" for that version. +- Not all operators have lifecycle entries — `olm-check` sets `error` on the + result for operators without data. - The `package` field in API responses maps to the OLM Subscription's `spec.name` — use this for exact matching, not product name. diff --git a/cluster-update/product-lifecycle/scripts/plc_lookup.py b/cluster-update/product-lifecycle/scripts/plc_lookup.py index b87b928..167ec9c 100755 --- a/cluster-update/product-lifecycle/scripts/plc_lookup.py +++ b/cluster-update/product-lifecycle/scripts/plc_lookup.py @@ -12,8 +12,8 @@ API_BASE = "https://access.redhat.com/product-life-cycles/api/v2/products" -def api_search(name): - url = f"{API_BASE}?{urllib.parse.urlencode({'name': name})}" +def _api_get(url): + """Fetch a PLC API URL and return the 'data' list.""" req = urllib.request.Request(url, headers={"User-Agent": "plc-lookup/1.0"}) try: with urllib.request.urlopen(req, timeout=30) as resp: @@ -27,6 +27,20 @@ def api_search(name): return body["data"] +def api_search(url=API_BASE, name=None): + """Fetch products, optionally filtering by name.""" + if name: + url = f"{url}?{urllib.parse.urlencode({'name': name})}" + return _api_get(url) + + +def _normalize_version(version): + """Normalize to major.minor, stripping leading 'v' (e.g. 'v1.9.0' → '1.9').""" + version = version.lstrip("v") + parts = version.split(".") + return ".".join(parts[:2]) if len(parts) >= 2 else version + + def parse_ocp_versions(compat_string): if not compat_string: return [] @@ -60,7 +74,7 @@ def format_product_version(product, version, target_ocp=None): def cmd_products(args, output=sys.stdout): - products = api_search(args.name) + products = api_search(name=args.name) if not products: json.dump({"error": "no products found", "query": args.name}, output, indent=2) output.write("\n") @@ -84,51 +98,76 @@ def cmd_olm_check(args, output=sys.stdout): operators = json.loads(args.operators) target = args.ocp - batch = api_search("OpenShift") + all_products = api_search() by_package = collections.defaultdict(list) - for p in batch: + for p in all_products: pkg = p.get("package") if pkg: by_package[pkg].append(p) results = [] - missed_packages = [] for op in operators: pkg = op.get("package", "") + requested_version = op.get("version", "") if not pkg: - results.append({ - "package": pkg, - "status": "lifecycle_unavailable", - "reason": "empty package name", - }) - missed_packages.append(pkg) + results.append({"package": pkg, "error": "empty package name"}) continue products = by_package.get(pkg) if not products: - extra = api_search(pkg.replace("-", " ")) - products = [p for p in extra if p.get("package") == pkg] + entry = {"package": pkg, "error": "package not found in PLC API"} + if requested_version: + entry["requested_version"] = _normalize_version(requested_version) + results.append(entry) + continue - if not products: + all_versions = {v.get("name") for p in products for v in p["versions"]} - {None} + + if not requested_version: + results.append({ + "package": pkg, + "product": products[0]["name"], + "available_versions": sorted(all_versions), + }) + continue + + norm = _normalize_version(requested_version) + matched = None + for p in products: + for v in p["versions"]: + if v.get("name") in (norm, f"{norm}.x"): + matched = (p, v) + break + if matched: + break + + if not matched: results.append({ "package": pkg, - "status": "lifecycle_unavailable", - "reason": "no lifecycle data found for this package", + "requested_version": norm, + "error": f"version {norm} not tracked", + "available_versions": sorted(all_versions), }) - missed_packages.append(pkg) continue - for product in products: - for v in product["versions"]: - results.append(format_product_version(product, v, target_ocp=target)) + product, version = matched + formatted = format_product_version(product, version, target_ocp=target) + results.append({ + "package": pkg, + "requested_version": norm, + "product": formatted["product"], + "status": formatted["status"], + "ocp_compatible": formatted.get("ocp_compatible"), + "phases": formatted["phases"], + }) json.dump({ "ocp_target": target, "operators_checked": len(operators), - "lifecycle_unavailable": missed_packages, + "lifecycle_unavailable": [r["package"] for r in results if "error" in r], "results": results, }, output, indent=2) output.write("\n") @@ -141,7 +180,8 @@ def main(args=None, output=sys.stdout): epilog="Examples:\n" ' %(prog)s products "logging for Red Hat OpenShift"\n' ' %(prog)s products "logging for Red Hat OpenShift" --ocp 4.21\n' - ' %(prog)s olm-check --ocp 4.21 --operators \'[{"package":"cluster-logging"}]\'\n', + ' %(prog)s olm-check --ocp 4.21 --operators \'[{"package":"cluster-logging"}]\'\n' + ' %(prog)s olm-check --ocp 4.21 --operators \'[{"package":"cluster-logging","version":"6.5.1"},{"package":"openshift-pipelines-operator-rh","version":"1.22.0"}]\'\n', formatter_class=argparse.RawDescriptionHelpFormatter, ) subparsers = parser.add_subparsers(dest="command", required=True) diff --git a/cluster-update/product-lifecycle/scripts/tests/test_plc_lookup.py b/cluster-update/product-lifecycle/scripts/tests/test_plc_lookup.py index c0f4a32..56557ee 100644 --- a/cluster-update/product-lifecycle/scripts/tests/test_plc_lookup.py +++ b/cluster-update/product-lifecycle/scripts/tests/test_plc_lookup.py @@ -73,7 +73,7 @@ "former_names": [], "versions": [ { - "name": "1.8", + "name": "1.8.x", "type": "Full Support", "openshift_compatibility": "4.16, 4.17", "phases": [], @@ -95,6 +95,11 @@ "openshift_compatibility": "4.16, 4.17", "phases": [], }, + { + "type": "Tech Preview", + "openshift_compatibility": "4.18", + "phases": [], + }, ], } @@ -104,6 +109,7 @@ def _mock_api_search(data): return patch.object(plc_lookup, "api_search", return_value=data) + def _run_main(args): """Run plc_lookup.main() with given args and return parsed JSON output.""" output = io.StringIO() @@ -244,112 +250,152 @@ def test_no_ocp_target(self): class TestCmdOlmCheck(unittest.TestCase): - def test_found_operator(self): + def test_found_with_version(self): + """Version provided and tracked → no error, with status.""" with _mock_api_search([SAMPLE_PRODUCT]): output = _run_main([ "olm-check", "--ocp", "4.21", - "--operators", '[{"package":"cluster-logging"}]', + "--operators", '[{"package":"cluster-logging","version":"6.5.0"}]', ]) self.assertEqual(output["ocp_target"], "4.21") self.assertEqual(output["operators_checked"], 1) - self.assertEqual( - output["lifecycle_unavailable"], [], - "cluster-logging should be found in the batch", - ) - found_products = [r["product"] for r in output["results"] if "product" in r] - self.assertIn( - "logging for Red Hat OpenShift", found_products, - "cluster-logging should resolve to 'logging for Red Hat OpenShift'", - ) - - def test_unavailable_operator(self): - with _mock_api_search([]): + self.assertEqual(output["lifecycle_unavailable"], []) + r = output["results"][0] + self.assertNotIn("error", r) + self.assertEqual(r["requested_version"], "6.5") + self.assertEqual(r["product"], "logging for Red Hat OpenShift") + self.assertEqual(r["status"], "Full Support") + self.assertTrue(r["ocp_compatible"]) + + def test_found_with_v_prefix(self): + """Leading 'v' in version is stripped before matching.""" + with _mock_api_search([SAMPLE_PRODUCT]): output = _run_main([ "olm-check", "--ocp", "4.21", - "--operators", '[{"package":"nonexistent-operator"}]', + "--operators", '[{"package":"cluster-logging","version":"v6.5.0"}]', ]) - self.assertEqual(output["operators_checked"], 1) - self.assertEqual(output["lifecycle_unavailable"], ["nonexistent-operator"]) - self.assertEqual(output["results"][0]["status"], "lifecycle_unavailable") + r = output["results"][0] + self.assertNotIn("error", r) + self.assertEqual(r["requested_version"], "6.5") + self.assertEqual(r["product"], "logging for Red Hat OpenShift") + + def test_found_with_x_suffix(self): + """API version '1.8.x' matches normalized '1.8'.""" + with _mock_api_search([SAMPLE_PRODUCT_DUPLICATE_PKG]): + output = _run_main([ + "olm-check", "--ocp", "4.17", + "--operators", '[{"package":"skupper-operator","version":"1.8.2"}]', + ]) + r = output["results"][0] + self.assertNotIn("error", r) + self.assertEqual(r["requested_version"], "1.8") + self.assertEqual(r["product"], "Red Hat Service Interconnect Operator") + + def test_version_entry_missing_name(self): + """Version entries with no 'name' key don't crash.""" + with _mock_api_search([SAMPLE_PRODUCT_DUPLICATE_PKG_2]): + output = _run_main([ + "olm-check", "--ocp", "4.17", + "--operators", '[{"package":"skupper-operator","version":"1.8.0"}]', + ]) + r = output["results"][0] + self.assertNotIn("error", r) + self.assertEqual(r["requested_version"], "1.8") - def test_duplicate_package_preserves_all_products(self): - """Multiple products sharing a package should all appear in results.""" - batch = [SAMPLE_PRODUCT_DUPLICATE_PKG, SAMPLE_PRODUCT_DUPLICATE_PKG_2] - with _mock_api_search(batch): + def test_version_entry_missing_name_no_version(self): + """No version + nameless entries don't crash; nameless excluded from available_versions.""" + with _mock_api_search([SAMPLE_PRODUCT_DUPLICATE_PKG_2]): output = _run_main([ "olm-check", "--ocp", "4.17", "--operators", '[{"package":"skupper-operator"}]', ]) - self.assertEqual( - output["lifecycle_unavailable"], [], - "skupper-operator should be found", - ) - product_names = {r["product"] for r in output["results"] if "product" in r} - self.assertIn("Red Hat Service Interconnect", product_names) - self.assertIn("Red Hat Service Interconnect Operator", product_names) + r = output["results"][0] + self.assertNotIn("error", r) + self.assertIn("1.8", r["available_versions"]) + self.assertNotIn(None, r["available_versions"]) - def test_mixed_found_and_unavailable(self): + def test_found_no_version(self): + """No version provided → no error, with available_versions.""" with _mock_api_search([SAMPLE_PRODUCT]): output = _run_main([ "olm-check", "--ocp", "4.21", - "--operators", - '[{"package":"cluster-logging"},{"package":"nonexistent"}]', + "--operators", '[{"package":"cluster-logging"}]', ]) - self.assertEqual(output["operators_checked"], 2) - self.assertEqual( - output["lifecycle_unavailable"], ["nonexistent"], - "Only nonexistent should be unavailable", - ) - - def test_fallback_search(self): - """Fallback searches product name with spaces, not hyphens.""" - search_names = [] - - def mock_search(name): - search_names.append(name) - if name == "OpenShift": - return [] - if name == "cluster logging": - return [SAMPLE_PRODUCT] - return [] - - with patch.object(plc_lookup, "api_search", side_effect=mock_search): + r = output["results"][0] + self.assertNotIn("error", r) + self.assertIn("available_versions", r) + self.assertIn("6.5", r["available_versions"]) + self.assertIn("5.9", r["available_versions"]) + + def test_version_not_tracked(self): + """Package exists but version doesn't → error with available_versions.""" + with _mock_api_search([SAMPLE_PRODUCT]): output = _run_main([ "olm-check", "--ocp", "4.21", - "--operators", '[{"package":"cluster-logging"}]', + "--operators", '[{"package":"cluster-logging","version":"9.9.0"}]', ]) - self.assertEqual( - output["lifecycle_unavailable"], [], - "cluster-logging should be found via fallback search", - ) - self.assertEqual( - search_names, ["OpenShift", "cluster logging"], - "Fallback should search with spaces, not hyphens", - ) - + r = output["results"][0] + self.assertEqual(r["requested_version"], "9.9") + self.assertIn("version 9.9 not tracked", r["error"]) + self.assertIn("6.5", r["available_versions"]) + self.assertEqual(output["lifecycle_unavailable"], ["cluster-logging"]) + + def test_package_not_found(self): + """Package not in API → error, in lifecycle_unavailable.""" + with _mock_api_search([]): + output = _run_main([ + "olm-check", "--ocp", "4.21", + "--operators", '[{"package":"nonexistent-operator"}]', + ]) + self.assertEqual(output["operators_checked"], 1) + self.assertEqual(output["lifecycle_unavailable"], ["nonexistent-operator"]) + r = output["results"][0] + self.assertIn("not found", r["error"]) - def test_empty_package_skips_api_call(self): - """Empty package name should not trigger an API call.""" + def test_mixed_results(self): + """Batch with success, version-missing, and package-missing operators.""" + with _mock_api_search([SAMPLE_PRODUCT]): + output = _run_main([ + "olm-check", "--ocp", "4.21", + "--operators", json.dumps([ + {"package": "cluster-logging", "version": "6.5.0"}, + {"package": "cluster-logging", "version": "9.9.0"}, + {"package": "nonexistent", "version": "1.0.0"}, + ]), + ]) + self.assertEqual(output["operators_checked"], 3) + self.assertNotIn("error", output["results"][0]) + self.assertIn("not tracked", output["results"][1]["error"]) + self.assertIn("not found", output["results"][2]["error"]) + self.assertEqual(output["lifecycle_unavailable"], ["cluster-logging", "nonexistent"]) + + def test_single_api_call(self): + """olm-check fetches all products in one call.""" call_count = [0] - def mock_search(name): + def counting_search(url=plc_lookup.API_BASE, name=None): call_count[0] += 1 - return [] + return [SAMPLE_PRODUCT] + + with patch.object(plc_lookup, "api_search", side_effect=counting_search): + output = _run_main([ + "olm-check", "--ocp", "4.21", + "--operators", '[{"package":"cluster-logging"}]', + ]) + self.assertNotIn("error", output["results"][0]) + self.assertEqual(call_count[0], 1, "Should call api_search exactly once") - with patch.object(plc_lookup, "api_search", side_effect=mock_search): + def test_empty_package(self): + """Empty package name → error.""" + with _mock_api_search([SAMPLE_PRODUCT]): output = _run_main([ "olm-check", "--ocp", "4.21", "--operators", '[{"package":""},{}]', ]) self.assertEqual(output["operators_checked"], 2) - self.assertEqual( - output["lifecycle_unavailable"], ["", ""], - "Both empty-package operators should be unavailable", - ) - self.assertEqual( - call_count[0], 1, - "Should only call api_search once (OpenShift batch), not for empty packages", - ) + self.assertEqual(output["lifecycle_unavailable"], ["", ""]) + self.assertIn("error", output["results"][0]) + self.assertIn("error", output["results"][1]) class TestApiSearchErrors(unittest.TestCase): @@ -359,7 +405,7 @@ def test_url_error_produces_json(self): with patch.object(plc_lookup.urllib.request, "urlopen", side_effect=plc_lookup.urllib.error.URLError("connection refused")): with self.assertRaises(SystemExit) as ctx: - plc_lookup.api_search("test") + plc_lookup.api_search(name="test") error = json.loads(str(ctx.exception)) self.assertEqual(error["error"], "api_request_failed") self.assertIn("connection refused", error["detail"]) @@ -371,7 +417,7 @@ def test_invalid_json_produces_error(self): mock_resp.__exit__ = MagicMock(return_value=False) with patch.object(plc_lookup.urllib.request, "urlopen", return_value=mock_resp): with self.assertRaises(SystemExit) as ctx: - plc_lookup.api_search("test") + plc_lookup.api_search(name="test") error = json.loads(str(ctx.exception)) self.assertEqual(error["error"], "invalid_response") @@ -382,7 +428,7 @@ def test_missing_data_key_produces_error(self): mock_resp.__exit__ = MagicMock(return_value=False) with patch.object(plc_lookup.urllib.request, "urlopen", return_value=mock_resp): with self.assertRaises(SystemExit) as ctx: - plc_lookup.api_search("test") + plc_lookup.api_search(name="test") error = json.loads(str(ctx.exception)) self.assertEqual(error["error"], "unexpected_response") self.assertIn("results", error["keys"])