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
8 # http://www.apache.org/licenses/LICENSE-2.0
12 Define the parent for Kubernetes testing.
15 from __future__ import division
23 from xtesting.core import testcase
26 class K8sTesting(testcase.TestCase):
27 """Kubernetes test runner"""
29 __logger = logging.getLogger(__name__)
31 def __init__(self, **kwargs):
32 super(K8sTesting, self).__init__(**kwargs)
33 self.config = '/root/.kube/config'
39 def run_kubetest(self): # pylint: disable=too-many-branches
40 """Run the test suites"""
42 self.__logger.info("Starting k8s test: '%s'.", cmd_line)
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))
50 if ('Error loading client' in output or
51 'Unexpected error' in output):
52 raise Exception(output)
55 lines = output.split('\n')
60 if '[Fail]' in lines[i] or 'Failures:' in lines[i]:
61 self.__logger.error(lines[i])
62 if re.search(r'\[(.)*[0-9]+ seconds\]', lines[i]):
63 self.__logger.debug(lines[i])
65 while i < len(lines) and lines[i] != '-' * len(lines[i]):
66 if lines[i].startswith('STEP:') or ('INFO:' in lines[i]):
68 self.__logger.debug(lines[i])
72 success = 'SUCCESS!' in lines[i]
73 failure = 'FAIL!' in lines[i]
74 if success or failure:
75 if i != 0 and 'seconds' in lines[i - 1]:
76 remarks.append(lines[i - 1])
77 remarks = remarks + lines[i].replace('--', '|').split('|')
81 self.__logger.debug('-' * 10)
82 self.__logger.info("Remarks:")
83 for remark in remarks:
84 if 'seconds' in remark:
85 self.__logger.debug(remark)
86 elif 'Passed' in remark:
87 self.__logger.info("Passed: %s", remark.split()[0])
88 elif 'Skipped' in remark:
89 self.__logger.info("Skipped: %s", remark.split()[0])
90 elif 'Failed' in remark:
91 self.__logger.info("Failed: %s", remark.split()[0])
98 def run(self, **kwargs):
100 if not os.path.isfile(self.config):
102 "Cannot run k8s testcases. Config file not found")
103 return self.EX_RUN_ERROR
105 self.start_time = time.time()
109 except Exception: # pylint: disable=broad-except
110 self.__logger.exception("Error with running kubetest:")
111 res = self.EX_RUN_ERROR
113 self.stop_time = time.time()
116 def check_envs(self): # pylint: disable=no-self-use
117 """Check if required environment variables are set"""
119 assert 'DEPLOY_SCENARIO' in os.environ
120 assert 'KUBE_MASTER_IP' in os.environ
121 assert 'KUBERNETES_PROVIDER' in os.environ
122 assert 'KUBE_MASTER_URL' in os.environ
123 except Exception as ex:
124 raise Exception("Cannot run k8s testcases. "
125 "Please check env var: %s" % str(ex))
128 class K8sSmokeTest(K8sTesting):
129 """Kubernetes smoke test suite"""
130 def __init__(self, **kwargs):
131 if "case_name" not in kwargs:
132 kwargs.get("case_name", 'k8s_smoke')
133 super(K8sSmokeTest, self).__init__(**kwargs)
135 self.cmd = ['/src/k8s.io/kubernetes/cluster/test-smoke.sh', '--host',
136 os.getenv('KUBE_MASTER_URL')]
139 class K8sConformanceTest(K8sTesting):
140 """Kubernetes conformance test suite"""
141 def __init__(self, **kwargs):
142 if "case_name" not in kwargs:
143 kwargs.get("case_name", 'k8s_conformance')
144 super(K8sConformanceTest, self).__init__(**kwargs)
146 self.cmd = ['/src/k8s.io/kubernetes/_output/bin/e2e.test',
147 '-ginkgo.focus', 'Conformance',
148 '-kubeconfig', self.config]