testcases/basic_func_tests/tc_008_storage_check.py

   1 import sys\r
   2 import os\r
   3 from robot.libraries.BuiltIn import BuiltIn\r
   4 from robot.api import logger\r
   5 from decorators_for_robot_functionalities import *\r
   6 from test_constants import *\r
   7 \r
   8 sys.path.append(os.path.join(os.path.dirname(__file__), '..', '../libraries/common'))\r
   9 import common_utils  # noqa\r
  10 \r
  11 BuiltIn().import_library('pabot.PabotLib')\r
  12 pabot = BuiltIn().get_library_instance('pabot.PabotLib')\r
  13 \r
  14 execute = BuiltIn().get_library_instance('execute_command')\r
  15 stack_infos = BuiltIn().get_library_instance('stack_infos')\r
  16 pv_name = ""\r
  17 \r
  18 \r
  19 def tc_008_storage_check():\r
  20     steps = ['step1_read_write_pv',\r
  21              'step2_check_pv_retaining',\r
  22              'step3_read_write_pv']\r
  23     BuiltIn().run_keyword("tc_008_storage_check.Setup")\r
  24     common_utils.keyword_runner(steps)\r
  25 \r
  26 \r
  27 def Setup():\r
  28     pabot.acquire_lock("pv_test_ip")\r
  29     install_charts()\r
  30 \r
  31 \r
  32 def step1_read_write_pv():\r
  33     read_write_pv("step1.log")\r
  34 \r
  35 \r
  36 @pabot_lock("health_check_2")\r
  37 def step2_check_pv_retaining():\r
  38     common_utils.helm_delete("storage-test")\r
  39     common_utils.check_kubernetes_object(kube_object=pv_test_pod,\r
  40                                          tester_function=common_utils.test_kubernetes_object_not_available,\r
  41                                          timeout=90)\r
  42     _install_storage_test_helm_chart()\r
  43     pabot.release_lock("pv_test_ip")\r
  44 \r
  45 \r
  46 def step3_read_write_pv():\r
  47     read_write_pv("step3.log")\r
  48 \r
  49 \r
  50 def read_write_pv(file_name):\r
  51     pod_list = execute.execute_unix_command("kubectl get pod | grep pv-test-deployment | grep -i running | "\r
  52                                             "awk '{print $1}'")\r
  53 \r
  54     # write log on persistent storage from pods\r
  55     for pod in pod_list.split("\n"):\r
  56         pod = pod.strip()\r
  57         logger.info("POD NAME: " + pod)\r
  58         execute.execute_unix_command(\r
  59             "kubectl exec " + pod + " -- sh -c 'echo test_log_" + pod + " >> /usr/share/storage_test/" + file_name +\r
  60             "'")\r
  61 \r
  62     # check if logs can be reached from containers\r
  63     for pod in pod_list.split("\n"):\r
  64         pod = pod.strip()\r
  65         log = execute.execute_unix_command(\r
  66             "kubectl exec " + pod + " -- sh -c 'cat /usr/share/storage_test/" + file_name + "'")\r
  67         for pod_log in pod_list.split("\n"):\r
  68             pod_log = pod_log.strip()\r
  69             if pod_log not in log:\r
  70                 raise Exception("Log entry: test_log_" + pod_log + " is not found in log file")\r
  71 \r
  72 \r
  73 @pabot_lock("health_check_2")\r
  74 def install_charts():\r
  75     common_utils.helm_install(chart_name="default/persistentvolume-claim", release_name="pvc")\r
  76     common_utils.wait_if_pressure()\r
  77     common_utils.check_kubernetes_object(kube_object=pv_test_pvc,\r
  78                                          tester_function=common_utils.test_kubernetes_object_available,\r
  79                                          additional_filter="Bound", timeout=90)\r
  80     _install_storage_test_helm_chart()\r
  81 \r
  82     global pv_name  # pylint: disable=global-statement\r
  83     pv_name = execute.execute_unix_command("kubectl get pvc | grep pvc- | awk {'print$3'}")\r
  84 \r
  85 \r
  86 def _install_storage_test_helm_chart():\r
  87     if stack_infos.get_worker_nodes():\r
  88         common_utils.helm_install(chart_name="default/storage-test-worker", release_name="storage-test")\r
  89     else:\r
  90         common_utils.helm_install(chart_name="default/storage-test-oam", release_name="storage-test")\r
  91     common_utils.wait_if_pressure()\r
  92     common_utils.check_kubernetes_object(kube_object=pv_test_pod,\r
  93                                          tester_function=common_utils.test_kubernetes_object_available,\r
  94                                          additional_filter="Running", timeout=60)\r