import abc
import logging
+import os
import subprocess
+import sys
import time
-import six
from xtesting.core import testcase
__author__ = ("Serena Feng <feng.xiaowei@zte.com.cn>, "
"Cedric Ollivier <cedric.ollivier@orange.com>")
-@six.add_metaclass(abc.ABCMeta)
-class Feature(testcase.TestCase):
+class Feature(testcase.TestCase, metaclass=abc.ABCMeta):
"""Base model for single feature."""
__logger = logging.getLogger(__name__)
def __init__(self, **kwargs):
super(BashFeature, self).__init__(**kwargs)
- dir_results = "/var/lib/xtesting/results"
- self.result_file = "{}/{}.log".format(dir_results, self.case_name)
+ 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()
- self.__logger.info(
- "Test result is stored in '%s'", self.result_file)
- 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