helpers to run any python method or any bash command.
"""
+import abc
import logging
+import os
import subprocess
+import sys
import time
from xtesting.core import testcase
"Cedric Ollivier <cedric.ollivier@orange.com>")
-class Feature(testcase.TestCase):
+class Feature(testcase.TestCase, metaclass=abc.ABCMeta):
"""Base model for single feature."""
__logger = logging.getLogger(__name__)
- dir_results = "/home/opnfv/xtesting/results"
-
- def __init__(self, **kwargs):
- super(Feature, self).__init__(**kwargs)
- self.result_file = "{}/{}.log".format(self.dir_results, self.case_name)
- try:
- module = kwargs['run']['module']
- self.logger = logging.getLogger(module)
- except KeyError:
- self.__logger.warning(
- "Cannot get module name %s. Using %s as fallback",
- kwargs, self.case_name)
- self.logger = logging.getLogger(self.case_name)
- handler = logging.StreamHandler()
- handler.setLevel(logging.WARN)
- self.logger.addHandler(handler)
- handler = logging.FileHandler(self.result_file)
- handler.setLevel(logging.DEBUG)
- self.logger.addHandler(handler)
- formatter = logging.Formatter(
- '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
- handler.setFormatter(formatter)
- self.logger.addHandler(handler)
+ @abc.abstractmethod
def execute(self, **kwargs):
"""Execute the Python method.
Args:
kwargs: Arbitrary keyword arguments.
-
- Returns:
- -1.
"""
- # pylint: disable=unused-argument,no-self-use
- return -1
def run(self, **kwargs):
"""Run the feature.
self.result = 100
except Exception: # pylint: disable=broad-except
self.__logger.exception("%s FAILED", self.project_name)
- self.__logger.info("Test result is stored in '%s'", self.result_file)
self.stop_time = time.time()
return exit_code
__logger = logging.getLogger(__name__)
+ def __init__(self, **kwargs):
+ super(BashFeature, self).__init__(**kwargs)
+ self.result_file = "{}/{}.log".format(self.res_dir, self.case_name)
+
def execute(self, **kwargs):
"""Execute the cmd passed as arg
0 if cmd returns 0,
-1 otherwise.
"""
- ret = -1
try:
cmd = kwargs["cmd"]
- with open(self.result_file, 'w+') as f_stdout:
- proc = subprocess.Popen(cmd.split(), stdout=f_stdout,
- stderr=subprocess.STDOUT)
- ret = proc.wait()
- if ret != 0:
- self.__logger.error("Execute command: %s failed", cmd)
+ console = kwargs["console"] if "console" in kwargs else False
+ if not os.path.isdir(self.res_dir):
+ os.makedirs(self.res_dir)
+ with open(self.result_file, 'w') as f_stdout:
+ self.__logger.info("Calling %s", cmd)
+ process = subprocess.Popen(
+ cmd, shell=True, stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT)
+ for line in iter(process.stdout.readline, b''):
+ if console:
+ sys.stdout.write(line.decode("utf-8"))
+ f_stdout.write(line.decode("utf-8"))
+ process.wait()
+ with open(self.result_file, 'r') as f_stdin:
+ self.__logger.debug("$ %s\n%s", cmd, f_stdin.read().rstrip())
+ return process.returncode
except KeyError:
self.__logger.error("Please give cmd as arg. kwargs: %s", kwargs)
- return ret
+ return -1