xtesting/core/feature.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2016 ZTE Corp and others.
   4 #
   5 # All rights reserved. This program and the accompanying materials
   6 # are made available under the terms of the Apache License, Version 2.0
   7 # which accompanies this distribution, and is available at
   8 # http://www.apache.org/licenses/LICENSE-2.0
   9
  10 """Define the parent classes of all Xtesting Features.
  11
  12 Feature is considered as TestCase offered by Third-party. It offers
  13 helpers to run any python method or any bash command.
  14 """
  15
  16 import abc
  17 import logging
  18 import os
  19 import subprocess
  20 import sys
  21 import time
  22
  23 import six
  24 from xtesting.core import testcase
  25
  26 __author__ = ("Serena Feng <feng.xiaowei@zte.com.cn>, "
  27               "Cedric Ollivier <cedric.ollivier@orange.com>")
  28
  29
  30 @six.add_metaclass(abc.ABCMeta)
  31 class Feature(testcase.TestCase):
  32     """Base model for single feature."""
  33
  34     __logger = logging.getLogger(__name__)
  35
  36     @abc.abstractmethod
  37     def execute(self, **kwargs):
  38         """Execute the Python method.
  39
  40         The subclasses must override the default implementation which
  41         is false on purpose.
  42
  43         The new implementation must return 0 if success or anything
  44         else if failure.
  45
  46         Args:
  47             kwargs: Arbitrary keyword arguments.
  48         """
  49
  50     def run(self, **kwargs):
  51         """Run the feature.
  52
  53         It allows executing any Python method by calling execute().
  54
  55         It sets the following attributes required to push the results
  56         to DB:
  57
  58             * result,
  59             * start_time,
  60             * stop_time.
  61
  62         It doesn't fulfill details when pushing the results to the DB.
  63
  64         Args:
  65             kwargs: Arbitrary keyword arguments.
  66
  67         Returns:
  68             TestCase.EX_OK if execute() returns 0,
  69             TestCase.EX_RUN_ERROR otherwise.
  70         """
  71         self.start_time = time.time()
  72         exit_code = testcase.TestCase.EX_RUN_ERROR
  73         self.result = 0
  74         try:
  75             if self.execute(**kwargs) == 0:
  76                 exit_code = testcase.TestCase.EX_OK
  77                 self.result = 100
  78         except Exception:  # pylint: disable=broad-except
  79             self.__logger.exception("%s FAILED", self.project_name)
  80         self.stop_time = time.time()
  81         return exit_code
  82
  83
  84 class BashFeature(Feature):
  85     """Class designed to run any bash command."""
  86
  87     __logger = logging.getLogger(__name__)
  88
  89     def __init__(self, **kwargs):
  90         super(BashFeature, self).__init__(**kwargs)
  91         self.res_dir = "/var/lib/xtesting/results/{}".format(self.case_name)
  92         self.result_file = "{}/{}.log".format(self.res_dir, self.case_name)
  93
  94     def execute(self, **kwargs):
  95         """Execute the cmd passed as arg
  96
  97         Args:
  98             kwargs: Arbitrary keyword arguments.
  99
 100         Returns:
 101             0 if cmd returns 0,
 102             -1 otherwise.
 103         """
 104         try:
 105             cmd = kwargs["cmd"]
 106             console = kwargs["console"] if "console" in kwargs else False
 107             if not os.path.isdir(self.res_dir):
 108                 os.makedirs(self.res_dir)
 109             with open(self.result_file, 'w') as f_stdout:
 110                 self.__logger.info("Calling %s", cmd)
 111                 process = subprocess.Popen(
 112                     cmd, shell=True, stdout=subprocess.PIPE,
 113                     stderr=subprocess.STDOUT)
 114                 for line in iter(process.stdout.readline, b''):
 115                     if console:
 116                         sys.stdout.write(line.decode())
 117                     f_stdout.write(line.decode())
 118                 process.wait()
 119             with open(self.result_file, 'r') as f_stdin:
 120                 self.__logger.debug("$ %s\n%s", cmd, f_stdin.read().rstrip())
 121             return process.returncode
 122         except KeyError:
 123             self.__logger.error("Please give cmd as arg. kwargs: %s", kwargs)
 124         return -1