diff --git a/test/pytest/test_resources.py b/test/pytest/test_resources.py index 4246f8296724b2bd88b568c0f4ff1880f51f8e33..83d29ae2cd6e14947994196bfc353dba280f56c9 100644 --- a/test/pytest/test_resources.py +++ b/test/pytest/test_resources.py @@ -166,164 +166,3 @@ def test_kustomizations(resource, namespace): plural='kustomizations', name=resource, namespace=namespace) - - -@pytest.mark.resource -@pytest.mark.helmreleases -def test_helmreleases(resource, namespace): - """ - Test if a single or all HelmRelease(s) installed by weave flux are in - 'Ready' state. - - If you want to test a single helmrelease please provide both name and - namespace, otherwise all helmreleases from all namespaces are tested. - - :param name: The resource object's name - :type name: str - :param namespace: The resource object's namespace - :type namespace: str - """ - - print('\n') - custom_objects = client.CustomObjectsApi() - check_custom_objects( - api=custom_objects, - api_group='helm.toolkit.fluxcd.io', - api_version='v2beta1', - plural='helmreleases', - name=resource, - namespace=namespace) - - -@pytest.mark.resource -@pytest.mark.deployments -def test_deployments(resource, namespace): - """ - Test if a single or all deployment(s) are ready - - If you want to test a single deployment please provide both name and - namespace, otherwise all deployments from all namespaces are tested. - - :param name: The resource object's name - :type name: str - :param namespace: The resource object's namespace - :type namespace: str - """ - - failed = 0 - failed_resources = [] - api = client.AppsV1Api() - print('\n') - - if resource != 'all': - deployments = api.list_namespaced_deployment( - namespace=namespace, - label_selector=f'helm.toolkit.fluxcd.io/name={resource}') - else: - deployments = api.list_deployment_for_all_namespaces() - - assert deployments.items, 'No deployments found !!' - for deployment in deployments.items: - resource = deployment.metadata.name - namespace = deployment.metadata.namespace - deployment = api.read_namespaced_deployment_status(resource, namespace) - - for condition in deployment.status.conditions: - if condition.type == 'Available': - available = condition.status == 'True' - replicas = deployment.spec.replicas - ready_replicas = deployment.status.ready_replicas - # Check if all replicas from the spec are ready - print(f'Deployment "{resource}" in namespace "{namespace}": ' - f'{available} ({ready_replicas}/{replicas})') - - - if not available or replicas != ready_replicas: - failed += 1 - failed_resources.append(resource) - - assert failed == 0, f"Error: {failed} resources not ready ({failed_resources})!" - -@pytest.mark.resource -@pytest.mark.statefulsets -def test_statefulsets(resource, namespace): - """ - Test if a single or all statefulset(s) are ready - - If you want to test a single statefulset please provide both name and - namespace, otherwise all statefulsets from all namespaces are tested. - - :param name: The resource object's name - :type name: str - :param namespace: The resource object's namespace - :type namespace: str - """ - - failed = 0 - failed_resources = [] - api = client.AppsV1Api() - print('\n') - - if resource != 'all': - statefulsets = api.list_namespaced_stateful_set( - namespace=namespace, - label_selector=f'helm.toolkit.fluxcd.io/name={resource}') - else: - statefulsets = api.list_stateful_set_for_all_namespaces() - - assert statefulsets.items, 'No statefulsets found !!' - for statefulset in statefulsets.items: - resource = statefulset.metadata.name - namespace = statefulset.metadata.namespace - statefulset = api.read_namespaced_stateful_set_status(resource, namespace) - replicas = statefulset.spec.replicas - ready_replicas = statefulset.status.ready_replicas - print(f'statefulset "{resource}" in namespace "{namespace}": ' - f'{ready_replicas}/{replicas}') - if replicas != ready_replicas: - failed += 1 - failed_resources.append(resource) - - assert failed == 0, f"Error: {failed} resources not ready ({failed_resources})!" - -@pytest.mark.resource -@pytest.mark.daemonsets -def test_daemonsets(resource, namespace): - """ - Test if a single or all daemonset(s) are ready - - If you want to test a single daemonset please provide both name and - namespace, otherwise all daemonsets from all namespaces are tested. - - :param name: The resource object's name - :type name: str - :param namespace: The resource object's namespace - :type namespace: str - """ - - failed = 0 - failed_resources = [] - api = client.AppsV1Api() - print('\n') - - if resource != 'all': - daemonsets = api.list_namespaced_daemon_set( - namespace=namespace, - label_selector=f'helm.toolkit.fluxcd.io/name={resource}') - else: - daemonsets = api.list_daemon_set_for_all_namespaces() - - assert daemonsets.items, 'No daemonsets found !!' - for daemonset in daemonsets.items: - resource = daemonset.metadata.name - namespace = daemonset.metadata.namespace - daemonset = api.read_namespaced_daemon_set_status(resource, namespace) - desired_number_scheduled = daemonset.status.desired_number_scheduled - number_ready = daemonset.status.number_ready - print(f'daemonset "{resource}" in namespace "{namespace}": ' - f'{number_ready}/{desired_number_scheduled}') - if desired_number_scheduled != number_ready: - failed += 1 - failed_resources.append(resource) - - assert failed == 0, f"Error: {failed} resources not ready ({failed_resources})!"