feat(info_collector): add test to fetch versions for tools in cluster

testsuite/component_metadata.py

@@ -117,17 +117,41 @@ def get_component_images(project) -> list[tuple]:
                     if normalised_image in seen:
                         continue
                     seen.add(normalised_image)
-                    image_name = normalised_image.split("/")[-1]
-                    if ":" in image_name:
-                        name, tag = image_name.rsplit(":", 1)
-                        images.append((name, tag, image))
+
+                    name = normalised_image.split("/")[-1]
+                    if ":" in name:
+                        name, tag = name.rsplit(":", 1)
                     else:
-                        images.append((image_name, None, image))
-        except (oc.OpenShiftPythonException, AttributeError, KeyError, IndexError, ValueError) as e:
-            logger.warning("Failed to get images from %s: %s", project, e)
+                        tag = ""
+                    images.append((name, tag, image))
+        except (oc.OpenShiftPythonException, AttributeError, KeyError, IndexError, ValueError) as exc:
+            logger.warning("Failed to get pod images from %s: %s", project, exc)
 
         return images
 
+    @staticmethod
+    def get_subscription_versions(project) -> dict[str, str]:
+        """Get installed operator versions from OLM Subscriptions in a namespace."""
+        versions: dict[str, str] = {}
+        try:
+            with project.context:
+                subs = oc.selector("subscription.operators.coreos.com").objects()
+            for sub in subs:
+                try:
+                    csv_name = sub.model.status.installedCSV
+                    if not csv_name:
+                        continue
+                except AttributeError:
+                    continue
+                match = re.match(r"^(.+)\.v(.+)$", csv_name)
+                if match:
+                    versions[match.group(1)] = f"v{match.group(2)}"
+                else:
+                    versions[csv_name] = csv_name
+        except (oc.OpenShiftPythonException, AttributeError, KeyError, IndexError, ValueError) as exc:
+            logger.warning("Failed to get subscriptions from %s: %s", project, exc)
+        return versions
+
     @staticmethod
     def get_istio_type(cluster) -> tuple[str, Optional[str]]:
         """Determine Istio installation type via GatewayClass and namespace from Istio CRs.
@@ -155,9 +179,12 @@ def get_istio_metadata(project) -> dict[str, str]:
             with project.context:
                 istio = oc.selector("istio").objects()
                 if istio:
-                    version = istio[0].model.spec.version
-                    if version:
-                        metadata["istio_version"] = version
+                    try:
+                        version = istio[0].model.spec.version
+                        if version:
+                            metadata["istio_version"] = version
+                    except AttributeError:
+                        pass
 
                 pods = oc.selector("pods", labels={"app": "istiod"}).objects()
                 if pods:

testsuite/tests/info_collector.py

@@ -119,6 +119,8 @@ def test_kuadrant_properties(record_testsuite_property):
         cluster_data[cluster_name] = []
         kuadrant_images = ReportPortalMetadataCollector.get_component_images(project)
         for name, tag, full_image in kuadrant_images:
+            if "testsuite-pipelines-tools" in full_image:
+                continue
             if tag:
                 cluster_data[cluster_name].append(f"{name}:{tag} ({full_image})")
                 properties.append((name, tag))
@@ -129,6 +131,51 @@ def test_kuadrant_properties(record_testsuite_property):
     _record_unique(record_testsuite_property, properties)
 
 
+def test_tools_properties(record_testsuite_property):
+    """Record tools version properties from all clusters."""
+    tools_ns = "tools"
+    properties = []
+    cluster_data = {}
+    for cluster_name, _, project in _all_cluster_projects(tools_ns):
+        if project is None:
+            cluster_data[cluster_name] = [f"namespace '{tools_ns}' not found"]
+            continue
+        cluster_data[cluster_name] = []
+        tools_images = ReportPortalMetadataCollector.get_component_images(project)
+        for name, tag, full_image in tools_images:
+            if name not in {"jaeger", "redis", "dragonfly", "valkey"}:
+                continue
+            if tag:
+                cluster_data[cluster_name].append(f"{name}:{tag} ({full_image})")
+                properties.append((name, tag))
+            else:
+                cluster_data[cluster_name].append(full_image)
+
+    _print_cluster_data(cluster_data)
+    _record_unique(record_testsuite_property, properties)
+
+
+def test_tools_operator_properties(record_testsuite_property):
+    """Record OLM operator version properties from all clusters."""
+    namespaces = ["tools", "cert-manager-operator"]
+    properties = []
+    cluster_data = {}
+    for cluster_name, cluster in ReportPortalMetadataCollector.get_cluster_configurations():
+        cluster_data[cluster_name] = []
+        for ns in namespaces:
+            project = cluster.change_project(ns)
+            if not project.connected:
+                cluster_data[cluster_name].append(f"namespace '{ns}' not found")
+                continue
+            versions = ReportPortalMetadataCollector.get_subscription_versions(project)
+            for name, version in versions.items():
+                cluster_data[cluster_name].append(f"{name}:{version}")
+                properties.append((name, version))
+
+    _print_cluster_data(cluster_data)
+    _record_unique(record_testsuite_property, properties)
+
+
 def test_istio_properties(record_testsuite_property):
     """Record Istio installation type and metadata from all clusters."""
     properties = []