[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Author: pre-commit-ci[bot]

tests/model_explainability/guardrails/conftest.py

@@ -2,37 +2,28 @@
 
 import portforward
 import pytest
-import yaml
-from _pytest.fixtures import FixtureRequest
 from kubernetes.dynamic import DynamicClient
 from kubernetes.dynamic.exceptions import ResourceNotFoundError
 from ocp_resources.cluster_service_version import ClusterServiceVersion
-from ocp_resources.config_map import ConfigMap
 from ocp_resources.deployment import Deployment
-from ocp_resources.exceptions import MissingRequiredArgumentError
-from ocp_resources.guardrails_orchestrator import GuardrailsOrchestrator
 from ocp_resources.inference_service import InferenceService
-from ocp_resources.jaeger import Jaeger
 from ocp_resources.namespace import Namespace
 from ocp_resources.open_telemetry_collector import OpenTelemetryCollector
 from ocp_resources.persistent_volume_claim import PersistentVolumeClaim
 from ocp_resources.pod import Pod
-from ocp_resources.resource import ResourceEditor, NamespacedResource
 from ocp_resources.route import Route
 from ocp_resources.secret import Secret
 from ocp_resources.service import Service
 from ocp_resources.serving_runtime import ServingRuntime
-from ocp_resources.subscription import Subscription
 from ocp_utilities.operators import install_operator, uninstall_operator
-from pytest_testconfig import py_config
 from timeout_sampler import TimeoutSampler
 
 from tests.model_explainability.guardrails.constants import AUTOCONFIG_DETECTOR_LABEL
 from utilities.certificates_utils import create_ca_bundle_file
 from utilities.constants import (
     KServeDeploymentType,
-    Labels,
-    Timeout, OPENSHIFT_OPERATORS, RuntimeTemplates, Protocols,
+    Timeout,
+    RuntimeTemplates,
 )
 from utilities.inference_utils import create_isvc, LOGGER
 from utilities.operator_utils import get_cluster_service_version
@@ -157,10 +148,9 @@ def hap_detector_route(
         wait_for_resource=True,
     )
 
+
 @pytest.fixture(scope="class")
-def installed_tempo_operator(
-        admin_client: DynamicClient, model_namespace: Namespace
-) -> Generator[None, Any, None]:
+def installed_tempo_operator(admin_client: DynamicClient, model_namespace: Namespace) -> Generator[None, Any, None]:
     """
     Installs the Tempo operator and waits for its deployment.
     """
@@ -196,13 +186,10 @@ def installed_tempo_operator(
         clean_up_namespace=False,
     )
 
+
 @pytest.fixture(scope="class")
 def tempo_stack(
-    admin_client: DynamicClient,
-    installed_tempo_operator: None,
-    model_namespace: Namespace,
-    minio_secret_otel: Secret
-
+    admin_client: DynamicClient, installed_tempo_operator: None, model_namespace: Namespace, minio_secret_otel: Secret
 ) -> Generator[Any, Any, None]:
     """
     Create a TempoStack CR in the test namespace, configured to use the MinIO backend.
@@ -225,14 +212,11 @@ def tempo_stack(
     tempo_stack_dict: dict[str, Any] = next(
         example
         for example in alm_examples
-        if example["kind"] == "TempoStack"
-        and example["apiVersion"].startswith("tempo.grafana.com/")
+        if example["kind"] == "TempoStack" and example["apiVersion"].startswith("tempo.grafana.com/")
     )
 
     if not tempo_stack_dict:
-        raise ResourceNotFoundError(
-            f"No TempoStack dict found in alm_examples for CSV {tempo_csv.name}"
-        )
+        raise ResourceNotFoundError(f"No TempoStack dict found in alm_examples for CSV {tempo_csv.name}")
 
     # Customize metadata
     tempo_stack_dict["metadata"]["namespace"] = model_namespace.name
@@ -274,6 +258,7 @@ def tempo_stack(
     )
     yield created_tempo_stack
 
+
 def wait_for_tempo_pods(
     client: DynamicClient,
     tempo_name: str,
@@ -283,6 +268,7 @@ def wait_for_tempo_pods(
     """
     Wait for pods created by a Tempo instance to be ready.
     """
+
     def _get_tempo_pods() -> list[Pod]:
         return [
             _pod
@@ -293,9 +279,7 @@ def _get_tempo_pods() -> list[Pod]:
             )
         ]
 
-    sampler = TimeoutSampler(
-        wait_timeout=timeout, sleep=1, func=lambda: bool(_get_tempo_pods())
-    )
+    sampler = TimeoutSampler(wait_timeout=timeout, sleep=1, func=lambda: bool(_get_tempo_pods()))
 
     for sample in sampler:
         if sample:
@@ -308,6 +292,7 @@ def _get_tempo_pods() -> list[Pod]:
             status="True",
         )
 
+
 @pytest.fixture(scope="class")
 def installed_opentelemetry_operator(admin_client: DynamicClient) -> Generator[None, Any, None]:
     """
@@ -346,13 +331,14 @@ def installed_opentelemetry_operator(admin_client: DynamicClient) -> Generator[N
         clean_up_namespace=False,
     )
 
+
 @pytest.fixture(scope="class")
 def otel_collector(
     admin_client: DynamicClient,
     installed_opentelemetry_operator: None,
     tempo_stack,
     model_namespace: Namespace,
-    minio_service_otel
+    minio_service_otel,
 ) -> Generator[OpenTelemetryCollector, Any, Any]:
     """
     Create an OpenTelemetryCollector CR in the test namespace.
@@ -370,14 +356,11 @@ def otel_collector(
     otel_cr_dict: dict[str, Any] = next(
         example
         for example in alm_examples
-        if example["kind"] == "OpenTelemetryCollector"
-        and example["apiVersion"] == "opentelemetry.io/v1beta1"
+        if example["kind"] == "OpenTelemetryCollector" and example["apiVersion"] == "opentelemetry.io/v1beta1"
     )
 
     if not otel_cr_dict:
-        raise ResourceNotFoundError(
-            f"No OpenTelemetryCollector example found in ALM examples for {otel_csv.name}"
-        )
+        raise ResourceNotFoundError(f"No OpenTelemetryCollector example found in ALM examples for {otel_csv.name}")
 
     # Update the metadata and spec to match test namespace and Tempo endpoint
     namespace = model_namespace.name
@@ -408,17 +391,7 @@ def otel_collector(
                 }
             },
             "telemetry": {
-                "metrics": {
-                    "readers": [
-                        {
-                            "pull": {
-                                "exporter": {
-                                    "prometheus": {"host": "0.0.0.0", "port": 8888}
-                                }
-                            }
-                        }
-                    ]
-                }
+                "metrics": {"readers": [{"pull": {"exporter": {"prometheus": {"host": "0.0.0.0", "port": 8888}}}}]}
             },
         },
     }
@@ -428,6 +401,7 @@ def otel_collector(
         wait_for_collector_pods(admin_client, namespace=namespace)
         yield otel_cr
 
+
 def wait_for_collector_pods(
     client: DynamicClient,
     namespace: str,
@@ -443,14 +417,12 @@ def _get_collector_pods() -> list[Pod]:
             for _pod in Pod.get(
                 dyn_client=client,
                 namespace=namespace,
-                label_selector=f"app.kubernetes.io/component=opentelemetry-collector",
+                label_selector="app.kubernetes.io/component=opentelemetry-collector",
             )
         ]
         return pods
 
-    sampler = TimeoutSampler(
-        wait_timeout=timeout, sleep=1, func=lambda: bool(_get_collector_pods())
-    )
+    sampler = TimeoutSampler(wait_timeout=timeout, sleep=1, func=lambda: bool(_get_collector_pods()))
 
     for sample in sampler:
         if sample:
@@ -463,6 +435,7 @@ def _get_collector_pods() -> list[Pod]:
             status="True",
         )
 
+
 @pytest.fixture(scope="class")
 def minio_pvc_otel(
     admin_client: DynamicClient,
@@ -478,34 +451,39 @@ def minio_pvc_otel(
         "client": admin_client,
         "size": "2Gi",
         "accessmodes": "ReadWriteOnce",
-        "label": {"app.kubernetes.io/name": "minio"}
+        "label": {"app.kubernetes.io/name": "minio"},
     }
 
     with PersistentVolumeClaim(**pvc_kwargs) as pvc:
         pvc.wait_for_status(status=pvc.Status.PENDING, timeout=120)
         yield pvc
 
+
 @pytest.fixture(scope="class")
 def minio_deployment_otel(admin_client, model_namespace):
     selector = {"matchLabels": {"app.kubernetes.io/name": "minio"}}
     pod_template = {
         "metadata": {"labels": {"app.kubernetes.io/name": "minio"}},
         "spec": {
-            "containers": [{
-                "name": "minio",
-                "image": "quay.io/minio/minio",
-                "command": ["/bin/sh", "-c", "mkdir -p /storage/tempo && minio server /storage"],
-                "env": [
-                    {"name": "MINIO_ACCESS_KEY", "value": "tempo"},
-                    {"name": "MINIO_SECRET_KEY", "value": "supersecret"},
-                ],
-                "ports": [{"containerPort": 9000}],
-                "volumeMounts": [{"mountPath": "/storage", "name": "storage"}],
-            }],
-            "volumes": [{
-                "name": "storage",
-                "persistentVolumeClaim": {"claimName": "minio"},
-            }],
+            "containers": [
+                {
+                    "name": "minio",
+                    "image": "quay.io/minio/minio",
+                    "command": ["/bin/sh", "-c", "mkdir -p /storage/tempo && minio server /storage"],
+                    "env": [
+                        {"name": "MINIO_ACCESS_KEY", "value": "tempo"},
+                        {"name": "MINIO_SECRET_KEY", "value": "supersecret"},
+                    ],
+                    "ports": [{"containerPort": 9000}],
+                    "volumeMounts": [{"mountPath": "/storage", "name": "storage"}],
+                }
+            ],
+            "volumes": [
+                {
+                    "name": "storage",
+                    "persistentVolumeClaim": {"claimName": "minio"},
+                }
+            ],
         },
     }
 
@@ -551,6 +529,7 @@ def minio_service_otel(admin_client, model_namespace):
 
     yield service
 
+
 @pytest.fixture(scope="class")
 def minio_secret_otel(admin_client, model_namespace):
     secret = Secret(
@@ -578,11 +557,13 @@ def otelcol_metrics_endpoint(admin_client: DynamicClient, model_namespace: Names
     """
     namespace = model_namespace.name
 
-    service = next(Service.get(
-        dyn_client=admin_client,
-        namespace=namespace,
-        label_selector="app.kubernetes.io/component=opentelemetry-collector",
-    ))
+    service = next(
+        Service.get(
+            dyn_client=admin_client,
+            namespace=namespace,
+            label_selector="app.kubernetes.io/component=opentelemetry-collector",
+        )
+    )
 
     service_name = service.name
 
@@ -600,6 +581,7 @@ def tempo_traces_endpoint(tempo_stack, model_namespace: Namespace):
     port = 4317
     return f"http://{service_name}.{namespace}.svc.cluster.local:{port}"
 
+
 @pytest.fixture(scope="class")
 def otel_exporter_config(otelcol_metrics_endpoint, tempo_traces_endpoint):
     return {
@@ -610,6 +592,7 @@ def otel_exporter_config(otelcol_metrics_endpoint, tempo_traces_endpoint):
         "tracesProtocol": "grpc",
     }
 
+
 @pytest.fixture(scope="class")
 def tempo_traces_service_portforward(
     admin_client: DynamicClient,
@@ -638,4 +621,4 @@ def tempo_traces_service_portforward(
             yield local_url
     except Exception as e:
         LOGGER.error(f"Failed to set up port forwarding for {service_name}: {e}")
-        raise
+        raise