feat: add ResourceList and NamespacedResourceList classes (#2437)

Author: adolfo-ab

examples/resource_list.py

@@ -0,0 +1,44 @@
+"""
+openshift-python-wrapper allows the creation of several similar resources at the same time
+by using the ResourceList and NamespacedResourceList classes.
+When used as a contextmanager, deployment and deletion is handled automatically like in other classes.
+"""
+
+from ocp_resources.namespace import Namespace
+from ocp_resources.resource import get_client, ResourceList, NamespacedResourceList
+from ocp_resources.role import Role
+
+client = get_client()
+
+# We create a list of three namespaces: ns-1, ns-2, ns-3
+namespaces = ResourceList(resource_class=Namespace, num_resources=3, client=client, name="ns")
+namespaces.deploy()
+
+assert namespaces[2].name == "ns-3"
+
+# Now we create one role on each namespace
+roles = NamespacedResourceList(
+    client=client,
+    resource_class=Role,
+    name="role",
+    namespaces=[ns.name for ns in namespaces],
+    rules=[
+        {
+            "apiGroups": ["serving.kserve.io"],
+            "resources": ["inferenceservices"],
+            "verbs": ["get", "list", "watch"],
+        }
+    ],
+)
+
+assert roles[2].namespace == "ns-3"
+
+# We clean up all the resources we created
+namespaces.clean_up()
+roles.clean_up()
+
+
+# We can also work with these classes using contextmanagers
+# for automatic clean up
+with ResourceList(client=client, resource_class=Namespace, name="ns", num_resources=3) as namespaces:
+    assert namespaces[0].name == "ns-1"

ocp_resources/resource.py

@@ -8,11 +8,12 @@
 import re
 import sys
 import threading
+from abc import abstractmethod, ABC
 from collections.abc import Callable, Generator
 from io import StringIO
 from signal import SIGINT, signal
 from types import TracebackType
-from typing import Any
+from typing import Any, Type
 from urllib.parse import parse_qs, urlencode, urlparse
 from warnings import warn
 
@@ -59,6 +60,7 @@
 from ocp_resources.utils.resource_constants import ResourceConstants
 from ocp_resources.utils.utils import skip_existing_resource_creation_teardown
 
+
 LOGGER = get_logger(name=__name__)
 MAX_SUPPORTED_API_VERSION = "v2"
 
@@ -736,7 +738,7 @@ def _sigint_handler(self, signal_received: int, frame: Any) -> None:
         self.__exit__()
         sys.exit(signal_received)
 
-    def deploy(self, wait: bool = False) -> Any:
+    def deploy(self, wait: bool = False) -> Resource | NamespacedResource:
         """
         For debug, export REUSE_IF_RESOURCE_EXISTS to skip resource create.
         Spaces are important in the export dict
@@ -1756,3 +1758,160 @@ def _apply_patches_sampler(self, patches: dict[Any, Any], action_text: str, acti
             timeout=TIMEOUT_30SEC,
             sleep_time=TIMEOUT_5SEC,
         )
+
+
+class BaseResourceList(ABC):
+    """
+    Abstract base class for managing collections of resources.
+
+    Provides common functionality for resource lists including context management,
+    iteration, indexing, deployment, and cleanup operations.
+    """
+
+    def __init__(self, client: DynamicClient):
+        self.resources: list[Resource] = []
+        self.client = client
+
+    def __enter__(self):
+        """Enters the runtime context and deploys all resources."""
+        self.deploy()
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> None:
+        """Exits the runtime context and cleans up all resources."""
+        self.clean_up()
+
+    def __iter__(self) -> Generator[Resource | NamespacedResource, None, None]:
+        """Allows iteration over the resources in the list."""
+        yield from self.resources
+
+    def __getitem__(self, index: int) -> Resource | NamespacedResource:
+        """Retrieves a resource from the list by its index."""
+        return self.resources[index]
+
+    def __len__(self) -> int:
+        """Returns the number of resources in the list."""
+        return len(self.resources)
+
+    def deploy(self, wait: bool = False) -> list[Resource | NamespacedResource]:
+        """
+        Deploys all resources in the list.
+
+        Args:
+            wait (bool): If True, wait for each resource to be ready.
+
+        Returns:
+            List[Any]: A list of the results from each resource's deploy() call.
+        """
+        return [resource.deploy(wait=wait) for resource in self.resources]
+
+    def clean_up(self, wait: bool = True) -> bool:
+        """
+        Deletes all resources in the list.
+
+        Args:
+            wait (bool): If True, wait for each resource to be deleted.
+
+        Returns:
+            bool: Returns True if all resources are cleaned up correclty.
+        """
+        # Deleting in reverse order to resolve dependencies correctly.
+        return all(resource.clean_up(wait=wait) for resource in reversed(self.resources))
+
+    @abstractmethod
+    def _create_resources(self, resource_class: Type, **kwargs: Any) -> None:
+        """Abstract method to create resources based on specific logic."""
+        pass
+
+
+class ResourceList(BaseResourceList):
+    """
+    A class to manage a collection of a specific resource type.
+
+    This class creates and manages N copies of a given resource,
+    each with a unique name derived from a base name.
+    """
+
+    def __init__(
+        self,
+        resource_class: Type[Resource],
+        num_resources: int,
+        client: DynamicClient,
+        **kwargs: Any,
+    ) -> None:
+        """
+        Initializes a list of N resource objects.
+
+        Args:
+            resource_class (Type[Resource]): The resource class to instantiate (e.g., Namespace).
+            num_resources (int): The number of resource copies to create.
+            client (DynamicClient): The dynamic client to use. Defaults to None.
+            **kwargs (Any): Arguments to be passed to the constructor of the resource_class.
+                              A 'name' key is required in kwargs to serve as the base name for the resources.
+        """
+        super().__init__(client)
+
+        self.num_resources = num_resources
+        self._create_resources(resource_class, **kwargs)
+
+    def _create_resources(self, resource_class: Type[Resource], **kwargs: Any) -> None:
+        """Creates N resources with indexed names."""
+        base_name = kwargs["name"]
+
+        for i in range(1, self.num_resources + 1):
+            resource_name = f"{base_name}-{i}"
+            resource_kwargs = kwargs.copy()
+            resource_kwargs["name"] = resource_name
+
+            instance = resource_class(client=self.client, **resource_kwargs)
+            self.resources.append(instance)
+
+
+class NamespacedResourceList(BaseResourceList):
+    """
+    Manages a collection of a specific namespaced resource (e.g., Pod, Service, etc), creating one instance per provided namespace.
+
+    This class creates one copy of a given namespaced resource in each of the
+    namespaces provided in a list.
+    """
+
+    def __init__(
+        self,
+        resource_class: Type[NamespacedResource],
+        namespaces: ResourceList,
+        client: DynamicClient,
+        **kwargs: Any,
+    ) -> None:
+        """
+        Initializes a list of resource objects, one for each specified namespace.
+
+        Args:
+            resource_class (Type[NamespacedResource]): The namespaced resource class to instantiate (e.g., Pod).
+            namespaces (ResourceList): A ResourceList containing namespaces where the resources will be created.
+            client (DynamicClient): The dynamic client to use for cluster communication.
+            **kwargs (Any): Additional arguments to be passed to the resource_class constructor.
+                              A 'name' key is required in kwargs to serve as the base name for the resources.
+        """
+        for ns in namespaces:
+            if ns.kind != "Namespace":
+                raise TypeError("All the resources in namespaces should be namespaces.")
+
+        super().__init__(client)
+
+        self.namespaces = namespaces
+        self._create_resources(resource_class, **kwargs)
+
+    def _create_resources(self, resource_class: Type[NamespacedResource], **kwargs: Any) -> None:
+        """Creates one resource per namespace."""
+        for ns in self.namespaces:
+            instance = resource_class(
+                namespace=ns.name,
+                client=self.client,
+                **kwargs,
+            )
+            self.resources.append(instance)

tests/test_resource.py

@@ -8,10 +8,14 @@
 from ocp_resources.exceptions import ResourceTeardownError
 from ocp_resources.namespace import Namespace
 from ocp_resources.pod import Pod
-from ocp_resources.resource import Resource
+from ocp_resources.resource import Resource, ResourceList, NamespacedResourceList
 from ocp_resources.secret import Secret
 from fake_kubernetes_client import FakeDynamicClient
 
+BASE_NAMESPACE_NAME: str = "test-namespace"
+BASE_POD_NAME: str = "test-pod"
+POD_CONTAINERS: list[dict[str, str]] = [{"name": "test-container", "image": "nginx:latest"}]
+
 
 class SecretTestExit(Secret):
     def deploy(self, wait: bool = False):
@@ -28,17 +32,22 @@ def client():
 
 @pytest.fixture(scope="class")
 def namespace(client):
-    return Namespace(client=client, name="test-namespace")
+    return Namespace(client=client, name=BASE_NAMESPACE_NAME)
+
+
+@pytest.fixture(scope="class")
+def namespaces(client):
+    return ResourceList(client=client, resource_class=Namespace, num_resources=3, name=BASE_NAMESPACE_NAME)
 
 
 @pytest.fixture(scope="class")
 def pod(client):
     # Create a test pod for testing purposes
     test_pod = Pod(
         client=client,
-        name="test-pod",
+        name=BASE_POD_NAME,
         namespace="default",
-        containers=[{"name": "test-container", "image": "nginx:latest"}],
+        containers=POD_CONTAINERS,
         annotations={"fake-client.io/ready": "false"},  # Create pod with Ready status FALSE
     )
     deployed_pod = test_pod.deploy()
@@ -47,6 +56,17 @@ def pod(client):
     test_pod.clean_up()
 
 
+@pytest.fixture(scope="class")
+def pods(client, namespaces):
+    return NamespacedResourceList(
+        client=client,
+        resource_class=Pod,
+        namespaces=namespaces,
+        name=BASE_POD_NAME,
+        containers=POD_CONTAINERS,
+    )
+
+
 @pytest.mark.incremental
 class TestResource:
     def test_get(self, client):
@@ -119,6 +139,60 @@ def test_resource_context_manager_exit(self, client):
                 pass
 
 
+@pytest.mark.incremental
+class TestResourceList:
+    def test_resource_list_deploy(self, namespaces):
+        namespaces.deploy()
+        assert namespaces
+
+    def test_resource_list_len(self, namespaces):
+        assert len(namespaces) == 3
+
+    def test_resource_list_name(self, namespaces):
+        for i, ns in enumerate(namespaces.resources, start=1):
+            assert ns.name == f"{BASE_NAMESPACE_NAME}-{i}"
+
+    def test_resource_list_teardown(self, namespaces):
+        namespaces.clean_up(wait=False)
+
+    def test_resource_list_context_manager(self, client):
+        with ResourceList(
+            client=client, resource_class=Namespace, name=BASE_NAMESPACE_NAME, num_resources=3
+        ) as namespaces:
+            assert namespaces
+
+
+@pytest.mark.incremental
+class TestNamespacedResourceList:
+    def test_namespaced_resource_list_deploy(self, client, pods):
+        pods.deploy()
+        assert pods
+
+    def test_resource_list_len(self, namespaces, pods):
+        assert len(pods) == len(namespaces)
+
+    def test_resource_list_name(self, pods):
+        for pod in pods.resources:
+            assert pod.name == BASE_POD_NAME
+
+    def test_namespaced_resource_list_namespace(self, namespaces, pods):
+        for pod, namespace in zip(pods.resources, namespaces):
+            assert pod.namespace == namespace.name
+
+    def test_resource_list_teardown(self, pods):
+        pods.clean_up(wait=False)
+
+    def test_namespaced_resource_list_context_manager(self, client, namespaces):
+        with NamespacedResourceList(
+            client=client,
+            resource_class=Pod,
+            namespaces=namespaces,
+            name=BASE_POD_NAME,
+            containers=POD_CONTAINERS,
+        ) as pods:
+            assert pods
+
+
 @pytest.mark.xfail(reason="Need debug")
 class TestClientProxy:
     @patch.dict(os.environ, {"HTTP_PROXY": "http://env-http-proxy.com"})