[loggers]
-keys=root,functest,functest_kubernetes
+keys=root,functest,functest_kubernetes,k8s_smoke,k8s_conformance
[handlers]
-keys=console,wconsole,file,null
+keys=console,wconsole,file,null,k8s_smoke_logfile,k8s_conformance_logfile
[formatters]
keys=standard
[logger_functest_kubernetes]
level=NOTSET
-handlers=console
+handlers=console,file
qualname=functest_kubernetes
[logger_functest]
handlers=console
qualname=functest
+[logger_k8s_smoke]
+level=NOTSET
+handlers=k8s_smoke_logfile
+qualname=k8s_smoke
+
+[logger_k8s_conformance]
+level=NOTSET
+handlers=k8s_conformance_logfile
+qualname=k8s_conformance
+
[handler_null]
class=NullHandler
level=NOTSET
formatter=standard
args=("/home/opnfv/functest/results/functest-kubernetes.log",)
+[handler_k8s_smoke_logfile]
+class=FileHandler
+level=DEBUG
+formatter=standard
+args=("/home/opnfv/functest/results/k8s_smoke.log",)
+
+[handler_k8s_conformance_logfile]
+class=FileHandler
+level=DEBUG
+formatter=standard
+args=("/home/opnfv/functest/results/k8s_conformance.log",)
+
[formatter_standard]
format=%(asctime)s - %(name)s - %(levelname)s - %(message)s
datefmt=
import logging
import os
+import re
import subprocess
import time
self.start_time = 0
self.stop_time = 0
- def run_kubetest(self):
+ def run_kubetest(self): # pylint: disable=too-many-branches
"""Run the test suites"""
cmd_line = self.cmd
LOGGER.info("Starting k8s test: '%s'.", cmd_line)
process = subprocess.Popen(cmd_line, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT)
- remark = []
- lines = process.stdout.readlines()
- for i in range(len(lines) - 1, -1, -1):
- new_line = str(lines[i])
-
- if 'SUCCESS!' in new_line or 'FAIL!' in new_line:
- remark = new_line.replace('--', '|').split('|')
+ output = process.stdout.read()
+ # Remove color code escape sequences
+ output = re.sub(r'\x1B\[[0-?]*[ -/]*[@-~]', '', str(output))
+
+ file_logger = logging.getLogger(self.case_name)
+ remarks = []
+ lines = output.split('\n')
+ i = 0
+ while i < len(lines):
+ if '[k8s.io]' in lines[i]:
+ if i != 0 and 'seconds' in lines[i-1]:
+ file_logger.debug(lines[i-1])
+ while lines[i] != '-'*len(lines[i]):
+ if lines[i].startswith('STEP:') or ('INFO:' in lines[i]):
+ break
+ file_logger.debug(lines[i])
+ i = i+1
+
+ success = 'SUCCESS!' in lines[i]
+ failure = 'FAIL!' in lines[i]
+ if success or failure:
+ if i != 0 and 'seconds' in lines[i-1]:
+ remarks.append(lines[i-1])
+ remarks = remarks + lines[i].replace('--', '|').split('|')
break
-
- if remark and 'SUCCESS!' in remark[0]:
+ i = i+1
+
+ file_logger.debug('-'*10)
+ file_logger.debug("Remarks:")
+ for remark in remarks:
+ if 'seconds' in remark:
+ file_logger.debug(remark)
+ elif 'Passed' in remark:
+ file_logger.debug("Passed: %s", remark.split()[0])
+ elif 'Skipped' in remark:
+ file_logger.debug("Skipped: %s", remark.split()[0])
+ elif 'Failed' in remark:
+ file_logger.debug("Failed: %s", remark.split()[0])
+
+ if success:
self.result = 100
+ elif failure:
+ self.result = 0
- def run(self):
+ def run(self, **kwargs):
if not os.path.isfile(os.getenv('KUBECONFIG')):
LOGGER.error("Cannot run k8s testcases. Config file not found ")
self.check_envs()
self.cmd = ['/src/k8s.io/kubernetes/cluster/test-smoke.sh', '--host',
os.getenv('KUBE_MASTER_URL')]
+
+
+class K8sConformanceTest(K8sTesting):
+ """Kubernetes conformance test suite"""
+ def __init__(self, **kwargs):
+ if "case_name" not in kwargs:
+ kwargs.get("case_name", 'k8s_conformance')
+ super(K8sConformanceTest, self).__init__(**kwargs)
+ self.check_envs()
+ self.cmd = ['/src/k8s.io/kubernetes/_output/bin/e2e.test',
+ '--ginkgo.focus', 'Conformance']