[pre-commit.ci] pre-commit autoupdate (#450)

* [pre-commit.ci] pre-commit autoupdate

updates:
- [github.com/astral-sh/ruff-pre-commit: v0.15.1 → v0.15.2](astral-sh/ruff-pre-commit@v0.15.1...v0.15.2)

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

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

* fix: resolve ruff v0.15.2 linting errors

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Meni Yakove <myakove@gmail.com>

Author: pre-commit-ci[bot]

.pre-commit-config.yaml

@@ -42,7 +42,7 @@ repos:
       - id: detect-secrets
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.15.1
+    rev: v0.15.2
     hooks:
       - id: ruff
         args: [--fix]

ocp_utilities/cluster_versions.py

@@ -1,14 +1,14 @@
 from __future__ import annotations
+
 import functools
 import re
-from typing import Dict, List
 
 import requests
 from bs4 import BeautifulSoup, ResultSet
 from kubernetes.dynamic import DynamicClient
 from ocp_resources.cluster_version import ClusterVersion
-from simple_logger.logger import get_logger
 from semver import Version
+from simple_logger.logger import get_logger
 
 from ocp_utilities.exceptions import ClusterVersionNotFoundError
 
@@ -25,7 +25,7 @@ def parse_openshift_release_url() -> ResultSet:
 
 
 @functools.cache
-def get_accepted_cluster_versions() -> Dict[str, Dict[str, List[str]]]:
+def get_accepted_cluster_versions() -> dict[str, dict[str, list[str]]]:
     """
     Get all accepted cluster versions from https://openshift-release.apps.ci.l2s4.p1.openshiftapps.com
 
@@ -48,7 +48,7 @@ def get_accepted_cluster_versions() -> Dict[str, Dict[str, List[str]]]:
          'rc': {'4.15': ['4.15.0-0.rc-2022-05-25-113430']},
          'fc': {'4.15': ['4.15.0-0.fc-2022-05-25-113430']}}
     """
-    _accepted_version_dict: Dict[str, Dict[str, List[str]]] = {}
+    _accepted_version_dict: dict[str, dict[str, list[str]]] = {}
     for tr in parse_openshift_release_url():
         version, status = [_tr for _tr in tr.text.splitlines() if _tr][:2]
         if status == "Accepted":

ocp_utilities/debugger.py

@@ -16,4 +16,4 @@ class WebDebugger(WebPdb):
     """
 
     def __init__(self) -> None:
-        super().__init__(host="0.0.0.0", port=int(os.environ.get("PYTHON_REMOTE_DEBUG_PORT", 1212)))
+        super().__init__(host="0.0.0.0", port=int(os.environ.get("PYTHON_REMOTE_DEBUG_PORT", "1212")))

ocp_utilities/infra.py

@@ -6,7 +6,7 @@
 import os
 import re
 import shlex
-from typing import Any, Dict, List, Optional
+from typing import Any
 
 import urllib3
 from deprecation import deprecated
@@ -44,7 +44,7 @@ def get_client(**kwargs: Any) -> DynamicClient:
     return get_dynamic_client(**kwargs)
 
 
-def assert_nodes_ready(nodes: List[Node]) -> None:
+def assert_nodes_ready(nodes: list[Node]) -> None:
     """
     Validates all nodes are in ready
 
@@ -60,7 +60,7 @@ def assert_nodes_ready(nodes: List[Node]) -> None:
         raise NodeNotReadyError(f"Following nodes are not in ready state: {not_ready_nodes}")
 
 
-def assert_nodes_schedulable(nodes: List[Node]) -> None:
+def assert_nodes_schedulable(nodes: list[Node]) -> None:
     """
     Validates all nodes are in schedulable state
 
@@ -76,7 +76,7 @@ def assert_nodes_schedulable(nodes: List[Node]) -> None:
         raise NodeUnschedulableError(f"Following nodes are in unscheduled state: {unschedulable_nodes}")
 
 
-def assert_pods_failed_or_pending(pods: List[Pod]) -> None:
+def assert_pods_failed_or_pending(pods: list[Pod]) -> None:
     """
     Validates all pods are not in failed nor pending phase
 
@@ -103,8 +103,8 @@ def assert_pods_failed_or_pending(pods: List[Pod]) -> None:
 
 
 def assert_nodes_in_healthy_condition(
-    nodes: List[Node],
-    healthy_node_condition_type: Optional[Dict[str, str]],
+    nodes: list[Node],
+    healthy_node_condition_type: dict[str, str] | None,
 ) -> None:
     """
     Validates nodes are in a healthy condition.
@@ -169,7 +169,7 @@ class DynamicClassCreator:
     """
 
     def __init__(self) -> None:
-        self.created_classes: Dict[Any, Any] = {}
+        self.created_classes: dict[Any, Any] = {}
 
     def __call__(self, base_class: Any) -> Any:  # TODO: return `BaseResource` class
         if base_class in self.created_classes:
@@ -206,7 +206,7 @@ def clean_up(self) -> bool:
                             resource_object=self,
                             collect_pod_logs=data_collector_dict.get("collect_pod_logs", False),
                         )
-                except Exception as exception_:
+                except Exception as exception_:  # noqa: BLE001
                     LOGGER.warning(
                         f"[Data collector] failed to collect data for {self.kind} {self.name}\nexception: {exception_}"
                     )
@@ -309,13 +309,13 @@ def create_icsp_from_file(icsp_file_path: str) -> ImageContentSourcePolicy:
     return icsp
 
 
-def create_icsp(icsp_name: str, repository_digest_mirrors: List[Dict[str, Any]]) -> ImageContentSourcePolicy:
+def create_icsp(icsp_name: str, repository_digest_mirrors: list[dict[str, Any]]) -> ImageContentSourcePolicy:
     icsp = ImageContentSourcePolicy(name=icsp_name, repository_digest_mirrors=repository_digest_mirrors)
     icsp.deploy()
     return icsp
 
 
-def dict_base64_encode(_dict: Dict[Any, Any]) -> str:
+def dict_base64_encode(_dict: dict[Any, Any]) -> str:
     """
     Encoding dict in base64
 
@@ -329,7 +329,7 @@ def dict_base64_encode(_dict: Dict[Any, Any]) -> str:
 
 
 def create_update_secret(
-    secret_data_dict: Dict[str, Dict[str, Dict[str, str]]],
+    secret_data_dict: dict[str, dict[str, dict[str, str]]],
     name: str,
     namespace: str,
     admin_client: DynamicClient = None,
@@ -374,7 +374,7 @@ def create_update_secret(
     return secret.deploy()
 
 
-def get_pods_by_name_prefix(client: DynamicClient, pod_prefix: str, namespace: str) -> List[Pod]:
+def get_pods_by_name_prefix(client: DynamicClient, pod_prefix: str, namespace: str) -> list[Pod]:
     """
     Args:
         client (DynamicClient): OCP Client to use.

ocp_utilities/monitoring.py

@@ -1,25 +1,24 @@
 from __future__ import annotations
+
 import json
 import re
 from json import JSONDecodeError
-from typing import Any, Dict, List
+from typing import Any
 
 import requests
 from kubernetes.dynamic import DynamicClient
+from ocp_resources.resource import get_client
 from ocp_resources.route import Route
-from timeout_sampler import TimeoutExpiredError, TimeoutSampler
 from simple_logger.logger import get_logger
-
-from ocp_resources.resource import get_client
-
+from timeout_sampler import TimeoutExpiredError, TimeoutSampler
 
 TIMEOUT_2MIN = 2 * 60
 TIMEOUT_10MIN = 10 * 60
 
 LOGGER = get_logger(name=__name__)
 
 
-class Prometheus(object):
+class Prometheus:
     """
     For accessing Prometheus cluster metrics
 
@@ -69,7 +68,7 @@ def _get_route(self) -> str:
         route = Route(namespace=self.namespace, name=self.resource_name, client=self.client).instance.spec.host
         return f"https://{route}"
 
-    def _get_response(self, query: str) -> Dict[str, Any]:
+    def _get_response(self, query: str) -> dict[str, Any]:
         response = requests.get(f"{self.api_url}{query}", headers=self.headers, verify=self.verify_ssl)
 
         try:
@@ -81,7 +80,7 @@ def _get_response(self, query: str) -> Dict[str, Any]:
             )
             raise
 
-    def query(self, query: str) -> Dict[str, Any]:
+    def query(self, query: str) -> dict[str, Any]:
         """
         get the prometheus query result
 
@@ -93,7 +92,7 @@ def query(self, query: str) -> Dict[str, Any]:
         """
         return self._get_response(query=f"{self.api_v1}/query?query={query}")
 
-    def get_all_alerts_by_alert_name(self, alert_name: str) -> List[Dict[str, Any]]:
+    def get_all_alerts_by_alert_name(self, alert_name: str) -> list[dict[str, Any]]:
         """
         Get alert by alert name if it's an active alert
 
@@ -107,13 +106,9 @@ def get_all_alerts_by_alert_name(self, alert_name: str) -> List[Dict[str, Any]]:
              list: list containing alert metrics
         """
         alerts = self.alerts()
-        alert_list = []
-        for alert in alerts["data"]["alerts"]:
-            if alert["labels"]["alertname"] == alert_name:
-                alert_list.append(alert)
-        return alert_list
+        return [alert for alert in alerts["data"]["alerts"] if alert["labels"]["alertname"] == alert_name]
 
-    def get_firing_alerts(self, alert_name: str) -> List[Dict[str, Any]]:
+    def get_firing_alerts(self, alert_name: str) -> list[dict[str, Any]]:
         """
         get all the firing alerts from list of active alerts
         """
@@ -145,14 +140,17 @@ def get_scrape_interval(self) -> int:
         response = self._get_response(query=f"{self.api_v1}/targets")
         result = response.get("data", {}).get("activeTargets", [])
         for item in result:
-            if item and item.get("labels", {}).get("job") == "prometheus-k8s":
-                if scrape_interval := item.get("scrapeInterval"):
-                    if scrape_interval_match := re.match(r"\d+", scrape_interval):
-                        return int(scrape_interval_match.group())
+            if (
+                item
+                and item.get("labels", {}).get("job") == "prometheus-k8s"
+                and (scrape_interval := item.get("scrapeInterval"))
+                and (scrape_interval_match := re.match(r"\d+", scrape_interval))
+            ):
+                return int(scrape_interval_match.group())
 
         return 30
 
-    def query_sampler(self, query: str, timeout: int = TIMEOUT_2MIN) -> List[Any]:
+    def query_sampler(self, query: str, timeout: int = TIMEOUT_2MIN) -> list[Any]:
         """
         Sample output for query function
 
@@ -183,13 +181,13 @@ def query_sampler(self, query: str, timeout: int = TIMEOUT_2MIN) -> List[Any]:
 
         return []
 
-    def alerts(self) -> Dict[str, Any]:
+    def alerts(self) -> dict[str, Any]:
         """
         get all the active alerts
         """
         return self._get_response(query=f"{self.api_v1}/alerts")
 
-    def get_alerts_by_state(self, alert_name: str, state: str = "firing") -> List[Dict[str, Any]]:
+    def get_alerts_by_state(self, alert_name: str, state: str = "firing") -> list[dict[str, Any]]:
         """
         get all the alerts from list of active alerts according the state
         """
@@ -198,7 +196,7 @@ def get_alerts_by_state(self, alert_name: str, state: str = "firing") -> List[Di
 
     def wait_for_alert_by_state_sampler(
         self, alert_name: str, timeout: int = TIMEOUT_10MIN, state: str = "firing"
-    ) -> List[Dict[str, Any]]:
+    ) -> list[dict[str, Any]]:
         """
         Sample output for an alert if found in the state provided in the args.
 

ocp_utilities/must_gather.py

@@ -1,12 +1,11 @@
+import os
 import shlex
 import shutil
 from pathlib import Path
-import os
 
 from pyhelper_utils.shell import run_command
 from simple_logger.logger import get_logger
 
-
 LOGGER = get_logger(name=__name__)
 
 
@@ -89,7 +88,7 @@ def collect_must_gather(
         )
         LOGGER.success("must-gather collected")
 
-    except Exception as ex:
+    except Exception as ex:  # noqa: BLE001
         LOGGER.error(
             f"Failed to run must-gather \n{ex}",
         )

ocp_utilities/operators.py

@@ -1,5 +1,7 @@
+from __future__ import annotations
+
 from pprint import pformat
-from typing import Any, Dict, List, Optional
+from typing import Any
 
 from kubernetes.dynamic import DynamicClient
 from kubernetes.dynamic.exceptions import ResourceNotFoundError
@@ -12,13 +14,12 @@
 from ocp_resources.operator_group import OperatorGroup
 from ocp_resources.resource import ResourceEditor
 from ocp_resources.subscription import Subscription
-from timeout_sampler import TimeoutExpiredError, TimeoutSampler, TimeoutWatch
 from ocp_resources.validating_webhook_config import ValidatingWebhookConfiguration
 from simple_logger.logger import get_logger
-from ocp_utilities.must_gather import collect_must_gather
+from timeout_sampler import TimeoutExpiredError, TimeoutSampler, TimeoutWatch
 
 from ocp_utilities.infra import cluster_resource, create_icsp, create_update_secret
-
+from ocp_utilities.must_gather import collect_must_gather
 
 LOGGER = get_logger(name=__name__)
 TIMEOUT_5MIN = 5 * 60
@@ -148,7 +149,7 @@ def get_csv_by_name(admin_client: DynamicClient, csv_name: str, namespace: str)
 
 def install_operator(
     admin_client: DynamicClient,
-    target_namespaces: Optional[List[str]],
+    target_namespaces: list[str] | None,
     name: str,
     channel: str,
     source: str = "",
@@ -190,9 +191,8 @@ def install_operator(
     catalog_source = None
     operator_market_namespace = "openshift-marketplace"
 
-    if must_gather_output_dir:
-        if not cluster_name:
-            raise ValueError("'cluster_name' param is required for running must-gather of cluster")
+    if must_gather_output_dir and not cluster_name:
+        raise ValueError("'cluster_name' param is required for running must-gather of cluster")
     try:
         if iib_index_image:
             if not brew_token:
@@ -345,7 +345,7 @@ def create_catalog_source_for_iib_install(
 
     def _manipulate_validating_webhook_configuration(
         _validating_webhook_configuration: ValidatingWebhookConfiguration,
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         _resource_name = "imagecontentsourcepolicies"
         _validating_webhook_configuration_dict = _validating_webhook_configuration.instance.to_dict()
         for webhook in _validating_webhook_configuration_dict["webhooks"]:
@@ -358,7 +358,7 @@ def _manipulate_validating_webhook_configuration(
 
         return _validating_webhook_configuration_dict
 
-    def _icsp(_repository_digest_mirrors: List[Dict[str, Any]]) -> None:
+    def _icsp(_repository_digest_mirrors: list[dict[str, Any]]) -> None:
         if icsp.exists:
             ResourceEditor(
                 patches={icsp: {"spec:": {"repository_digest_mirrors": _repository_digest_mirrors}}}
@@ -375,7 +375,7 @@ def _icsp(_repository_digest_mirrors: List[Dict[str, Any]]) -> None:
     _iib_index_image = iib_index_image.replace(source_iib_registry, brew_registry)
     icsp = ImageContentSourcePolicy(name="brew-registry")
     validating_webhook_configuration = ValidatingWebhookConfiguration(name="sre-imagecontentpolicies-validation")
-    repository_digest_mirrors: List[Dict[str, Any]] = [
+    repository_digest_mirrors: list[dict[str, Any]] = [
         {
             "source": f"{source_iib_registry}/{brew_image_repo}",
             "mirrors": [f"{brew_registry}/{brew_image_repo}"],
@@ -412,13 +412,12 @@ def _icsp(_repository_digest_mirrors: List[Dict[str, Any]]) -> None:
         admin_client=admin_client,
     )
 
-    iib_catalog_source = create_catalog_source_from_image(
+    return create_catalog_source_from_image(
         admin_client=admin_client,
         name=name,
         namespace=operator_market_namespace,
         image=_iib_index_image,
     )
-    return iib_catalog_source
 
 
 def create_catalog_source_from_image(