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..082dc78 100755 --- a/cluster-update/product-lifecycle/scripts/plc_lookup.py +++ b/cluster-update/product-lifecycle/scripts/plc_lookup.py @@ -12,8 +12,11 @@ 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_search(name=None): + """Fetch products from the PLC API, optionally filtering by name.""" + url = API_BASE + if name: + url = f"{url}?{urllib.parse.urlencode({'name': name})}" req = urllib.request.Request(url, headers={"User-Agent": "plc-lookup/1.0"}) try: with urllib.request.urlopen(req, timeout=30) as resp: @@ -27,6 +30,13 @@ def api_search(name): return body["data"] +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 +70,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 +94,78 @@ 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, "error": "empty package name"}) + continue + + products = by_package.get(pkg) + + if not products: + entry = {"package": pkg, "error": "package not found in PLC API"} + if requested_version: + entry["requested_version"] = _normalize_version(requested_version) + results.append(entry) + continue + + all_versions = {v.get("name") for p in products for v in p["versions"]} - {None} + + if not requested_version: results.append({ "package": pkg, - "status": "lifecycle_unavailable", - "reason": "empty package name", + "product": products[0]["name"], + "available_versions": sorted(all_versions), }) - missed_packages.append(pkg) continue - products = by_package.get(pkg) + norm = _normalize_version(requested_version) + matches = [] + for p in products: + for v in p["versions"]: + if v.get("name") in (norm, f"{norm}.x"): + matches.append((p, v)) - if not products: - extra = api_search(pkg.replace("-", " ")) - products = [p for p in extra if p.get("package") == pkg] + matched = next( + (m for m in matches if m[1].get("openshift_compatibility")), + matches[0] if matches else None, + ) - if not products: + 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 +178,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..94e3fda 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,45 @@ "openshift_compatibility": "4.16, 4.17", "phases": [], }, + { + "type": "Tech Preview", + "openshift_compatibility": "4.18", + "phases": [], + }, + ], +} + +SAMPLE_SKUPPER_NO_COMPAT = { + "name": "Red Hat Service Interconnect", + "package": "skupper-operator", + "is_operator": True, + "is_layered_product": False, + "is_retired": False, + "former_names": [], + "versions": [ + { + "name": "2.1.x", + "type": "Full Support", + "openshift_compatibility": None, + "phases": [{"name": "Full support", "start_date": "2025-01-01", "end_date": "2026-01-01"}], + }, + ], +} + +SAMPLE_SKUPPER_WITH_COMPAT = { + "name": "Red Hat Service Interconnect Operator", + "package": "skupper-operator", + "is_operator": True, + "is_layered_product": False, + "is_retired": False, + "former_names": [], + "versions": [ + { + "name": "2.1.x", + "type": "Full Support", + "openshift_compatibility": "4.12, 4.14, 4.16, 4.17, 4.18, 4.19, 4.20, 4.21", + "phases": [{"name": "Full support", "start_date": "2025-01-01", "end_date": "2026-01-01"}], + }, ], } @@ -104,6 +143,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 +284,176 @@ 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_prefers_product_with_ocp_compatibility(self): + """When multiple products share a package, prefer the one with openshift_compatibility.""" + with _mock_api_search([SAMPLE_SKUPPER_NO_COMPAT, SAMPLE_SKUPPER_WITH_COMPAT]): + output = _run_main([ + "olm-check", "--ocp", "4.17", + "--operators", '[{"package":"skupper-operator","version":"2.1.0"}]', + ]) + r = output["results"][0] + self.assertNotIn("error", r) + self.assertEqual(r["product"], "Red Hat Service Interconnect Operator") + self.assertTrue(r["ocp_compatible"]) + + def test_prefers_product_with_ocp_compatibility_reversed_order(self): + """Same preference regardless of product order in API response.""" + with _mock_api_search([SAMPLE_SKUPPER_WITH_COMPAT, SAMPLE_SKUPPER_NO_COMPAT]): + output = _run_main([ + "olm-check", "--ocp", "4.17", + "--operators", '[{"package":"skupper-operator","version":"2.1.0"}]', + ]) + r = output["results"][0] + self.assertNotIn("error", r) + self.assertEqual(r["product"], "Red Hat Service Interconnect Operator") + self.assertTrue(r["ocp_compatible"]) + + 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 +463,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 +475,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 +486,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"])