fix: Add missing typing for already typed functions (#2451)

Author: myakove

class_generator/class_generator.py

@@ -1,5 +1,3 @@
-from __future__ import annotations
-
 import ast
 import filecmp
 import json
@@ -143,7 +141,7 @@ def read_resources_mapping_file() -> dict[Any, Any]:
         return {}
 
 
-def get_server_version(client: str):
+def get_server_version(client: str) -> str:
     rc, out, _ = run_command(command=shlex.split(f"{client} version -o json"), check=False)
     if not rc:
         LOGGER.error("Failed to get server version")

ocp_resources/application_aware_cluster_resource_quota.py

@@ -1,7 +1,6 @@
 # API reference: https://github.com/kubevirt/application-aware-quota/tree/main/staging/src/kubevirt.io/application-aware-quota-api/pkg/apis/core/v1alpha1
 # TODO: update API reference when OCP doc is available
-from __future__ import annotations
-from typing import Dict, Any
+from typing import Any
 
 from ocp_resources.resource import MissingRequiredArgumentError, Resource
 
@@ -11,8 +10,8 @@ class ApplicationAwareClusterResourceQuota(Resource):
 
     def __init__(
         self,
-        quota: Dict[str, Any] | None = None,
-        selector: Dict[str, Any] | None = None,
+        quota: dict[str, Any] | None = None,
+        selector: dict[str, Any] | None = None,
         **kwargs: Any,
     ) -> None:
         """

ocp_resources/application_aware_resource_quota.py

@@ -1,7 +1,6 @@
 # API reference: https://github.com/kubevirt/application-aware-quota/tree/main/staging/src/kubevirt.io/application-aware-quota-api/pkg/apis/core/v1alpha1
 # TODO: update API reference when OCP doc is available
-from __future__ import annotations
-from typing import Dict, Any, List
+from typing import Any
 
 from ocp_resources.resource import MissingRequiredArgumentError, NamespacedResource
 
@@ -11,9 +10,9 @@ class ApplicationAwareResourceQuota(NamespacedResource):
 
     def __init__(
         self,
-        hard: Dict[str, Any] | None = None,
-        scope_selector: Dict[str, Any] | None = None,
-        scopes: List[str] | None = None,
+        hard: dict[str, Any] | None = None,
+        scope_selector: dict[str, Any] | None = None,
+        scopes: list[str] | None = None,
         **kwargs: Any,
     ) -> None:
         """

ocp_resources/backup.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
-from __future__ import annotations
-from typing import Any, List
+from typing import Any
 
 from ocp_resources.resource import NamespacedResource
 
@@ -15,8 +14,8 @@ class Backup(NamespacedResource):
 
     def __init__(
         self,
-        included_namespaces: List[str] | None = None,
-        excluded_resources: List[str] | None = None,
+        included_namespaces: list[str] | None = None,
+        excluded_resources: list[str] | None = None,
         snapshot_move_data: bool = False,
         storage_location: str = "",
         **kwargs: Any,

ocp_resources/benchmark.py

@@ -1,9 +1,10 @@
-from typing import Any, Optional
+from typing import Any
 
-from ocp_resources.utils.constants import NOT_FOUND_ERROR_EXCEPTION_DICT, TIMEOUT_30SEC
-from ocp_resources.resource import NamespacedResource
 from timeout_sampler import TimeoutSampler
 
+from ocp_resources.resource import NamespacedResource
+from ocp_resources.utils.constants import NOT_FOUND_ERROR_EXCEPTION_DICT, TIMEOUT_30SEC
+
 
 class Benchmark(NamespacedResource):
     """
@@ -75,7 +76,7 @@ def workload_kind(self) -> str:
         """
         return self.workload_arg(arg="kind", default="pod")
 
-    def workload_arg(self, arg: str, default: Optional[Any] = None) -> Any:
+    def workload_arg(self, arg: str, default: Any = None) -> Any:
         """
         Retrieve the value of spec.workload.args[arg]
 

ocp_resources/cdi_config.py

@@ -1,12 +1,12 @@
 # Generated using https://github.com/RedHatQE/openshift-python-wrapper/blob/main/scripts/resource/README.md
 
-from __future__ import annotations
-from __future__ import annotations
 from typing import Any
-from ocp_resources.utils.constants import PROTOCOL_ERROR_EXCEPTION_DICT, TIMEOUT_4MINUTES
-from ocp_resources.resource import Resource
+
 from timeout_sampler import TimeoutSampler
 
+from ocp_resources.resource import Resource
+from ocp_resources.utils.constants import PROTOCOL_ERROR_EXCEPTION_DICT, TIMEOUT_4MINUTES
+
 
 class CDIConfig(Resource):
     """

ocp_resources/chaos_engine.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 
-from typing import Any, Dict, List
+from typing import Any
+
 from ocp_resources.resource import NamespacedResource
 
 
@@ -17,8 +18,8 @@ def engine_status(self) -> str:
         return self.instance.status["engineStatus"]
 
     @property
-    def experiments_status(self) -> Dict[str, Dict[str, Any]]:
-        experiments: List[Dict[Any, Any]] = self.instance.status["experiments"]
+    def experiments_status(self) -> dict[str, dict[str, Any]]:
+        experiments: list[dict[Any, Any]] = self.instance.status["experiments"]
         ret_value = {}
         for experiment in experiments:
             exp = {"verdict": experiment["verdict"], "status": experiment["status"]}

ocp_resources/cluster_deployment.py

@@ -1,4 +1,5 @@
-from typing import Any, Dict
+from typing import Any
+
 from ocp_resources.resource import NamespacedResource
 
 
@@ -13,8 +14,8 @@ def __init__(
         self,
         base_domain: str = "",
         cluster_name: str = "",
-        platform: Dict[str, Any] | None = None,
-        provisioning: Dict[str, Any] | None = None,
+        platform: dict[str, Any] | None = None,
+        provisioning: dict[str, Any] | None = None,
         pull_secret_ref_name: str = "",
         **kwargs: Any,
     ) -> None:

ocp_resources/cluster_pool.py

@@ -1,5 +1,5 @@
-from __future__ import annotations
-from typing import Any, Dict
+from typing import Any
+
 from ocp_resources.resource import NamespacedResource
 
 
@@ -14,7 +14,7 @@ def __init__(
         self,
         base_domain: str = "",
         image_set_ref_name: str = "",
-        platform: Dict[str, Any] | None = None,
+        platform: dict[str, Any] | None = None,
         pull_secret_ref_name: str = "",
         running_count: int | None = None,
         size: int | None = None,

ocp_resources/cluster_service_version.py

@@ -1,5 +1,5 @@
 import json
-from typing import List, Any, Dict
+from typing import Any
 
 from ocp_resources.resource import NamespacedResource
 
@@ -10,13 +10,13 @@ class ClusterServiceVersion(NamespacedResource):
     class Status(NamespacedResource.Status):
         INSTALLING = "Installing"
 
-    def get_alm_examples(self) -> List[Dict[str, Any]]:
+    def get_alm_examples(self) -> list[dict[str, Any]]:
         """
         Parse the alm-examples annotation from the CSV instance and return a list of dictionaries.
         Returns an empty list if no annotation is found or if the JSON is invalid.
 
         Returns:
-            Union[List[Dict[str, Any]], List[]]: A list of dictionaries from alm-examples, or an empty list if parsing fails.
+            Union[List[dict[str, Any]], List[]]: A list of dictionaries from alm-examples, or an empty list if parsing fails.
         """
         alm_examples = self.instance.metadata.annotations.get("alm-examples")