Skip to content
Snippets Groups Projects
Verified Commit 9e0336b7 authored by Arie Peterson's avatar Arie Peterson
Browse files

Remove unused pytest code for resource checking

parent 2d967da0
No related branches found
No related tags found
No related merge requests found
......@@ -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})!"
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment