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.res_dir = "/home/opnfv/functest/results/{}".format(
  37             self.case_name)
  38         self.result = 0
  39         self.start_time = 0
  40         self.stop_time = 0
  41
  42     def run_kubetest(self):  # pylint: disable=too-many-branches
  43         """Run the test suites"""
  44         cmd_line = self.cmd
  45         self.__logger.info("Starting k8s test: '%s'.", cmd_line)
  46
  47         process = subprocess.Popen(cmd_line, stdout=subprocess.PIPE,
  48                                    stderr=subprocess.STDOUT)
  49         boutput = process.stdout.read()
  50         with open(os.path.join(self.res_dir, 'e2e.log'), 'wb') as foutput:
  51             foutput.write(boutput)
  52         grp = re.search(
  53             r'^(FAIL|SUCCESS)!.* ([0-9]+) Passed \| ([0-9]+) Failed \|'
  54             r' ([0-9]+) Pending \| ([0-9]+) Skipped', boutput.decode("utf-8"),
  55             re.MULTILINE | re.DOTALL)
  56         assert grp
  57         self.details['passed'] = int(grp.group(2))
  58         self.details['failed'] = int(grp.group(3))
  59         self.details['pending'] = int(grp.group(4))
  60         self.details['skipped'] = int(grp.group(5))
  61         self.__logger.debug("details: %s", self.details)
  62         self.result = self.details['passed'] * 100 / (
  63             self.details['passed'] + self.details['failed'] +
  64             self.details['pending'])
  65         self.__logger.debug("result: %s", self.result)
  66         if grp.group(1) == 'FAIL':
  67             grp2 = re.search(
  68                 r'^(Summarizing [0-9]+ Failure.*)Ran', boutput.decode("utf-8"),
  69                 re.MULTILINE | re.DOTALL)
  70             if grp2:
  71                 self.__logger.error(grp2.group(1))
  72
  73     def run(self, **kwargs):
  74         if not os.path.isfile(self.config):
  75             self.__logger.error(
  76                 "Cannot run k8s testcases. Config file not found")
  77             return self.EX_RUN_ERROR
  78         self.start_time = time.time()
  79         try:
  80             self.run_kubetest()
  81             res = self.EX_OK
  82         except Exception:  # pylint: disable=broad-except
  83             self.__logger.exception("Error with running kubetest:")
  84             res = self.EX_RUN_ERROR
  85         self.stop_time = time.time()
  86         return res
  87
  88
  89 class K8sSmokeTest(K8sTesting):
  90     """Kubernetes smoke test suite"""
  91     def __init__(self, **kwargs):
  92         if "case_name" not in kwargs:
  93             kwargs.get("case_name", 'k8s_smoke')
  94         super(K8sSmokeTest, self).__init__(**kwargs)
  95         self.cmd = ['e2e.test', '-ginkgo.focus', 'Guestbook.application',
  96                     '-ginkgo.noColor', '-kubeconfig', self.config,
  97                     '-provider', 'local', '-report-dir', self.res_dir,
  98                     '-disable-log-dump', 'true']
  99
 100
 101 class K8sConformanceTest(K8sTesting):
 102     """Kubernetes conformance test suite"""
 103     def __init__(self, **kwargs):
 104         if "case_name" not in kwargs:
 105             kwargs.get("case_name", 'k8s_conformance')
 106         super(K8sConformanceTest, self).__init__(**kwargs)
 107         self.cmd = [
 108             'e2e.test', '-ginkgo.focus', r'\[Conformance\]', '-ginkgo.noColor',
 109             '-ginkgo.skip', r'Alpha|\[(Disruptive|Feature:[^\]]+|Flaky)\]',
 110             '-kubeconfig', self.config, '-provider', 'local',
 111             '-report-dir', self.res_dir, '-disable-log-dump', 'true']