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().__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         with subprocess.Popen(
  47                 ['subunit-stats'], stdin=subprocess.PIPE,
  48                 stdout=subprocess.PIPE) as stats:
  49             output, _ = stats.communicate(stream.read())
  50             cls.__logger.info("\n\n%s", output.decode("utf-8"))
  51
  52     def generate_xunit(self, stream):
  53         """Generate junit report from subunit stream
  54
  55         Raises:
  56             Exception
  57         """
  58         stream.seek(0)
  59         with open(f"{self.res_dir}/results.xml", "w", encoding='utf-8') as xml:
  60             with subprocess.Popen(
  61                     ['subunit2junitxml'], stdin=subprocess.PIPE,
  62                     stdout=subprocess.PIPE) as stats:
  63                 output, _ = stats.communicate(stream.read())
  64                 xml.write(output.decode("utf-8"))
  65
  66     def generate_html(self, stream):
  67         """Generate html report from subunit stream
  68
  69         Raises:
  70             CalledProcessError
  71         """
  72         cmd = ['subunit2html', stream, f'{self.res_dir}/results.html']
  73         output = subprocess.check_output(cmd)
  74         self.__logger.debug("\n%s\n\n%s", ' '.join(cmd), output)
  75
  76     def run(self, **kwargs):
  77         """Run the test suite.
  78
  79         It allows running any unittest.TestSuite and getting its
  80         execution status.
  81
  82         By default, it runs the suite defined as instance attribute.
  83         It can be overriden by passing name as arg. It must
  84         conform with TestLoader.loadTestsFromName().
  85
  86         It sets the following attributes required to push the results
  87         to DB:
  88
  89             * result,
  90             * start_time,
  91             * stop_time,
  92             * details.
  93
  94         Args:
  95             kwargs: Arbitrary keyword arguments.
  96
  97         Return:
  98             TestCase.EX_OK if any TestSuite has been run
  99             TestCase.EX_RUN_ERROR otherwise.
 100         """
 101         try:
 102             name = kwargs["name"]
 103             try:
 104                 self.suite = unittest.TestLoader().loadTestsFromName(name)
 105             except ImportError:
 106                 self.__logger.error("Can not import %s", name)
 107                 return testcase.TestCase.EX_RUN_ERROR
 108         except KeyError:
 109             pass
 110         try:
 111             assert self.suite
 112             self.start_time = time.time()
 113             if not os.path.isdir(self.res_dir):
 114                 os.makedirs(self.res_dir)
 115             stream = BytesIO()
 116             result = SubunitTestRunner(
 117                 stream=stream, verbosity=2).run(self.suite).decorated
 118             self.generate_stats(stream)
 119             self.generate_xunit(stream)
 120             with open(f'{self.res_dir}/subunit_stream', 'wb') as subfd:
 121                 stream.seek(0)
 122                 shutil.copyfileobj(stream, subfd)
 123             self.generate_html(f'{self.res_dir}/subunit_stream')
 124             self.stop_time = time.time()
 125             self.details = {
 126                 "testsRun": result.testsRun,
 127                 "failures": len(result.failures),
 128                 "errors": len(result.errors)}
 129             self.result = 100 * (
 130                 (result.testsRun - (len(result.failures) +
 131                                     len(result.errors))) /
 132                 result.testsRun)
 133             return testcase.TestCase.EX_OK
 134         except AssertionError:
 135             self.__logger.error("No suite is defined")
 136             return testcase.TestCase.EX_RUN_ERROR
 137         except ZeroDivisionError:
 138             self.__logger.error("No test has been run")
 139             return testcase.TestCase.EX_RUN_ERROR
 140         except Exception:  # pylint: disable=broad-except
 141             self.__logger.exception("something wrong occurs")
 142             return testcase.TestCase.EX_RUN_ERROR