testcases/HPA_check/Custom_HPA_check.py

   1 import sys
   2 import os
   3 from robot.libraries.BuiltIn import BuiltIn
   4 sys.path.append(os.path.join(os.path.dirname(__file__), '..', '../libraries/common'))
   5 from decorators_for_robot_functionalities import *  # noqa
   6 import common_utils  # noqa
   7 from test_constants import *  # noqa
   8
   9
  10 execute = BuiltIn().get_library_instance('execute_command')
  11 stack_infos = BuiltIn().get_library_instance('stack_infos')
  12
  13
  14 def Custom_HPA_check():
  15     steps = ['step1_check_initial_replica_count_custom',
  16              'step2_check_scale_out_custom',
  17              'step3_check_scale_in_custom']
  18     BuiltIn().run_keyword("Custom_HPA_check.setup")
  19     common_utils.keyword_runner(steps)
  20
  21
  22 GET_POD_REPLICA_COUNT = "kubectl get hpa --namespace=kube-system | grep podinfo | awk '{print $6}'"
  23
  24
  25 def setup():
  26     # flags = ["--horizontal-pod-autoscaler-downscale-stabilization=10s", "--horizontal-pod-autoscaler-sync-period=10s"]
  27     # common_utils.modify_static_pod_config(common_utils.add_flag_to_command, "cm.yml", flags)
  28     common_utils.helm_install(chart_name="default/custom-metrics", release_name="podinfo",
  29                               values="registry_url={reg_url}".format(reg_url=reg))
  30     common_utils.check_kubernetes_object(kube_object=podinfo_pod,
  31                                          tester_function=common_utils.test_kubernetes_object_available,
  32                                          additional_filter="Running",
  33                                          timeout=90)
  34
  35
  36 def step1_check_initial_replica_count_custom():
  37     expected_initial_replica_num = 2
  38     timeout = 1000
  39     check_scaling(expected_initial_replica_num, timeout)
  40
  41
  42 def step2_check_scale_out_custom():
  43     common_utils.helm_install(chart_name="default/http-traffic-gen", release_name="http-traffic-gen",
  44                               values="registry_url={reg_url}".format(reg_url=reg))
  45     common_utils.check_kubernetes_object(kube_object=http_traffic_gen,
  46                                          tester_function=common_utils.test_kubernetes_object_available,
  47                                          additional_filter="Running",
  48                                          timeout=45)
  49     expected_replicas = 3
  50     timeout = 1000
  51     check_scaling(expected_replicas, timeout)
  52
  53
  54 def step3_check_scale_in_custom():
  55     expected_replicas = 2
  56     timeout = 1000
  57     check_scaling(expected_replicas, timeout)
  58
  59
  60 @robot_log
  61 def check_scaling(expected_replicas, timeout=60):
  62     for _ in range(timeout):
  63         BuiltIn().sleep('1s')
  64         actual_replicas = int(execute.execute_unix_command(GET_POD_REPLICA_COUNT))
  65         if actual_replicas == expected_replicas:
  66             break
  67     BuiltIn().should_be_equal(actual_replicas, expected_replicas)