OTA-1975: cluster-update/product-lifecycle: check for on-cluster products-data cincinnati service

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}

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()