quay/config_app/config_util/test/test_k8saccessor.py

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)