Rename test_app_deployments.py -> test_resources.py

test/pytest/test_certs.py

@@ -10,7 +10,7 @@ import pytest
 import requests
 from OpenSSL import SSL
 
-@pytest.mark.app
+@pytest.mark.resource
 @pytest.mark.certs
 def test_cert_validation(host, app): # pylint: disable=too-many-statements
     """Checks for proper cluster certs from exposed services.

test/pytest/test_app_deployments.py → test/pytest/test_resources.py

@@ -17,7 +17,7 @@ import pytest
 
 # Helper functions
 
-def get_resource_status(api, api_group, api_version, plural, name, namespace='flux-system'):
+def check_resource(api, api_group, api_version, plural, name, namespace='flux-system'):
     """Returns status contition for resource in given namespace"""
     print(f'Resource {name} in namespace {namespace} '
           f'api version: {api_group}/{api_version}): ', end='')
@@ -43,37 +43,13 @@ def get_resource_status(api, api_group, api_version, plural, name, namespace='fl
         print(resource)
     return status
 
-
-def check_all_pods_running(pods):
-    """
-    Loop through all the pods in an API result.
-    If a pod does not have an element `status.phase` with the value "Running",
-    return False. Otherwise, returns True.
-
-    :param kubernetes.V1Pod[] pods: list of V1Pod elements to check
-    """
-    ret = True
-    for pod in pods:
-        print("- {}: {}".format(pod.metadata.name, pod.status.phase))
-        if pod.status.phase != "Running":
-            ret = False
-    return ret
-
-
-@pytest.fixture(autouse=True)
-def run_around_tests():
-    """
-    Prepare kube config before running a test
+def check_custom_objects(api, api_group, api_version, plural, name, namespace='flux-system'):
     """
-    cluster_dir = os.environ.get("CLUSTER_DIR")
-    kubeconfig = os.path.join(cluster_dir, 'kube_config_cluster.yml')
-    config.load_kube_config(config_file=kubeconfig)
-    yield
+    Checks if all resoures of a given custom object are ready.
 
-
-def check_resources(api, api_group, api_version, plural, name, namespace='flux-system'):
-    """
-    Checks if all resoures of a given kind are ready.
+    We can't get a list of custom objects from all namespaces,
+    so we have to iterate over all namespaces. See
+    https://github.com/kubernetes-client/python/issues/1377
     """
 
     failed = 0
@@ -81,7 +57,7 @@ def check_resources(api, api_group, api_version, plural, name, namespace='flux-s
     print('\n')
 
     if name != 'all':
-        status = get_resource_status(
+        status = check_resource(
             api=api,
             api_group=api_group,
             api_version=api_version,
@@ -92,9 +68,6 @@ def check_resources(api, api_group, api_version, plural, name, namespace='flux-s
             failed += 1
             failed_resources.append(name)
     else:
-        # we can't get a list of custom objects from all namespaces,
-        # so we have to iterate over all namespaces. See
-        # https://github.com/kubernetes-client/python/issues/1377
         core_api = client.CoreV1Api()
         namespaces = core_api.list_namespace(watch=False)
         for namespace_object in namespaces.items:
@@ -108,7 +81,7 @@ def check_resources(api, api_group, api_version, plural, name, namespace='flux-s
             )
             for custom_object in custom_objects['items']:
                 resource_name = custom_object['metadata']['name']
-                status = get_resource_status(
+                status = check_resource(
                     api=api,
                     api_group=api_group,
                     api_version=api_version,
@@ -120,6 +93,76 @@ def check_resources(api, api_group, api_version, plural, name, namespace='flux-s
                     failed_resources.append(resource_name)
     assert failed == 0, f"Error: {failed} kustomizations not ready ({failed_resources})!"
 
+def check_objects(api, api_group, api_version, plural, name, namespace='flux-system'):
+    """
+    Checks if all resoures of a given object are ready.
+    """
+
+    failed = 0
+    failed_resources = []
+    print('\n')
+
+    if name != 'all':
+        status = check_resource(
+            api=api,
+            api_group=api_group,
+            api_version=api_version,
+            plural=plural,
+            name=name,
+            namespace=namespace)
+        if status != 'True':
+            failed += 1
+            failed_resources.append(name)
+    else:
+        objects = api.list_namespaced_custom_object(
+            group=api_group,
+            version=api_version,
+            plural=plural,
+            namespace=namespace_name,
+            watch=False
+        )
+        for custom_object in custom_objects['items']:
+            resource_name = custom_object['metadata']['name']
+            status = check_resource(
+                api=api,
+                api_group=api_group,
+                api_version=api_version,
+                plural=plural,
+                name=resource_name,
+                namespace=namespace_name)
+            if status != 'True':
+                failed += 1
+                failed_resources.append(resource_name)
+    assert failed == 0, f"Error: {failed} kustomizations not ready ({failed_resources})!"
+
+
+def check_all_pods_running(pods):
+    """
+    Loop through all the pods in an API result.
+    If a pod does not have an element `status.phase` with the value "Running",
+    return False. Otherwise, returns True.
+
+    :param kubernetes.V1Pod[] pods: list of V1Pod elements to check
+    """
+    ret = True
+    for pod in pods:
+        print("- {}: {}".format(pod.metadata.name, pod.status.phase))
+        if pod.status.phase != "Running":
+            ret = False
+    return ret
+
+
+@pytest.fixture(autouse=True)
+def run_around_tests():
+    """
+    Prepare kube config before running a test
+    """
+    cluster_dir = os.environ.get("CLUSTER_DIR")
+    kubeconfig = os.path.join(cluster_dir, 'kube_config_cluster.yml')
+    config.load_kube_config(config_file=kubeconfig)
+    yield
+
+
 # Pytest functions
 
 @pytest.mark.resource
@@ -131,7 +174,7 @@ def test_kustomizations(resource, namespace):
     """
 
     custom_objects = client.CustomObjectsApi()
-    check_resources(
+    check_custom_objects(
         api=custom_objects,
         api_group='kustomize.toolkit.fluxcd.io',
         api_version='v1beta1',
@@ -149,7 +192,7 @@ def test_helmreleases(resource, namespace):
     """
 
     custom_objects = client.CustomObjectsApi()
-    check_resources(
+    check_custom_objects(
         api=custom_objects,
         api_group='helm.toolkit.fluxcd.io',
         api_version='v2beta1',
@@ -167,7 +210,7 @@ def test_deployments(resource, namespace):
     """
 
     api = client.CoreV1Api()
-    check_resources(
+    check_custom_objects(
         api=api,
         api_group='',
         api_version='',