116 lines
5.1 KiB
Python
116 lines
5.1 KiB
Python
import pytest
|
|
|
|
from httmock import urlmatch, HTTMock, response
|
|
|
|
from config_app.config_util.k8saccessor import KubernetesAccessorSingleton, _deployment_rollout_status_message
|
|
from config_app.config_util.k8sconfig import KubernetesConfig
|
|
|
|
|
|
@pytest.mark.parametrize('deployment_object, expected_status, expected_message', [
|
|
({'metadata': {'generation': 1},
|
|
'status': {'observedGeneration': 0, 'conditions': []},
|
|
'spec': {'replicas': 0}},
|
|
'progressing',
|
|
'Waiting for deployment spec to be updated...'),
|
|
({'metadata': {'generation': 0},
|
|
'status': {'observedGeneration': 0, 'conditions': [{'type': 'Progressing', 'reason': 'ProgressDeadlineExceeded'}]},
|
|
'spec': {'replicas': 0}},
|
|
'failed',
|
|
"Deployment my-deployment's rollout failed. Please try again later."),
|
|
({'metadata': {'generation': 0},
|
|
'status': {'observedGeneration': 0, 'conditions': []},
|
|
'spec': {'replicas': 0}},
|
|
'available',
|
|
'Deployment my-deployment updated (no replicas, so nothing to roll out)'),
|
|
({'metadata': {'generation': 0},
|
|
'status': {'observedGeneration': 0, 'conditions': [], 'replicas': 1},
|
|
'spec': {'replicas': 2}},
|
|
'progressing',
|
|
'Waiting for rollout to finish: 0 out of 2 new replicas have been updated...'),
|
|
({'metadata': {'generation': 0},
|
|
'status': {'observedGeneration': 0, 'conditions': [], 'replicas': 1, 'updatedReplicas': 1},
|
|
'spec': {'replicas': 2}},
|
|
'progressing',
|
|
'Waiting for rollout to finish: 1 out of 2 new replicas have been updated...'),
|
|
({'metadata': {'generation': 0},
|
|
'status': {'observedGeneration': 0, 'conditions': [], 'replicas': 2, 'updatedReplicas': 1},
|
|
'spec': {'replicas': 1}},
|
|
'progressing',
|
|
'Waiting for rollout to finish: 1 old replicas are pending termination...'),
|
|
({'metadata': {'generation': 0},
|
|
'status': {'observedGeneration': 0, 'conditions': [], 'replicas': 1, 'updatedReplicas': 2, 'availableReplicas': 0},
|
|
'spec': {'replicas': 0}},
|
|
'progressing',
|
|
'Waiting for rollout to finish: 0 of 2 updated replicas are available...'),
|
|
({'metadata': {'generation': 0},
|
|
'status': {'observedGeneration': 0, 'conditions': [], 'replicas': 1, 'updatedReplicas': 2, 'availableReplicas': 2},
|
|
'spec': {'replicas': 0}},
|
|
'available',
|
|
'Deployment my-deployment successfully rolled out.'),
|
|
])
|
|
def test_deployment_rollout_status_message(deployment_object, expected_status, expected_message):
|
|
deployment_status = _deployment_rollout_status_message(deployment_object, 'my-deployment')
|
|
assert deployment_status.status == expected_status
|
|
assert deployment_status.message == expected_message
|
|
|
|
|
|
@pytest.mark.parametrize('kube_config, expected_api, expected_query', [
|
|
({'api_host': 'www.customhost.com'},
|
|
'/apis/extensions/v1beta1/namespaces/quay-enterprise/deployments', 'labelSelector=quay-enterprise-component%3Dapp'),
|
|
|
|
({'api_host': 'www.customhost.com', 'qe_deployment_selector': 'custom-selector'},
|
|
'/apis/extensions/v1beta1/namespaces/quay-enterprise/deployments',
|
|
'labelSelector=quay-enterprise-component%3Dcustom-selector'),
|
|
|
|
({'api_host': 'www.customhost.com', 'qe_namespace': 'custom-namespace'},
|
|
'/apis/extensions/v1beta1/namespaces/custom-namespace/deployments', 'labelSelector=quay-enterprise-component%3Dapp'),
|
|
|
|
({'api_host': 'www.customhost.com', 'qe_namespace': 'custom-namespace', 'qe_deployment_selector': 'custom-selector'},
|
|
'/apis/extensions/v1beta1/namespaces/custom-namespace/deployments',
|
|
'labelSelector=quay-enterprise-component%3Dcustom-selector'),
|
|
])
|
|
def test_get_qe_deployments(kube_config, expected_api, expected_query):
|
|
config = KubernetesConfig(**kube_config)
|
|
url_hit = [False]
|
|
|
|
@urlmatch(netloc=r'www.customhost.com')
|
|
def handler(request, _):
|
|
assert request.path == expected_api
|
|
assert request.query == expected_query
|
|
url_hit[0] = True
|
|
return response(200, '{}')
|
|
|
|
with HTTMock(handler):
|
|
KubernetesAccessorSingleton._instance = None
|
|
assert KubernetesAccessorSingleton.get_instance(config).get_qe_deployments() is not None
|
|
|
|
assert url_hit[0]
|
|
|
|
|
|
@pytest.mark.parametrize('kube_config, deployment_names, expected_api_hits', [
|
|
({'api_host': 'www.customhost.com'}, [], []),
|
|
({'api_host': 'www.customhost.com'}, ['myDeployment'],
|
|
['/apis/extensions/v1beta1/namespaces/quay-enterprise/deployments/myDeployment']),
|
|
({'api_host': 'www.customhost.com', 'qe_namespace': 'custom-namespace'},
|
|
['myDeployment', 'otherDeployment'],
|
|
['/apis/extensions/v1beta1/namespaces/custom-namespace/deployments/myDeployment',
|
|
'/apis/extensions/v1beta1/namespaces/custom-namespace/deployments/otherDeployment']),
|
|
])
|
|
def test_cycle_qe_deployments(kube_config, deployment_names, expected_api_hits):
|
|
KubernetesAccessorSingleton._instance = None
|
|
|
|
config = KubernetesConfig(**kube_config)
|
|
url_hit = [False] * len(expected_api_hits)
|
|
i = [0]
|
|
|
|
@urlmatch(netloc=r'www.customhost.com', method='PATCH')
|
|
def handler(request, _):
|
|
assert request.path == expected_api_hits[i[0]]
|
|
url_hit[i[0]] = True
|
|
i[0] += 1
|
|
return response(200, '{}')
|
|
|
|
with HTTMock(handler):
|
|
KubernetesAccessorSingleton.get_instance(config).cycle_qe_deployments(deployment_names)
|
|
|
|
assert all(url_hit)
|