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 functest.core import testcase
  24
  25
  26 LOGGER = logging.getLogger(__name__)
  27
  28
  29 class K8sTesting(testcase.TestCase):
  30     """Kubernetes test runner"""
  31
  32     def __init__(self, **kwargs):
  33         super(K8sTesting, self).__init__(**kwargs)
  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         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         file_logger = logging.getLogger(self.case_name)
  51         remarks = []
  52         lines = output.split('\n')
  53         i = 0
  54         while i < len(lines):
  55             if '[k8s.io]' in lines[i]:
  56                 if i != 0 and 'seconds' in lines[i-1]:
  57                     file_logger.debug(lines[i-1])
  58                 while lines[i] != '-'*len(lines[i]):
  59                     if lines[i].startswith('STEP:') or ('INFO:' in lines[i]):
  60                         break
  61                     file_logger.debug(lines[i])
  62                     i = i+1
  63
  64             success = 'SUCCESS!' in lines[i]
  65             failure = 'FAIL!' in lines[i]
  66             if success or failure:
  67                 if i != 0 and 'seconds' in lines[i-1]:
  68                     remarks.append(lines[i-1])
  69                 remarks = remarks + lines[i].replace('--', '|').split('|')
  70                 break
  71             i = i+1
  72
  73         file_logger.debug('-'*10)
  74         file_logger.debug("Remarks:")
  75         for remark in remarks:
  76             if 'seconds' in remark:
  77                 file_logger.debug(remark)
  78             elif 'Passed' in remark:
  79                 file_logger.debug("Passed: %s", remark.split()[0])
  80             elif 'Skipped' in remark:
  81                 file_logger.debug("Skipped: %s", remark.split()[0])
  82             elif 'Failed' in remark:
  83                 file_logger.debug("Failed: %s", remark.split()[0])
  84
  85         if success:
  86             self.result = 100
  87         elif failure:
  88             self.result = 0
  89
  90     def run(self, **kwargs):
  91
  92         if not os.path.isfile(os.getenv('KUBECONFIG')):
  93             LOGGER.error("Cannot run k8s testcases. Config file not found ")
  94             return self.EX_RUN_ERROR
  95
  96         self.start_time = time.time()
  97         try:
  98             self.run_kubetest()
  99             res = self.EX_OK
 100         except Exception as ex:  # pylint: disable=broad-except
 101             LOGGER.error("Error with running %s", str(ex))
 102             res = self.EX_RUN_ERROR
 103
 104         self.stop_time = time.time()
 105         return res
 106
 107     def check_envs(self):  # pylint: disable=no-self-use
 108         """Check if required environment variables are set"""
 109         try:
 110             assert 'DEPLOY_SCENARIO' in os.environ
 111             assert 'KUBECONFIG' in os.environ
 112             assert 'KUBE_MASTER_IP' in os.environ
 113             assert 'KUBERNETES_PROVIDER' in os.environ
 114             assert 'KUBE_MASTER_URL' in os.environ
 115         except Exception as ex:
 116             raise Exception("Cannot run k8s testcases. "
 117                             "Please check env var: %s" % str(ex))
 118
 119
 120 class K8sSmokeTest(K8sTesting):
 121     """Kubernetes smoke test suite"""
 122     def __init__(self, **kwargs):
 123         if "case_name" not in kwargs:
 124             kwargs.get("case_name", 'k8s_smoke')
 125         super(K8sSmokeTest, self).__init__(**kwargs)
 126         self.check_envs()
 127         self.cmd = ['/src/k8s.io/kubernetes/cluster/test-smoke.sh', '--host',
 128                     os.getenv('KUBE_MASTER_URL')]
 129
 130
 131 class K8sConformanceTest(K8sTesting):
 132     """Kubernetes conformance test suite"""
 133     def __init__(self, **kwargs):
 134         if "case_name" not in kwargs:
 135             kwargs.get("case_name", 'k8s_conformance')
 136         super(K8sConformanceTest, self).__init__(**kwargs)
 137         self.check_envs()
 138         self.cmd = ['/src/k8s.io/kubernetes/_output/bin/e2e.test',
 139                     '--ginkgo.focus', 'Conformance']