Pre commit use ruff 4.14

.flake8

@@ -1,6 +1,5 @@
 [flake8]
-
-max-line-length = 120
+select=FCN,M511
 
 exclude =
     doc,
@@ -11,12 +10,10 @@ exclude =
     docs/*,
     .cache/*
 
-fcfn_exclude_functions =
+fcn_exclude_functions =
     update,
-    getLogger,
     loads,
     get,
-    appens,
     next,
     setdefault,
     dumps,
@@ -44,6 +41,9 @@ fcfn_exclude_functions =
     re,
     requests,
     signal,
+    yaml,
+    benedict,
+    logger,
 
 enable-extensions =
     FCN,

.pre-commit-config.yaml

@@ -2,33 +2,8 @@ default_language_version:
     python: python3
 
 repos:
-  - repo: https://github.com/PyCQA/autoflake
-    rev: "v2.2.0"
-    hooks:
-      - id: autoflake
-        args:
-          - --in-place
-          - --remove-unused-variables
-          - --remove-all-unused-imports
-
-  - repo: https://github.com/hadialqattan/pycln
-    rev: "v2.1.5"
-    hooks:
-        - id: pycln
-
-  - repo: https://github.com/PyCQA/isort
-    rev: "5.12.0"
-    hooks:
-      - id: isort
-
-  - repo: https://github.com/psf/black
-    rev: "23.7.0"
-    hooks:
-    - id: black
-      args: [--preview]
-
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: "v4.4.0"
+    rev: "v4.5.0"
     hooks:
     - id: check-merge-conflict
     - id: debug-statements
@@ -37,21 +12,26 @@ repos:
     - id: end-of-file-fixer
     - id: check-ast
     - id: check-builtin-literals
+    - id: check-docstring-first
+    - id: check-toml
 
   - repo: https://github.com/PyCQA/flake8
-    rev: "6.0.0"
+    rev: "6.1.0"
     hooks:
       - id: flake8
-        args: [--max-line-length=120]
+        args: [--config=.flake8]
         additional_dependencies: [
-          "git+https://github.com/RedHatQE/flake8-plugins.git",
-          "pep8-naming",
-          "flake8-comprehensions",
+          "git+https://github.com/RedHatQE/flake8-plugins.git@v0.0.2",
           "flake8-mutable",
-          "flake8-simplify",
         ]
 
   - repo: https://github.com/Yelp/detect-secrets
     rev: v1.4.0
     hooks:
       - id: detect-secrets
+
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    rev: v0.1.5
+    hooks:
+      - id: ruff
+      - id: ruff-format

examples/special_cases.py

@@ -2,6 +2,7 @@
 Some resources have the same `kind` but different API groups.
 For example: `Network` which exists in both operator.openshift.io and config.openshift.io API groups
 """
+
 from ocp_resources.network import Network
 
 # To get the Network resource which uses the default API in the class ("config.openshift.io")

ocp_resources/catalog_source.py

@@ -31,16 +31,12 @@ def __init__(
         self.image = image
         self.display_name = display_name
         self.publisher = publisher
-        self.update_strategy_registry_poll_interval = (
-            update_strategy_registry_poll_interval
-        )
+        self.update_strategy_registry_poll_interval = update_strategy_registry_poll_interval
 
     def to_dict(self):
         super().to_dict()
         if not self.yaml_file:
-            if not all(
-                [self.source_type, self.image, self.display_name, self.publisher]
-            ):
+            if not all([self.source_type, self.image, self.display_name, self.publisher]):
                 raise ValueError(
                     "Passing yaml_file or all parameters 'source_type', 'image',"
                     " 'display_name' and 'publisher' is required."

ocp_resources/catalog_source_config.py

@@ -80,7 +80,5 @@ def wait_for_csc_status(self, status, timeout=120):
 
         except TimeoutExpiredError:
             if current_status:
-                self.logger.error(
-                    f"Status of {self.kind} {self.name} is {current_status}"
-                )
+                self.logger.error(f"Status of {self.kind} {self.name} is {current_status}")
             raise

ocp_resources/cdi_config.py

@@ -34,9 +34,7 @@ def wait_until_upload_url_changed(self, uploadproxy_url, timeout=TIMEOUT_4MINUTE
         Returns:
             bool: True if url is equal to uploadProxyURL.
         """
-        self.logger.info(
-            f"Wait for {self.kind} {self.name} to ensure current URL == uploadProxyURL"
-        )
+        self.logger.info(f"Wait for {self.kind} {self.name} to ensure current URL == uploadProxyURL")
         samples = TimeoutSampler(
             wait_timeout=timeout,
             sleep=1,

ocp_resources/cluster_role_binding.py

@@ -30,9 +30,7 @@ def to_dict(self):
         super().to_dict()
         if not self.yaml_file:
             if not self.cluster_role:
-                raise ValueError(
-                    "Passing yaml_file or parameter 'cluster_role' is required."
-                )
+                raise ValueError("Passing yaml_file or parameter 'cluster_role' is required.")
             self.res.setdefault("roleRef", {})
             self.res["roleRef"] = {
                 "apiGroup": self.api_group,

ocp_resources/controller_revision.py

@@ -38,8 +38,6 @@ def to_dict(self):
         super().to_dict()
         if not self.yaml_file:
             if self.owner_references:
-                self.res.setdefault("metadata", {}).update(
-                    {"ownerReference": self.owner_references}
-                )
+                self.res.setdefault("metadata", {}).update({"ownerReference": self.owner_references})
             if self.revision_object:
                 self.res.update({"data": self.revision_object.res})

ocp_resources/cron_job.py

@@ -48,10 +48,7 @@ def to_dict(self):
         super().to_dict()
         if not self.yaml_file:
             if not (self.job_template and self.schedule):
-                raise ValueError(
-                    "yaml_file or parameters 'job_template' and 'schedule' are"
-                    " required."
-                )
+                raise ValueError("yaml_file or parameters 'job_template' and 'schedule' are" " required.")
             self.res.update(
                 {
                     "spec": {
@@ -65,12 +62,8 @@ def to_dict(self):
             if self.suspend:
                 self.res["spec"]["suspend"] = self.suspend
             if self.successful_jobs_history_limit:
-                self.res["spec"][
-                    "successfulJobsHistoryLimit"
-                ] = self.successful_jobs_history_limit
+                self.res["spec"]["successfulJobsHistoryLimit"] = self.successful_jobs_history_limit
             if self.failed_jobs_history_limit:
-                self.res["spec"][
-                    "failedJobsHistoryLimit"
-                ] = self.failed_jobs_history_limit
+                self.res["spec"]["failedJobsHistoryLimit"] = self.failed_jobs_history_limit
             if self.concurrency_policy:
                 self.res["spec"]["concurrencyPolicy"] = self.concurrency_policy

ocp_resources/csi_storage_capacity.py

@@ -61,9 +61,7 @@ def to_dict(self):
         super().to_dict()
         if not self.yaml_file:
             if not self.storage_class_name:
-                raise ValueError(
-                    "yaml_file or parameter 'storage_class_name' is required."
-                )
+                raise ValueError("yaml_file or parameter 'storage_class_name' is required.")
             self.res.update(
                 {
                     "storageClassName": self.storage_class_name,

ocp_resources/daemonset.py

@@ -36,10 +36,7 @@ def wait_until_deployed(self, timeout=TIMEOUT_4MINUTES):
                 status = sample.items[0].status
                 desired_number_scheduled = status.desiredNumberScheduled
                 number_ready = status.numberReady
-                if (
-                    desired_number_scheduled > 0
-                    and desired_number_scheduled == number_ready
-                ):
+                if desired_number_scheduled > 0 and desired_number_scheduled == number_ready:
                     return
 
     def delete(self, wait=False, timeout=TIMEOUT_4MINUTES, body=None):

ocp_resources/data_import_cron.py

@@ -64,12 +64,8 @@ def to_dict(self):
                     "spec": {
                         "template": {
                             "spec": {
-                                "source": {
-                                    "registry": {"pullMethod": self.pull_method}
-                                },
-                                "storage": {
-                                    "resources": {"requests": {"storage": self.size}}
-                                },
+                                "source": {"registry": {"pullMethod": self.pull_method}},
+                                "storage": {"resources": {"requests": {"storage": self.size}}},
                             }
                         }
                     }
@@ -78,11 +74,7 @@ def to_dict(self):
             spec = self.res["spec"]["template"]["spec"]
             if self.bind_immediate_annotation:
                 self.res["metadata"].setdefault("annotations", {}).update(
-                    {
-                        f"{NamespacedResource.ApiGroup.CDI_KUBEVIRT_IO}/storage.bind.immediate.requested": (
-                            "true"
-                        )
-                    }
+                    {f"{NamespacedResource.ApiGroup.CDI_KUBEVIRT_IO}/storage.bind.immediate.requested": ("true")}
                 )
             if self.image_stream:
                 spec["source"]["registry"]["imageStream"] = self.image_stream

ocp_resources/datavolume.py

@@ -179,26 +179,16 @@ def to_dict(self):
             if self.source == "http" or "registry":
                 self.res["spec"]["source"][self.source]["url"] = self.url
             if self.cert_configmap:
-                self.res["spec"]["source"][self.source][
-                    "certConfigMap"
-                ] = self.cert_configmap
+                self.res["spec"]["source"][self.source]["certConfigMap"] = self.cert_configmap
             if self.source == "upload" or self.source == "blank":
                 self.res["spec"]["source"][self.source] = {}
             if self.hostpath_node:
                 self.res["metadata"].setdefault("annotations", {}).update(
-                    {
-                        f"{NamespacedResource.ApiGroup.KUBEVIRT_IO}/provisionOnNode": (
-                            self.hostpath_node
-                        )
-                    }
+                    {f"{NamespacedResource.ApiGroup.KUBEVIRT_IO}/provisionOnNode": (self.hostpath_node)}
                 )
             if self.multus_annotation:
                 self.res["metadata"].setdefault("annotations", {}).update(
-                    {
-                        f"{NamespacedResource.ApiGroup.K8S_V1_CNI_CNCF_IO}/networks": (
-                            self.multus_annotation
-                        )
-                    }
+                    {f"{NamespacedResource.ApiGroup.K8S_V1_CNI_CNCF_IO}/networks": (self.multus_annotation)}
                 )
             if self.bind_immediate_annotation:
                 self.res["metadata"].setdefault("annotations", {}).update(
@@ -213,11 +203,7 @@ def to_dict(self):
                 self.res["spec"]["preallocation"] = self.preallocation
             if self.delete_after_completion:
                 self.res["metadata"].setdefault("annotations", {}).update(
-                    {
-                        f"{self.api_group}/storage.deleteAfterCompletion": (
-                            self.delete_after_completion
-                        )
-                    }
+                    {f"{self.api_group}/storage.deleteAfterCompletion": (self.delete_after_completion)}
                 )
 
     def wait_deleted(self, timeout=TIMEOUT_4MINUTES):
@@ -238,9 +224,7 @@ def wait(self, timeout=TIMEOUT_10MINUTES, failure_timeout=TIMEOUT_2MINUTES):
 
         # If DV's status is not Pending, continue with the flow
         self.wait_for_status(status=self.Status.SUCCEEDED, timeout=timeout)
-        self.pvc.wait_for_status(
-            status=PersistentVolumeClaim.Status.BOUND, timeout=timeout
-        )
+        self.pvc.wait_for_status(status=PersistentVolumeClaim.Status.BOUND, timeout=timeout)
 
     @property
     def pvc(self):
@@ -252,9 +236,7 @@ def pvc(self):
 
     @property
     def scratch_pvc(self):
-        scratch_pvc_prefix = (
-            self.pvc.prime_pvc.name if self.pvc.use_populator else self.name
-        )
+        scratch_pvc_prefix = self.pvc.prime_pvc.name if self.pvc.use_populator else self.name
         return PersistentVolumeClaim(
             name=f"{scratch_pvc_prefix}-scratch",
             namespace=self.namespace,
@@ -318,24 +300,17 @@ def test_dv():
         self._check_none_pending_status(failure_timeout=failure_timeout)
 
         sample = None
-        status_of_dv_str = (
-            f"Status of {self.kind} '{self.name}' in namespace '{self.namespace}':\n"
-        )
+        status_of_dv_str = f"Status of {self.kind} '{self.name}' in namespace '{self.namespace}':\n"
         try:
             for sample in TimeoutSampler(
                 sleep=1,
                 wait_timeout=timeout,
                 func=lambda: self.exists,
             ):
                 # DV reach to success if the status is succeeded or if the DV does not exist
-                if (
-                    sample is None
-                    or sample.get("status", {}).get("phase") == self.Status.SUCCEEDED
-                ):
+                if sample is None or sample.get("status", {}).get("phase") == self.Status.SUCCEEDED:
                     break
-                elif stop_status_func and stop_status_func(
-                    *stop_status_func_args, **stop_status_func_kwargs
-                ):
+                elif stop_status_func and stop_status_func(*stop_status_func_args, **stop_status_func_kwargs):
                     raise TimeoutExpiredError(
                         value=(
                             "Exited on the stop_status_func"
@@ -349,9 +324,7 @@ def test_dv():
             raise
 
         # For CSI storage, PVC gets Bound after DV succeeded
-        return self.pvc.wait_for_status(
-            status=PersistentVolumeClaim.Status.BOUND, timeout=TIMEOUT_1MINUTE
-        )
+        return self.pvc.wait_for_status(status=PersistentVolumeClaim.Status.BOUND, timeout=TIMEOUT_1MINUTE)
 
     def delete(self, wait=False, timeout=TIMEOUT_4MINUTES, body=None):
         """

ocp_resources/deployment.py

@@ -67,9 +67,6 @@ def wait_for_replicas(self, deployed=True, timeout=TIMEOUT_4MINUTES):
 
                 if (
                     (deployed and spec_replicas)
-                    and spec_replicas
-                    == updated_replicas
-                    == available_replicas
-                    == ready_replicas
+                    and spec_replicas == updated_replicas == available_replicas == ready_replicas
                 ) or not (deployed or spec_replicas or total_replicas):
                     return

ocp_resources/endpoint_slice.py

@@ -55,10 +55,7 @@ def to_dict(self):
         super().to_dict()
         if not self.yaml_file:
             if not (self.address_type and self.endpoints):
-                raise ValueError(
-                    "yaml_file or parameters 'address_type' and 'endpoints' are"
-                    " required."
-                )
+                raise ValueError("yaml_file or parameters 'address_type' and 'endpoints' are" " required.")
             self.res.update(
                 {
                     "addressTypes": self.address_type,

ocp_resources/endpoints.py

@@ -52,9 +52,7 @@ def to_dict(self):
         super().to_dict()
         if not self.yaml_file:
             if not (self.addresses and self.ports):
-                raise ValueError(
-                    "yaml_file or parameters 'addresses' and 'ports' are required."
-                )
+                raise ValueError("yaml_file or parameters 'addresses' and 'ports' are required.")
             self.res.update(
                 {
                     "subsets": {

ocp_resources/event.py

@@ -54,9 +54,7 @@ def get(
             f" resource_version={resource_version}, timeout={timeout}"
         )
 
-        event_listener = dyn_client.resources.get(
-            api_version=cls.api_version, kind=cls.__name__
-        )
+        event_listener = dyn_client.resources.get(api_version=cls.api_version, kind=cls.__name__)
         yield from event_listener.watch(
             namespace=namespace,
             name=name,