Pv pvc

examples/test_deployment.py

@@ -2,8 +2,6 @@
 
 import os
 
-import time
-
 
 def test_deployment(kube):
 

kubetest/client.py

@@ -231,6 +231,49 @@ def load_service(self, path: str, set_namespace: bool = True) -> objects.Service
             service.namespace = self.namespace
         return service
 
+    def load_persistentvolume(self, path, set_namespace=False) -> objects.PersistentVolume:
+        """Load a manifest YAML into a PersistentVolume object.
+
+        By default, this will not augment the PersistentVolume object with
+        the generated test case namespace. This behavior can be
+        enabled with the ``set_namespace`` flag.
+
+        Args:
+            path (str): The path to the PersistentVolume manifest.
+            set_namespace (bool): Enable/disable the automatic
+                augmentation of the PersistentVolume namespace.
+
+        Returns:
+            objects.PersistentVolume: The PersistentVolume for the specified manifest.
+        """
+        log.info('loading persistentvolume from path: %s', path)
+        persistentvolume = objects.PersistentVolume.load(path)
+        if set_namespace:
+            persistentvolume.namespace = self.namespace
+        return persistentvolume
+
+    def load_persistentvolumeclaim(self, path, set_namespace=True) -> objects.PersistentVolumeClaim:
+        """Load a manifest YAML into a PersistentVolumeClaim object.
+
+        By default, this will augment the PersistentVolumeClaim object with
+        the generated test case namespace. This behavior can be
+        disabled with the ``set_namespace`` flag.
+
+        Args:
+            path (str): The path to the PersistentVolumeClaim manifest.
+            set_namespace (bool): Enable/disable the automatic
+                augmentation of the PersistentVolumeClaim namespace.
+
+        Returns:
+            objects.PersistentVolumeClaim: The PersistentVolumeClaim for the specified
+            manifest.
+        """
+        log.info('loading persistentvolumeclaim from path: %s', path)
+        persistentvolumeclaim = objects.PersistentVolumeClaim.load(path)
+        if set_namespace:
+            persistentvolumeclaim.namespace = self.namespace
+        return persistentvolumeclaim
+
     def load_statefulset(self, path: str, set_namespace: bool = True) -> objects.StatefulSet:
         """Load a manifest YAML into a StatefulSet object.
 
@@ -592,6 +635,71 @@ def get_secrets(
 
         return secrets
 
+    def get_persistentvolumes(self, fields=None, labels=None):
+        """Get PersistentVolumes from the cluster.
+
+        Args:
+            fields (dict[str, str]): A dictionary of fields used to restrict
+                the returned collection of PersistentVolume to only those which match
+                these field selectors. By default, no restricting is done.
+            labels (dict[str, str]): A dictionary of labels used to restrict
+                the returned collection of PersistentVolume to only those which match
+                these label selectors. By default, no restricting is done.
+
+        Returns:
+            dict[str, objects.PersistentVolume]: A dictionary where the key is
+            the PersistentVolume name and the value is the PersistentVolume itself.
+        """
+        selectors = utils.selector_kwargs(fields, labels)
+
+        persistentvolume_list = client.CoreV1Api().list_persistent_volume(
+                **selectors
+        )
+
+        persistentvolumes = {}
+        for obj in persistentvolume_list.items:
+            persistentvolume = objects.PersistentVolume(obj)
+            persistentvolumes[persistentvolume.name] = persistentvolume
+
+        return persistentvolumes
+
+    def get_persistentvolumeclaims(self, namespace=None, fields=None, labels=None):
+        """Get PersistentVolumeClaims from the cluster.
+
+        Args:
+            namespace (str): The namespace to get the PersistentVolumeClaim from. If not
+                specified, it will use the auto-generated test case namespace
+                by default.
+            fields (dict[str, str]): A dictionary of fields used to restrict
+                the returned collection of PersistentVolumeClaim to only those which match
+                these field selectors. By default, no restricting is done.
+            labels (dict[str, str]): A dictionary of labels used to restrict
+                the returned collection of PersistentVolumeClaim to only those which match
+                these label selectors. By default, no restricting is done.
+
+        Returns:
+            dict[str, objects.PersistentVolumeClaim]: A dictionary where the key is
+            the PersistentVolumeClaim name and the value is the PersistentVolumeClaim
+            itself.
+        """
+        if namespace is None:
+            namespace = self.namespace
+
+        selectors = utils.selector_kwargs(fields, labels)
+
+        persistentvolumeclaim_list = client.CoreV1Api().\
+            list_namespaced_persistent_volume_claim(
+            namespace=namespace,
+            **selectors,
+        )
+
+        persistentvolumeclaims = {}
+        for obj in persistentvolumeclaim_list.items:
+            persistentvolumeclaim = objects.PersistentVolumeClaim(obj)
+            persistentvolumeclaims[persistentvolumeclaim.name] = persistentvolumeclaim
+
+        return persistentvolumeclaims
+
     def get_services(
             self,
             namespace: str = None,

kubetest/manager.py

@@ -33,6 +33,8 @@ class ObjectManager:
         'secret',
         'service',
         'configmap',
+        'persistentvolume',
+        'persistentvolumeclaim',
         'daemonset',
         'statefulset',
         'deployment',
@@ -102,6 +104,8 @@ def get_objects_in_apply_order(self) -> Generator[objects.ApiObject, None, None]
           - Secret
           - Service
           - ConfigMap
+          - PersistentVolume
+          - PersistentVolumeClaim
           - DaemonSet
           - StatefulSet
           - Deployment
@@ -156,6 +160,7 @@ def __init__(
         self.namespace_create = namespace_create
         self.rolebindings = []
         self.clusterrolebindings = []
+        self.persistentvolumes = []
 
         self.test_objects = ObjectManager()
 
@@ -191,6 +196,10 @@ def setup(self) -> None:
         for crb in self.clusterrolebindings:
             self.client.create(crb)
 
+        # if there are any persistent volumes, create them.
+        for pv in self.persistentvolumes:
+            self.client.create(pv)
+
         # if any objects were registered with the test case via the
         # `applymanifests` marker, register them to the test client
         # and add them to the cluster now
@@ -222,6 +231,11 @@ def teardown(self) -> None:
         for crb in self.clusterrolebindings:
             self.client.delete(crb)
 
+        # PersistentVolume are not bound to a namespace, so we will need
+        # to delete them ourselves.
+        for pv in self.persistentvolumes:
+            self.client.delete(pv)
+
     def yield_container_logs(self, tail_lines: int = None) -> Generator[str, None, None]:
         """Yield the container logs for the test case.
 
@@ -295,6 +309,18 @@ def register_clusterrolebindings(
         """
         self.clusterrolebindings.extend(clusterrolebindings)
 
+    def register_persistentvolumes(
+        self,
+        *persistentvolumes: objects.PersistentVolume
+    ) -> None:
+        """Register a PersistentVolume requirement with the test case.
+
+        Args:
+            *persistentvolumes (PersistentVolume): The PersistentVolumes
+                that are needed for the test case.
+        """
+        self.persistentvolumes.extend(persistentvolumes)
+
     def register_objects(self, api_objects: List[objects.ApiObject]):
         """Register the provided objects with the test case.
 

kubetest/markers.py

@@ -8,7 +8,8 @@
 
 from kubetest import manager
 from kubetest.manifest import load_file, load_path
-from kubetest.objects import ApiObject, ClusterRoleBinding, RoleBinding
+from kubetest.objects import (ApiObject, ClusterRoleBinding, PersistentVolume,
+                              RoleBinding)
 
 APPLYMANIFEST_INI = (
     'applymanifest(path): '
@@ -63,6 +64,13 @@
     'see: https://kubernetes.io/docs/reference/access-authn-authz/rbac/'
 )
 
+PERSISTENTVOLUME_INI = (
+    'persistentvolume(name, subject_kind=None, subject_name=None): '
+    'create and use a Kubernetes PersistentVolume for the test case. The generated '
+    'persistent volume will be automatically created and removed for each marked '
+    'test. The name of the volume must be specified. Only existing PersistentVolume can be '
+    'used.'
+)
 
 NAMESPACE_INI = (
     'namespace(create=True, name=None): '
@@ -85,6 +93,7 @@ def register(config) -> None:
     config.addinivalue_line('markers', APPLYMANIFESTS_INI)
     config.addinivalue_line('markers', CLUSTERROLEBINDING_INI)
     config.addinivalue_line('markers', ROLEBINDING_INI)
+    config.addinivalue_line('markers', PERSISTENTVOLUME_INI)
     config.addinivalue_line('markers', NAMESPACE_INI)
 
 
@@ -260,6 +269,37 @@ def clusterrolebindings_from_marker(item: pytest.Item, namespace: str) -> List[C
     return clusterrolebindings
 
 
+def persistentvolumes_from_marker(item: pytest.Item) -> List[PersistentVolume]:
+    """Create PersistentVolume for the test case if the test case is marked
+    with the `pytest.mark.persistentvolume` marker.
+
+    Args:
+        item: The pytest test item.
+
+    Return:
+        The PersistentVolumes which were generated from the test case markers.
+    """
+
+    persistentvolumes = []
+    for mark in item.iter_markers(name='persistentvolume'):
+        name = mark.args[0]
+        subj_kind = mark.kwargs.get('subject_kind')
+        subj_name = mark.kwargs.get('subject_name')
+
+        persistentvolumes.append(PersistentVolume(client.V1PersistentVolume(
+            metadata=client.V1ObjectMeta(
+                name=f'kubetest:{item.name}',
+            ),
+            spec=client.V1PersistentVolumeSpec(
+                api_group='storage.k8s.io',
+                kind='PersistentVolume',
+                name=name,
+            )
+        )))
+
+    return persistentvolumes
+
+
 def get_custom_rbac_subject(namespace: str, kind: str, name: str) -> List[client.V1Subject]:
     """Create a custom RBAC subject for the given namespace.
 

kubetest/objects/__init__.py

@@ -12,6 +12,8 @@
 from .event import Event
 from .namespace import Namespace
 from .node import Node
+from .persistentvolume import PersistentVolume
+from .persistentvolumeclaim import PersistentVolumeClaim
 from .pod import Pod
 from .rolebinding import RoleBinding
 from .secret import Secret

kubetest/objects/persistentvolume.py

@@ -0,0 +1,94 @@
+"""Kubetest wrapper for the Kubernetes ``PersistentVolume`` API Object."""
+
+import logging
+
+from kubernetes import client
+
+from .api_object import ApiObject
+
+log = logging.getLogger('kubetest')
+
+
+class PersistentVolume(ApiObject):
+    """Kubetest wrapper around a Kubernetes `PersistentVolume`_ API Object.
+
+    The actual ``kubernetes.client.V1PersistentVolume`` instance that this
+    wraps can be accessed via the ``obj`` instance member.
+
+    This wrapper provides some convenient functionality around the
+    API Object and provides some state management for the `PersistentVolume`_.
+
+    .. _PersistentVolume:
+        https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.10/#persistentvolume-v1-core
+    """
+
+    obj_type = client.V1PersistentVolume
+
+    api_clients = {
+        'preferred': client.CoreV1Api,
+        'v1': client.CoreV1Api,
+    }
+
+    def __str__(self):
+        return str(self.obj)
+
+    def __repr__(self):
+        return self.__str__()
+
+    def create(self, namespace=None):
+        """Create the PersistentVolume under the given namespace.
+
+        Args:
+            namespace (str): This argument is ignored for PersistentVolumes.
+        """
+        log.info('creating persistentvolume "%s"', self.name)
+        log.debug('persistentvolume: %s', self.obj)
+
+        self.obj = self.api_client.create_persistent_volume(
+            body=self.obj,
+        )
+
+    def delete(self, options):
+        """Delete the PersistentVolume.
+
+        Args:
+             options (client.V1DeleteOptions): Options for PersistentVolume deletion.
+
+        Returns:
+            client.V1Status: The status of the delete operation.
+        """
+        if options is None:
+            options = client.V1DeleteOptions()
+
+        log.info('deleting persistentvolume "%s"', self.name)
+        log.debug('delete options: %s', options)
+        log.debug('persistentvolume: %s', self.obj)
+
+        return self.api_client.delete_persistent_volume(
+            name=self.name,
+            body=options,
+        )
+
+    def refresh(self):
+        """Refresh the underlying Kubernetes PersistentVolume resource."""
+        self.obj = self.api_client.read_persistent_volume(
+            name=self.name,
+        )
+
+    def is_ready(self):
+        """Check if the PersistentVolume is in the ready state.
+
+        PersistentVolumes have a "status" field to check. However, as this
+        field may change from "available" to "bound" quickly if another
+        object is using the PersistentVolume, we will measure their
+        readiness status by whether or not they exist on the cluster.
+
+        Returns:
+            bool: True if in the ready state; False otherwise.
+        """
+        try:
+            self.refresh()
+        except:  # noqa
+            return False
+        else:
+            return True