functest_kubernetes/k8stest.py

   1 #!/usr/bin/env python
   2 #
   3 # Copyright (c) 2018 All rights reserved
   4 # This program and the accompanying materials
   5 # are made available under the terms of the Apache License, Version 2.0
   6 # which accompanies this distribution, and is available at
   7 #
   8 # http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10
  11 """
  12 Define the parent for Kubernetes testing.
  13 """
  14
  15 from __future__ import division
  16
  17 import logging
  18 import os
  19 import re
  20 import subprocess
  21 import time
  22
  23 from xtesting.core import testcase
  24
  25
  26 class K8sTesting(testcase.TestCase):
  27     """Kubernetes test runner"""
  28
  29     __logger = logging.getLogger(__name__)
  30
  31     def __init__(self, **kwargs):
  32         super(K8sTesting, self).__init__(**kwargs)
  33         self.config = '/root/.kube/config'
  34         self.cmd = []
  35         self.result = 0
  36         self.start_time = 0
  37         self.stop_time = 0
  38
  39     def run_kubetest(self):  # pylint: disable=too-many-branches
  40         """Run the test suites"""
  41         cmd_line = self.cmd
  42         self.__logger.info("Starting k8s test: '%s'.", cmd_line)
  43
  44         process = subprocess.Popen(cmd_line, stdout=subprocess.PIPE,
  45                                    stderr=subprocess.STDOUT)
  46         output = process.stdout.read()
  47         # Remove color code escape sequences
  48         output = re.sub(r'\x1B\[[0-?]*[ -/]*[@-~]', '', str(output))
  49
  50         if ('Error loading client' in output or
  51                 'Unexpected error' in output):
  52             raise Exception(output)
  53
  54         remarks = []
  55         lines = output.split('\n')
  56         success = False
  57         failure = False
  58         i = 0
  59         while i < len(lines):
  60             if '[Fail]' in lines[i] or 'Failures:' in lines[i]:
  61                 self.__logger.error(lines[i])
  62             if re.search(r'\[(.)*[0-9]+ seconds\]', lines[i]):
  63                 self.__logger.debug(lines[i])
  64                 i = i + 1
  65                 while i < len(lines) and lines[i] != '-' * len(lines[i]):
  66                     if lines[i].startswith('STEP:') or ('INFO:' in lines[i]):
  67                         break
  68                     self.__logger.debug(lines[i])
  69                     i = i + 1
  70             if i >= len(lines):
  71                 break
  72             success = 'SUCCESS!' in lines[i]
  73             failure = 'FAIL!' in lines[i]
  74             if success or failure:
  75                 if i != 0 and 'seconds' in lines[i - 1]:
  76                     remarks.append(lines[i - 1])
  77                 remarks = remarks + lines[i].replace('--', '|').split('|')
  78                 break
  79             i = i + 1
  80
  81         self.__logger.debug('-' * 10)
  82         self.__logger.info("Remarks:")
  83         for remark in remarks:
  84             if 'seconds' in remark:
  85                 self.__logger.debug(remark)
  86             elif 'Passed' in remark:
  87                 self.__logger.info("Passed: %s", remark.split()[0])
  88             elif 'Skipped' in remark:
  89                 self.__logger.info("Skipped: %s", remark.split()[0])
  90             elif 'Failed' in remark:
  91                 self.__logger.info("Failed: %s", remark.split()[0])
  92
  93         if success:
  94             self.result = 100
  95         elif failure:
  96             self.result = 0
  97
  98     def run(self, **kwargs):
  99
 100         if not os.path.isfile(self.config):
 101             self.__logger.error(
 102                 "Cannot run k8s testcases. Config file not found")
 103             return self.EX_RUN_ERROR
 104
 105         self.start_time = time.time()
 106         try:
 107             self.run_kubetest()
 108             res = self.EX_OK
 109         except Exception:  # pylint: disable=broad-except
 110             self.__logger.exception("Error with running kubetest:")
 111             res = self.EX_RUN_ERROR
 112
 113         self.stop_time = time.time()
 114         return res
 115
 116     def check_envs(self):  # pylint: disable=no-self-use
 117         """Check if required environment variables are set"""
 118         try:
 119             assert 'DEPLOY_SCENARIO' in os.environ
 120             assert 'KUBE_MASTER_IP' in os.environ
 121             assert 'KUBERNETES_PROVIDER' in os.environ
 122             assert 'KUBE_MASTER_URL' in os.environ
 123         except Exception as ex:
 124             raise Exception("Cannot run k8s testcases. "
 125                             "Please check env var: %s" % str(ex))
 126
 127
 128 class K8sSmokeTest(K8sTesting):
 129     """Kubernetes smoke test suite"""
 130     def __init__(self, **kwargs):
 131         if "case_name" not in kwargs:
 132             kwargs.get("case_name", 'k8s_smoke')
 133         super(K8sSmokeTest, self).__init__(**kwargs)
 134         self.check_envs()
 135         self.cmd = ['/src/k8s.io/kubernetes/cluster/test-smoke.sh', '--host',
 136                     os.getenv('KUBE_MASTER_URL')]
 137
 138
 139 class K8sConformanceTest(K8sTesting):
 140     """Kubernetes conformance test suite"""
 141     def __init__(self, **kwargs):
 142         if "case_name" not in kwargs:
 143             kwargs.get("case_name", 'k8s_conformance')
 144         super(K8sConformanceTest, self).__init__(**kwargs)
 145         self.check_envs()
 146         self.cmd = ['/src/k8s.io/kubernetes/_output/bin/e2e.test',
 147                     '-ginkgo.focus', 'Conformance',
 148                     '-kubeconfig', self.config]