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 subprocess
  20 import time
  21
  22 from functest.core import testcase
  23
  24
  25 LOGGER = logging.getLogger(__name__)
  26
  27
  28 class K8sTesting(testcase.TestCase):
  29     """Kubernetes test runner"""
  30
  31     def __init__(self, **kwargs):
  32         super(K8sTesting, self).__init__(**kwargs)
  33         self.cmd = []
  34         self.result = 0
  35         self.start_time = 0
  36         self.stop_time = 0
  37
  38     def run_kubetest(self):
  39         """Run the test suites"""
  40         cmd_line = self.cmd
  41         LOGGER.info("Starting k8s test: '%s'.", cmd_line)
  42
  43         process = subprocess.Popen(cmd_line, stdout=subprocess.PIPE,
  44                                    stderr=subprocess.STDOUT)
  45         remark = []
  46         lines = process.stdout.readlines()
  47         for i in range(len(lines) - 1, -1, -1):
  48             new_line = str(lines[i])
  49
  50             if 'SUCCESS!' in new_line or 'FAIL!' in new_line:
  51                 remark = new_line.replace('--', '|').split('|')
  52                 break
  53
  54         if remark and 'SUCCESS!' in remark[0]:
  55             self.result = 100
  56
  57     def run(self):
  58
  59         if not os.path.isfile(os.getenv('KUBECONFIG')):
  60             LOGGER.error("Cannot run k8s testcases. Config file not found ")
  61             return self.EX_RUN_ERROR
  62
  63         self.start_time = time.time()
  64         try:
  65             self.run_kubetest()
  66             res = self.EX_OK
  67         except Exception as ex:  # pylint: disable=broad-except
  68             LOGGER.error("Error with running %s", str(ex))
  69             res = self.EX_RUN_ERROR
  70
  71         self.stop_time = time.time()
  72         return res
  73
  74     def check_envs(self):  # pylint: disable=no-self-use
  75         """Check if required environment variables are set"""
  76         try:
  77             assert 'DEPLOY_SCENARIO' in os.environ
  78             assert 'KUBECONFIG' in os.environ
  79             assert 'KUBE_MASTER' in os.environ
  80             assert 'KUBE_MASTER_IP' in os.environ
  81             assert 'KUBERNETES_PROVIDER' in os.environ
  82             assert 'KUBE_MASTER_URL' in os.environ
  83         except Exception as ex:
  84             raise Exception("Cannot run k8s testcases. "
  85                             "Please check env var: %s" % str(ex))
  86
  87
  88 class K8sSmokeTest(K8sTesting):
  89     """Kubernetes smoke test suite"""
  90     def __init__(self, **kwargs):
  91         if "case_name" not in kwargs:
  92             kwargs.get("case_name", 'k8s_smoke')
  93         super(K8sSmokeTest, self).__init__(**kwargs)
  94         self.check_envs()
  95         self.cmd = ['/src/k8s.io/kubernetes/cluster/test-smoke.sh', '--host',
  96                     os.getenv('KUBE_MASTER_URL')]