xtesting/core/unit.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2016 Cable Television Laboratories, Inc. 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 class to run unittest.TestSuite as TestCase."""
  11
  12 from __future__ import division
  13 from io import BytesIO
  14 import logging
  15 import os
  16 import shutil
  17 import subprocess
  18 import time
  19 import unittest
  20
  21 from subunit.run import SubunitTestRunner
  22
  23 from xtesting.core import testcase
  24
  25 __author__ = ("Steven Pisarski <s.pisarski@cablelabs.com>, "
  26               "Cedric Ollivier <cedric.ollivier@orange.com>")
  27
  28
  29 class Suite(testcase.TestCase):
  30     """Base model for running unittest.TestSuite."""
  31
  32     __logger = logging.getLogger(__name__)
  33
  34     def __init__(self, **kwargs):
  35         super(Suite, self).__init__(**kwargs)
  36         self.suite = None
  37
  38     @classmethod
  39     def generate_stats(cls, stream):
  40         """Generate stats from subunit stream
  41
  42         Raises:
  43             Exception
  44         """
  45         stream.seek(0)
  46         stats = subprocess.Popen(
  47             ['subunit-stats'], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
  48         output, _ = stats.communicate(stream.read())
  49         cls.__logger.info("\n\n%s", output.decode("utf-8"))
  50
  51     def generate_xunit(self, stream):
  52         """Generate junit report from subunit stream
  53
  54         Raises:
  55             Exception
  56         """
  57         stream.seek(0)
  58         with open("{}/results.xml".format(self.res_dir), "w") as xml:
  59             stats = subprocess.Popen(
  60                 ['subunit2junitxml'], stdin=subprocess.PIPE,
  61                 stdout=subprocess.PIPE)
  62             output, _ = stats.communicate(stream.read())
  63             xml.write(output.decode("utf-8"))
  64
  65     def generate_html(self, stream):
  66         """Generate html report from subunit stream
  67
  68         Raises:
  69             CalledProcessError
  70         """
  71         cmd = ['subunit2html', stream, '{}/results.html'.format(self.res_dir)]
  72         output = subprocess.check_output(cmd)
  73         self.__logger.debug("\n%s\n\n%s", ' '.join(cmd), output)
  74
  75     def run(self, **kwargs):
  76         """Run the test suite.
  77
  78         It allows running any unittest.TestSuite and getting its
  79         execution status.
  80
  81         By default, it runs the suite defined as instance attribute.
  82         It can be overriden by passing name as arg. It must
  83         conform with TestLoader.loadTestsFromName().
  84
  85         It sets the following attributes required to push the results
  86         to DB:
  87
  88             * result,
  89             * start_time,
  90             * stop_time,
  91             * details.
  92
  93         Args:
  94             kwargs: Arbitrary keyword arguments.
  95
  96         Return:
  97             TestCase.EX_OK if any TestSuite has been run
  98             TestCase.EX_RUN_ERROR otherwise.
  99         """
 100         try:
 101             name = kwargs["name"]
 102             try:
 103                 self.suite = unittest.TestLoader().loadTestsFromName(name)
 104             except ImportError:
 105                 self.__logger.error("Can not import %s", name)
 106                 return testcase.TestCase.EX_RUN_ERROR
 107         except KeyError:
 108             pass
 109         try:
 110             assert self.suite
 111             self.start_time = time.time()
 112             if not os.path.isdir(self.res_dir):
 113                 os.makedirs(self.res_dir)
 114             stream = BytesIO()
 115             result = SubunitTestRunner(
 116                 stream=stream, verbosity=2).run(self.suite).decorated
 117             self.generate_stats(stream)
 118             self.generate_xunit(stream)
 119             with open('{}/subunit_stream'.format(self.res_dir), 'wb') as subfd:
 120                 stream.seek(0)
 121                 shutil.copyfileobj(stream, subfd)
 122             self.generate_html('{}/subunit_stream'.format(self.res_dir))
 123             self.stop_time = time.time()
 124             self.details = {
 125                 "testsRun": result.testsRun,
 126                 "failures": len(result.failures),
 127                 "errors": len(result.errors)}
 128             self.result = 100 * (
 129                 (result.testsRun - (len(result.failures) +
 130                                     len(result.errors))) /
 131                 result.testsRun)
 132             return testcase.TestCase.EX_OK
 133         except AssertionError:
 134             self.__logger.error("No suite is defined")
 135             return testcase.TestCase.EX_RUN_ERROR
 136         except ZeroDivisionError:
 137             self.__logger.error("No test has been run")
 138             return testcase.TestCase.EX_RUN_ERROR
 139         except Exception:  # pylint: disable=broad-except
 140             self.__logger.exception("something wrong occurs")
 141             return testcase.TestCase.EX_RUN_ERROR