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 config = '/root/.kube/config'
33 def __init__(self, **kwargs):
34 super(K8sTesting, self).__init__(**kwargs)
40 def run_kubetest(self): # pylint: disable=too-many-branches
41 """Run the test suites"""
43 self.__logger.info("Starting k8s test: '%s'.", cmd_line)
45 process = subprocess.Popen(cmd_line, stdout=subprocess.PIPE,
46 stderr=subprocess.STDOUT)
47 output = process.stdout.read()
48 # Remove color code escape sequences
49 output = re.sub(r'\x1B\[[0-?]*[ -/]*[@-~]', '', str(output))
51 if ('Error loading client' in output or
52 'Unexpected error' in output):
53 raise Exception(output)
56 lines = output.split('\n')
61 if '[Fail]' in lines[i] or 'Failures:' in lines[i]:
62 self.__logger.error(lines[i])
63 if re.search(r'\[(.)*[0-9]+ seconds\]', lines[i]):
64 self.__logger.debug(lines[i])
66 while i < len(lines) and lines[i] != '-' * len(lines[i]):
67 if lines[i].startswith('STEP:') or ('INFO:' in lines[i]):
69 self.__logger.debug(lines[i])
73 success = 'SUCCESS!' in lines[i]
74 failure = 'FAIL!' in lines[i]
75 if success or failure:
76 if i != 0 and 'seconds' in lines[i - 1]:
77 remarks.append(lines[i - 1])
78 remarks = remarks + lines[i].replace('--', '|').split('|')
82 self.__logger.debug('-' * 10)
83 self.__logger.info("Remarks:")
84 for remark in remarks:
85 if 'seconds' in remark:
86 self.__logger.debug(remark)
87 elif 'Passed' in remark:
88 self.__logger.info("Passed: %s", remark.split()[0])
89 elif 'Skipped' in remark:
90 self.__logger.info("Skipped: %s", remark.split()[0])
91 elif 'Failed' in remark:
92 self.__logger.info("Failed: %s", remark.split()[0])
99 def run(self, **kwargs):
101 if not os.path.isfile(self.config):
103 "Cannot run k8s testcases. Config file not found")
104 return self.EX_RUN_ERROR
106 self.start_time = time.time()
110 except Exception: # pylint: disable=broad-except
111 self.__logger.exception("Error with running kubetest:")
112 res = self.EX_RUN_ERROR
114 self.stop_time = time.time()
118 class K8sSmokeTest(K8sTesting):
119 """Kubernetes smoke test suite"""
120 def __init__(self, **kwargs):
121 if "case_name" not in kwargs:
122 kwargs.get("case_name", 'k8s_smoke')
123 super(K8sSmokeTest, self).__init__(**kwargs)
124 self.cmd = ["e2e.test", "-ginkgo.focus", "Guestbook.application",
125 "-kubeconfig", self.config, "--provider", "local"]
128 class K8sConformanceTest(K8sTesting):
129 """Kubernetes conformance test suite"""
130 def __init__(self, **kwargs):
131 if "case_name" not in kwargs:
132 kwargs.get("case_name", 'k8s_conformance')
133 super(K8sConformanceTest, self).__init__(**kwargs)
134 self.cmd = ['e2e.test', '-ginkgo.focus', 'Conformance',
135 '-kubeconfig', self.config, "--provider", "local"]