From b37f4571e098445f405c7673ab3c59e239b026e6 Mon Sep 17 00:00:00 2001 From: Ankita Thomas Date: Fri, 24 Jul 2026 09:57:13 -0400 Subject: [PATCH] check for on-cluster producs-data cincinnati service for plcc skill Signed-off-by: Ankita Thomas --- .../product-lifecycle/scripts/plc_lookup.py | 54 +++++++++++++++-- .../scripts/tests/test_plc_lookup.py | 60 +++++++++++++++++++ 2 files changed, 109 insertions(+), 5 deletions(-) diff --git a/cluster-update/product-lifecycle/scripts/plc_lookup.py b/cluster-update/product-lifecycle/scripts/plc_lookup.py index b87b928..6ffea1e 100755 --- a/cluster-update/product-lifecycle/scripts/plc_lookup.py +++ b/cluster-update/product-lifecycle/scripts/plc_lookup.py @@ -12,8 +12,40 @@ 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 check_connectivity(url, timeout=5): + """Quick connectivity check to an API endpoint.""" + try: + req = urllib.request.Request(url, headers={"User-Agent": "plc-lookup/1.0"}) + with urllib.request.urlopen(req, timeout=timeout) as resp: + return resp.status == 200 + except Exception: + return False + + +def get_products_api_base(cincinnati_url=None): + """Determine which products API to use with fallback logic.""" + public_api = API_BASE + + # Try public API first + if check_connectivity(public_api, timeout=5): + return public_api + + # Fall back to Cincinnati if provided + if cincinnati_url: + cincinnati_products = cincinnati_url.rstrip('/') + '/products' + if check_connectivity(cincinnati_products, timeout=5): + return cincinnati_products + + # Neither available + raise SystemExit(json.dumps({ + "error": "no_products_data", + "detail": "Neither public API nor Cincinnati products endpoint available" + }, indent=2)) + + +def api_search(name, base_url): + """Query products API (works for both public API and Cincinnati).""" + url = f"{base_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: @@ -22,6 +54,14 @@ def api_search(name): raise SystemExit(json.dumps({"error": "api_request_failed", "detail": str(e)}, indent=2)) except (json.JSONDecodeError, ValueError) as e: raise SystemExit(json.dumps({"error": "invalid_response", "detail": str(e)}, indent=2)) + + # Check if response is empty + if body == {}: + raise SystemExit(json.dumps({ + "error": "no_products_data", + "detail": "Products endpoint returned empty data" + }, indent=2)) + if "data" not in body: raise SystemExit(json.dumps({"error": "unexpected_response", "keys": list(body.keys())}, indent=2)) return body["data"] @@ -60,7 +100,8 @@ def format_product_version(product, version, target_ocp=None): def cmd_products(args, output=sys.stdout): - products = api_search(args.name) + api_base = get_products_api_base(getattr(args, 'cincinnati_url', None)) + products = api_search(args.name, api_base) if not products: json.dump({"error": "no products found", "query": args.name}, output, indent=2) output.write("\n") @@ -81,10 +122,11 @@ def cmd_products(args, output=sys.stdout): def cmd_olm_check(args, output=sys.stdout): + api_base = get_products_api_base(getattr(args, 'cincinnati_url', None)) operators = json.loads(args.operators) target = args.ocp - batch = api_search("OpenShift") + batch = api_search("OpenShift", api_base) by_package = collections.defaultdict(list) for p in batch: pkg = p.get("package") @@ -109,7 +151,7 @@ def cmd_olm_check(args, output=sys.stdout): products = by_package.get(pkg) if not products: - extra = api_search(pkg.replace("-", " ")) + extra = api_search(pkg.replace("-", " "), api_base) products = [p for p in extra if p.get("package") == pkg] if not products: @@ -152,6 +194,7 @@ def main(args=None, output=sys.stdout): ) p_products.add_argument("name", help="Product name (substring match)") p_products.add_argument("--ocp", help="Check compatibility against this OCP version (e.g. 4.21)") + p_products.add_argument("--cincinnati-url", help="Cincinnati base URL for fallback (e.g. https://cincinnati.example.com)") p_olm = subparsers.add_parser( "olm-check", @@ -163,6 +206,7 @@ def main(args=None, output=sys.stdout): required=True, help='JSON array of operators, e.g. \'[{"package":"cluster-logging"}]\'', ) + p_olm.add_argument("--cincinnati-url", help="Cincinnati base URL for fallback (e.g. https://cincinnati.example.com)") parsed = parser.parse_args(args) handlers = {"products": cmd_products, "olm-check": cmd_olm_check} 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..c789e1f 100644 --- a/cluster-update/product-lifecycle/scripts/tests/test_plc_lookup.py +++ b/cluster-update/product-lifecycle/scripts/tests/test_plc_lookup.py @@ -501,5 +501,65 @@ def test_former_names_preserved(self): ) +class TestConnectivityCheck(unittest.TestCase): + def test_successful_connectivity(self): + with patch.object(plc_lookup.urllib.request, "urlopen") as mock_urlopen: + mock_resp = MagicMock() + mock_resp.status = 200 + mock_resp.__enter__ = lambda s: s + mock_resp.__exit__ = MagicMock(return_value=False) + mock_urlopen.return_value = mock_resp + + result = plc_lookup.check_connectivity("https://example.com/products") + self.assertTrue(result) + + def test_failed_connectivity(self): + with patch.object(plc_lookup.urllib.request, "urlopen", + side_effect=Exception("Network error")): + result = plc_lookup.check_connectivity("https://example.com/products") + self.assertFalse(result) + + +class TestGetProductsApiBase(unittest.TestCase): + def test_public_api_available(self): + with patch.object(plc_lookup, "check_connectivity", side_effect=lambda url, **kw: url == plc_lookup.API_BASE): + base = plc_lookup.get_products_api_base() + self.assertEqual(base, plc_lookup.API_BASE) + + def test_cincinnati_fallback(self): + def mock_connectivity(url, **kw): + return url == "https://cincinnati.example.com/products" + + with patch.object(plc_lookup, "check_connectivity", side_effect=mock_connectivity): + base = plc_lookup.get_products_api_base("https://cincinnati.example.com") + self.assertEqual(base, "https://cincinnati.example.com/products") + + def test_no_api_available_raises(self): + with patch.object(plc_lookup, "check_connectivity", return_value=False): + with self.assertRaises(SystemExit) as ctx: + plc_lookup.get_products_api_base() + error = json.loads(str(ctx.exception)) + self.assertEqual(error["error"], "no_products_data") + + +class TestCincinnatiUrlParameter(unittest.TestCase): + def test_products_with_cincinnati_url(self): + with _mock_api_search([SAMPLE_PRODUCT]): + output = _run_main([ + "products", "logging", + "--cincinnati-url", "https://cincinnati.example.com" + ]) + self.assertGreater(output["total"], 0) + + def test_olm_check_with_cincinnati_url(self): + with _mock_api_search([SAMPLE_PRODUCT]): + output = _run_main([ + "olm-check", "--ocp", "4.21", + "--operators", '[{"package":"cluster-logging"}]', + "--cincinnati-url", "https://cincinnati.example.com" + ]) + self.assertEqual(output["operators_checked"], 1) + + if __name__ == "__main__": unittest.main()