Plan: add typing (#2235)

* Plan: add typing

* __enter__ type is Any (subclass)

* move _generate_hook_spec outside on __init__

* move _generate_hook_spec outside on __init__

* remove pdb line

* _generate_hook_spec: return the dict without saving it first

* added from __future__ import annotations

.gitignore

@@ -124,3 +124,6 @@ local-cluster/_hco/
 node_modules/
 package.json
 package-lock.json
+
+# Dev
+.local_dev/

ocp_resources/plan.py

@@ -1,4 +1,5 @@
-from ocp_resources.constants import TIMEOUT_4MINUTES
+from __future__ import annotations
+from typing import Any
 from ocp_resources.mtv import MTV
 from ocp_resources.resource import NamespacedResource
 
@@ -26,38 +27,24 @@ class Plan(NamespacedResource, MTV):
 
     def __init__(
         self,
-        name=None,
-        namespace=None,
-        source_provider_name=None,
-        source_provider_namespace=None,
-        destination_provider_name=None,
-        destination_provider_namespace=None,
-        storage_map_name=None,
-        storage_map_namespace=None,
-        network_map_name=None,
-        network_map_namespace=None,
-        virtual_machines_list=None,
-        target_namespace=None,
-        warm_migration=False,
-        pre_hook_name=None,
-        pre_hook_namespace=None,
-        after_hook_name=None,
-        after_hook_namespace=None,
-        client=None,
-        teardown=True,
-        yaml_file=None,
-        delete_timeout=TIMEOUT_4MINUTES,
+        source_provider_name: str | None = None,
+        source_provider_namespace: str | None = None,
+        destination_provider_name: str | None = None,
+        destination_provider_namespace: str | None = None,
+        storage_map_name: str | None = None,
+        storage_map_namespace: str | None = None,
+        network_map_name: str | None = None,
+        network_map_namespace: str | None = None,
+        virtual_machines_list: list[Any] | None = None,
+        target_namespace: str | None = None,
+        warm_migration: bool = False,
+        pre_hook_name: str | None = None,
+        pre_hook_namespace: str | None = None,
+        after_hook_name: str | None = None,
+        after_hook_namespace: str | None = None,
         **kwargs,
-    ):
-        super().__init__(
-            name=name,
-            namespace=namespace,
-            client=client,
-            teardown=teardown,
-            yaml_file=yaml_file,
-            delete_timeout=delete_timeout,
-            **kwargs,
-        )
+    ) -> None:
+        super().__init__(**kwargs)
         self.source_provider_name = source_provider_name
         self.source_provider_namespace = source_provider_namespace
         self.destination_provider_name = destination_provider_name
@@ -75,39 +62,29 @@ def __init__(
         self.target_namespace = target_namespace or self.namespace
         self.condition_message_ready = self.ConditionMessage.PLAN_READY
         self.condition_message_succeeded = self.ConditionMessage.PLAN_SUCCEEDED
+        self.hooks_array = []
 
-        def generate_hook_spec(hook_name, hook_namespace, hook_type):
-            hook = {
-                "hook": {
-                    "name": hook_name,
-                    "namespace": hook_namespace,
-                },
-                "step": hook_type,
-            }
-            return hook
-
-        hooks_array = []
         if self.pre_hook_name and self.pre_hook_namespace:
-            hooks_array.append(
-                generate_hook_spec(
+            self.hooks_array.append(
+                self._generate_hook_spec(
                     hook_name=self.pre_hook_name,
                     hook_namespace=self.pre_hook_namespace,
                     hook_type="PreHook",
                 )
             )
 
         if self.after_hook_name and self.after_hook_namespace:
-            hooks_array.append(
-                generate_hook_spec(
+            self.hooks_array.append(
+                self._generate_hook_spec(
                     hook_name=self.after_hook_name,
                     hook_namespace=self.after_hook_namespace,
                     hook_type="AfterHook",
                 )
             )
 
-        if hooks_array:
+        if self.hooks_array and self.virtual_machines_list:
             for vm in self.virtual_machines_list:
-                vm["hooks"] = hooks_array
+                vm["hooks"] = self.hooks_array
 
     def to_dict(self) -> None:
         super().to_dict()
@@ -139,3 +116,12 @@ def to_dict(self) -> None:
                     },
                 }
             })
+
+    def _generate_hook_spec(self, hook_name: str, hook_namespace: str, hook_type: str) -> dict[str, Any]:
+        return {
+            "hook": {
+                "name": hook_name,
+                "namespace": hook_namespace,
+            },
+            "step": hook_type,
+        }

ocp_resources/resource.py

@@ -1,4 +1,5 @@
 from __future__ import annotations
+
 from collections.abc import Callable, Generator
 import contextlib
 
@@ -47,6 +48,7 @@
 from ocp_resources.exceptions import MissingRequiredArgumentError, MissingResourceResError
 from ocp_resources.utils import skip_existing_resource_creation_teardown
 
+
 LOGGER = get_logger(name=__name__)
 MAX_SUPPORTED_API_VERSION = "v2"
 
@@ -612,7 +614,7 @@ def to_dict(self) -> None:
         """
         self._base_body()
 
-    def __enter__(self) -> "Resource":
+    def __enter__(self) -> Any:
         signal(SIGINT, self._sigint_handler)
         return self.deploy(wait=self.wait_for_resource)
 
@@ -629,7 +631,7 @@ def _sigint_handler(self, signal_received: int, frame: Any) -> None:
         self.__exit__()
         sys.exit(signal_received)
 
-    def deploy(self, wait: bool = False) -> "Resource":
+    def deploy(self, wait: bool = False) -> Any:
         """
         For debug, export REUSE_IF_RESOURCE_EXISTS to skip resource create.
         Spaces are important in the export dict