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     config = '/root/.kube/config'
  32
  33     def __init__(self, **kwargs):
  34         super(K8sTesting, self).__init__(**kwargs)
  35         self.cmd = []
  36         self.result = 0
  37         self.start_time = 0
  38         self.stop_time = 0
  39
  40     def run_kubetest(self):  # pylint: disable=too-many-branches
  41         """Run the test suites"""
  42         cmd_line = self.cmd
  43         self.__logger.info("Starting k8s test: '%s'.", cmd_line)
  44
  45         process = subprocess.Popen(cmd_line, stdout=subprocess.PIPE,
  46                                    stderr=subprocess.STDOUT)
  47         output = process.stdout.read()
  48         # Remove color code escape sequences
  49         output = re.sub(r'\x1B\[[0-?]*[ -/]*[@-~]', '', str(output))
  50
  51         if ('Error loading client' in output or
  52                 'Unexpected error' in output):
  53             raise Exception(output)
  54
  55         remarks = []
  56         lines = output.split('\n')
  57         success = False
  58         failure = False
  59         i = 0
  60         while i < len(lines):
  61             if '[Fail]' in lines[i] or 'Failures:' in lines[i]:
  62                 self.__logger.error(lines[i])
  63             if re.search(r'\[(.)*[0-9]+ seconds\]', lines[i]):
  64                 self.__logger.debug(lines[i])
  65                 i = i + 1
  66                 while i < len(lines) and lines[i] != '-' * len(lines[i]):
  67                     if lines[i].startswith('STEP:') or ('INFO:' in lines[i]):
  68                         break
  69                     self.__logger.debug(lines[i])
  70                     i = i + 1
  71             if i >= len(lines):
  72                 break
  73             success = 'SUCCESS!' in lines[i]
  74             failure = 'FAIL!' in lines[i]
  75             if success or failure:
  76                 if i != 0 and 'seconds' in lines[i - 1]:
  77                     remarks.append(lines[i - 1])
  78                 remarks = remarks + lines[i].replace('--', '|').split('|')
  79                 break
  80             i = i + 1
  81
  82         self.__logger.debug('-' * 10)
  83         self.__logger.info("Remarks:")
  84         for remark in remarks:
  85             if 'seconds' in remark:
  86                 self.__logger.debug(remark)
  87             elif 'Passed' in remark:
  88                 self.__logger.info("Passed: %s", remark.split()[0])
  89             elif 'Skipped' in remark:
  90                 self.__logger.info("Skipped: %s", remark.split()[0])
  91             elif 'Failed' in remark:
  92                 self.__logger.info("Failed: %s", remark.split()[0])
  93
  94         if success:
  95             self.result = 100
  96         elif failure:
  97             self.result = 0
  98
  99     def run(self, **kwargs):
 100
 101         if not os.path.isfile(self.config):
 102             self.__logger.error(
 103                 "Cannot run k8s testcases. Config file not found")
 104             return self.EX_RUN_ERROR
 105
 106         self.start_time = time.time()
 107         try:
 108             self.run_kubetest()
 109             res = self.EX_OK
 110         except Exception:  # pylint: disable=broad-except
 111             self.__logger.exception("Error with running kubetest:")
 112             res = self.EX_RUN_ERROR
 113
 114         self.stop_time = time.time()
 115         return res
 116
 117
 118 class K8sSmokeTest(K8sTesting):
 119     """Kubernetes smoke test suite"""
 120     def __init__(self, **kwargs):
 121         if "case_name" not in kwargs:
 122             kwargs.get("case_name", 'k8s_smoke')
 123         super(K8sSmokeTest, self).__init__(**kwargs)
 124         self.cmd = ["/src/k8s.io/kubernetes/_output/bin/e2e.test",
 125                     "-ginkgo.focus", "Guestbook.application",
 126                     "-kubeconfig", self.config, "--provider", "local"]
 127
 128
 129 class K8sConformanceTest(K8sTesting):
 130     """Kubernetes conformance test suite"""
 131     def __init__(self, **kwargs):
 132         if "case_name" not in kwargs:
 133             kwargs.get("case_name", 'k8s_conformance')
 134         super(K8sConformanceTest, self).__init__(**kwargs)
 135         self.cmd = ['/src/k8s.io/kubernetes/_output/bin/e2e.test',
 136                     '-ginkgo.focus', 'Conformance',
 137                     '-kubeconfig', self.config, "--provider", "local"]