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