feat(tests): handle clusters without OSSM Istio installation

pyproject.toml

@@ -65,6 +65,7 @@ markers = [
     "egress_gateway: Test is using egress gateway",
     "min_ocp_version: Minimum OpenShift version required for test (e.g., @pytest.mark.min_ocp_version((4, 20)))",
     "gateway_api_version: Gateway API version requirement (e.g., @pytest.mark.gateway_api_version((1, 5, 0)) or @pytest.mark.gateway_api_version((1, 5, 0), operator.eq))",
+    "user_managed_istio: Test requires user-managed Istio that can be modified (skipped with OCP-managed Istio)",
 ]
 filterwarnings = [
     "ignore: WARNING the new order is not taken into account:UserWarning",

testsuite/component_metadata.py

@@ -127,6 +127,30 @@ def get_component_images(project) -> list[tuple]:
 
         return images
 
+    @staticmethod
+    def get_istio_type(cluster) -> tuple[str, Optional[str]]:
+        """Determine Istio installation type and namespace from cluster-wide Istio CRs.
+
+        Returns (istio_type, namespace) where istio_type is one of:
+        - "ocp-managed" with namespace from the Istio CR spec
+        - "user-managed" with namespace from the Istio CR spec
+        - "none" with None
+        """
+        try:
+            with cluster.context:
+                istios = oc.selector("Istio", all_namespaces=True).objects()
+                if not istios:
+                    return "none", None
+                gateway_istio = next((i for i in istios if i.name() == "openshift-gateway"), None)
+                if gateway_istio:
+                    return "ocp-managed", gateway_istio.model.spec.namespace
+                default_istio = next((i for i in istios if i.name() == "default"), None)
+                if default_istio:
+                    return "user-managed", default_istio.model.spec.namespace
+        except (oc.OpenShiftPythonException, AttributeError, KeyError) as e:
+            logger.warning("Failed to detect Istio type: %s", e)
+        return "none", None
+
     @staticmethod
     def get_istio_metadata(project) -> dict[str, str]:
         """Get Istio version and istiod image from the cluster."""

testsuite/tests/conftest.py

@@ -446,3 +446,20 @@ def check_min_ocp_version(request, openshift_version):
             pytest.skip("Could not detect OpenShift version")
         if openshift_version < required_version:
             pytest.skip(f"Requires OCP {'.'.join(map(str, required_version))}+")
+
+
+@pytest.fixture(scope="session")
+def has_user_managed_istio(cluster):
+    """True if a user-managed Istio installation exists (not OCP-managed Istio)"""
+    with cluster.context:
+        istios = selector("Istio", all_namespaces=True).objects()
+        return bool(istios) and not any(i.name() == "openshift-gateway" for i in istios)
+
+
+@pytest.fixture(autouse=True)
+def check_user_managed_istio(request, has_user_managed_istio, skip_or_fail):
+    """Skip tests that require a user-managed Istio (e.g. mTLS, sidecar injection).
+    OCP-managed Istio does not allow modifications to the Istio CR."""
+    marker = request.node.get_closest_marker("user_managed_istio")
+    if marker and not has_user_managed_istio:
+        skip_or_fail("Test requires user-managed Istio installation")

testsuite/tests/info_collector.py

@@ -130,14 +130,22 @@ def test_kuadrant_properties(record_testsuite_property):
 
 
 def test_istio_properties(record_testsuite_property):
-    """Record Istio related properties from all clusters."""
+    """Record Istio installation type and metadata from all clusters."""
     properties = []
     cluster_data = {}
-    for cluster_name, _, project in _all_cluster_projects("istio-system"):
-        if project is None:
-            cluster_data[cluster_name] = ["namespace 'istio-system' not found"]
+    for cluster_name, cluster, _ in _all_cluster_projects("default"):
+        istio_type, namespace = ReportPortalMetadataCollector.get_istio_type(cluster)
+        cluster_data[cluster_name] = [f"istio_type:{istio_type}"]
+        properties.append(("istio_type", istio_type))
+
+        if namespace is None:
             continue
-        cluster_data[cluster_name] = []
+
+        project = cluster.change_project(namespace)
+        if not project.connected:
+            cluster_data[cluster_name].append(f"namespace '{namespace}' not found")
+            continue
+
         istio_metadata = ReportPortalMetadataCollector.get_istio_metadata(project)
         for key, value in istio_metadata.items():
             cluster_data[cluster_name].append(f"{key}:{value}")

testsuite/tests/singlecluster/authorino/identity/x509/test_xfcc_forwarding.py

@@ -12,7 +12,7 @@
 from testsuite.gateway.gateway_api.gateway import KuadrantGateway
 from .conftest import XFCC_HEADER_NAME
 
-pytestmark = [pytest.mark.authorino, pytest.mark.kuadrant_only]
+pytestmark = [pytest.mark.authorino, pytest.mark.kuadrant_only, pytest.mark.user_managed_istio]
 
 
 @pytest.fixture(scope="module")

testsuite/tests/singlecluster/gateway/mtls/test_mtls_behaviour.py

@@ -4,7 +4,7 @@
 
 import pytest
 
-pytestmark = [pytest.mark.disruptive]
+pytestmark = [pytest.mark.disruptive, pytest.mark.user_managed_istio]
 
 component_cases = [
     pytest.param(["limitador"], id="limitador-only"),

testsuite/tests/singlecluster/gateway/mtls/test_mtls_configuration.py

@@ -5,7 +5,7 @@
 import pytest
 from openshift_client import selector
 
-pytestmark = [pytest.mark.disruptive]
+pytestmark = [pytest.mark.disruptive, pytest.mark.user_managed_istio]
 
 component_cases = [
     pytest.param(["limitador"], id="limitador-only"),

testsuite/tests/singlecluster/tracing/data_plane_tracing/test_kuadrant_tracing.py

@@ -3,6 +3,10 @@
 
 This module validates that tracing correctly captures request flows through the entire
 Kuadrant stack, including wasm-shim, Authorino, Limitador, and gateway services.
+
+With OCP-managed Istio, gateway traces are not available because the Istio CR cannot be
+modified to enable tracing (meshConfig.enableTracing, extensionProviders) and the Telemetry
+resource cannot be created. Gateway service assertions are conditional on user-managed Istio.
 """
 
 import os
@@ -39,37 +43,39 @@ def trace_request_ids(client, auth):
 
 
 @pytest.fixture(scope="module")
-def trace_200(trace_request_ids, tracing):
+def trace_200(trace_request_ids, tracing, has_user_managed_istio):
     """Fetches and caches the full wasm-shim trace for the 200 response."""
     request_id = trace_request_ids[0]
-    traces = tracing.get_traces(service="wasm-shim", min_processes=4, tags={"request_id": request_id})
+    min_procs = 4 if has_user_managed_istio else 3
+    traces = tracing.get_traces(service="wasm-shim", min_processes=min_procs, tags={"request_id": request_id})
     assert len(traces) == 1, f"No trace was found in tracing backend with request_id: {request_id}"
     return traces[0]
 
 
 @pytest.fixture(scope="module")
-def trace_429(trace_request_ids, tracing):
+def trace_429(trace_request_ids, tracing, has_user_managed_istio):
     """Fetches and caches the full wasm-shim trace for the 429 response."""
     request_id = trace_request_ids[1]
-    traces = tracing.get_traces(service="wasm-shim", min_processes=4, tags={"request_id": request_id})
+    min_procs = 4 if has_user_managed_istio else 3
+    traces = tracing.get_traces(service="wasm-shim", min_processes=min_procs, tags={"request_id": request_id})
     assert len(traces) == 1, f"No trace was found in tracing backend with request_id: {request_id}"
     return traces[0]
 
 
 @pytest.fixture(scope="module")
-def trace_401(client, tracing):
-    """
-    Sends request producing 401 response and fetches the full wasm-shim trace"""
+def trace_401(client, tracing, has_user_managed_istio):
+    """Sends request producing 401 response and fetches the full wasm-shim trace"""
     response_401 = client.get("/get", headers={"Traceparent": f"00-{os.urandom(16).hex()}-{os.urandom(8).hex()}-01"})
     assert response_401.status_code == 401
 
     request_id = response_401.headers.get("x-request-id")
-    traces = tracing.get_traces(service="wasm-shim", min_processes=3, tags={"request_id": request_id})
+    min_procs = 3 if has_user_managed_istio else 2
+    traces = tracing.get_traces(service="wasm-shim", min_processes=min_procs, tags={"request_id": request_id})
     assert len(traces) == 1, f"No trace was found in tracing backend with request_id: {request_id}"
     return traces[0]
 
 
-def test_trace_includes_all_kuadrant_services(trace_200, label):
+def test_trace_includes_all_kuadrant_services(trace_200, label, has_user_managed_istio):
     """
     Test that distributed tracing captures all Kuadrant components in a single trace.
 
@@ -78,28 +84,32 @@ def test_trace_includes_all_kuadrant_services(trace_200, label):
     - wasm-shim: WASM plugin processing requests
     - authorino: Authorization service
     - limitador: Rate limiting service
-    - gateway: Istio/Envoy gateway service
+    - gateway: Istio/Envoy gateway service (only with user-managed Istio)
     """
 
     process_services = trace_200.get_process_services()
 
-    services = ["wasm-shim", "authorino", "limitador", f"{label}.kuadrant"]
+    services = ["wasm-shim", "authorino", "limitador"]
+    if has_user_managed_istio:
+        services.append(f"{label}.kuadrant")
     for service in services:
         assert service in process_services, f"Service '{service}' not found in trace processes: {process_services}"
 
 
-def test_relevant_services_on_auth_denied(trace_401, label):
+def test_relevant_services_on_auth_denied(trace_401, label, has_user_managed_istio):
     """
     Test that auth-denied traces only include services up to the authorization step.
 
     When a request is rejected with 401, the trace should contain wasm-shim and authorino
     but not limitador, since the request is short-circuited before reaching the rate limiter.
-    Note: gateway service is not included since the request was sent without traceparent header.
+    Gateway service is only present with user-managed Istio.
     """
 
     process_services = trace_401.get_process_services()
 
-    services = ["wasm-shim", "authorino", f"{label}.kuadrant"]
+    services = ["wasm-shim", "authorino"]
+    if has_user_managed_istio:
+        services.append(f"{label}.kuadrant")
     for service in services:
         assert service in process_services, f"Service '{service}' not found in trace processes: {process_services}"
 

testsuite/tests/singlecluster/tracing/data_plane_tracing/test_kuadrant_tracing_rate_limit_only.py

@@ -1,5 +1,9 @@
 """
 Tests for distributed tracing with only a RateLimitPolicy configured (no AuthPolicy).
+
+With OCP-managed Istio, gateway traces are not available because the Istio CR cannot be
+modified to enable tracing (meshConfig.enableTracing, extensionProviders) and the Telemetry
+resource cannot be created. Gateway service assertions are conditional on user-managed Istio.
 """
 
 import os
@@ -28,7 +32,7 @@ def commit(request, rate_limit):
 
 
 @pytest.fixture(scope="module")
-def trace_429(client, tracing):
+def trace_429(client, tracing, has_user_managed_istio):
     """
     Sends requests to exhaust the rate limit, produces a 429 response,
     and fetches the full wasm-shim trace.
@@ -40,19 +44,23 @@ def trace_429(client, tracing):
     assert response_429.status_code == 429
 
     request_id = response_429.headers.get("x-request-id")
-    traces = tracing.get_traces(service="wasm-shim", min_processes=3, tags={"request_id": request_id})
+    min_procs = 3 if has_user_managed_istio else 2
+    traces = tracing.get_traces(service="wasm-shim", min_processes=min_procs, tags={"request_id": request_id})
     assert len(traces) == 1, f"No trace was found in tracing backend with request_id: {request_id}"
     return traces[0]
 
 
-def test_relevant_services_rate_limit_only(trace_429, label):
+def test_relevant_services_rate_limit_only(trace_429, label, has_user_managed_istio):
     """
-    Test that traces with only a RateLimitPolicy include all relevant services (wasm-shim, limitador, and gateway).
+    Test that traces with only a RateLimitPolicy include relevant Kuadrant services.
+    Gateway service is only present with user-managed Istio.
     Trace should not contain authorino since no authorization is involved.
     """
 
     process_services = trace_429.get_process_services()
-    services = ["wasm-shim", "limitador", f"{label}.kuadrant"]
+    services = ["wasm-shim", "limitador"]
+    if has_user_managed_istio:
+        services.append(f"{label}.kuadrant")
     for service in services:
         assert service in process_services, f"Service '{service}' not found in trace processes: {process_services}"