3 # Copyright (c) 2016 Cable Television Laboratories, Inc. and others.
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
10 """Define the parent class to run unittest.TestSuite as TestCase."""
12 from __future__ import division
21 from subunit.run import SubunitTestRunner
24 from xtesting.core import testcase
26 __author__ = ("Steven Pisarski <s.pisarski@cablelabs.com>, "
27 "Cedric Ollivier <cedric.ollivier@orange.com>")
30 class Suite(testcase.TestCase):
31 """Base model for running unittest.TestSuite."""
33 __logger = logging.getLogger(__name__)
35 def __init__(self, **kwargs):
36 super(Suite, self).__init__(**kwargs)
37 self.res_dir = "/var/lib/xtesting/results/{}".format(self.case_name)
41 def generate_stats(cls, stream):
42 """Generate stats from subunit stream
48 stats = subprocess.Popen(
49 ['subunit-stats'], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
50 output, _ = stats.communicate(stream.read())
51 cls.__logger.info("\n\n%s", output.decode("utf-8"))
53 def generate_xunit(self, stream):
54 """Generate junit report from subunit stream
60 with open("{}/results.xml".format(self.res_dir), "w") as xml:
61 stats = subprocess.Popen(
62 ['subunit2junitxml'], stdin=subprocess.PIPE,
63 stdout=subprocess.PIPE)
64 output, _ = stats.communicate(stream.read())
65 xml.write(output.decode("utf-8"))
67 def generate_html(self, stream):
68 """Generate html report from subunit stream
73 cmd = ['subunit2html', stream, '{}/results.html'.format(self.res_dir)]
74 output = subprocess.check_output(cmd)
75 self.__logger.debug("\n%s\n\n%s", ' '.join(cmd), output)
77 def run(self, **kwargs):
78 """Run the test suite.
80 It allows running any unittest.TestSuite and getting its
83 By default, it runs the suite defined as instance attribute.
84 It can be overriden by passing name as arg. It must
85 conform with TestLoader.loadTestsFromName().
87 It sets the following attributes required to push the results
96 kwargs: Arbitrary keyword arguments.
99 TestCase.EX_OK if any TestSuite has been run
100 TestCase.EX_RUN_ERROR otherwise.
103 name = kwargs["name"]
105 self.suite = unittest.TestLoader().loadTestsFromName(name)
107 self.__logger.error("Can not import %s", name)
108 return testcase.TestCase.EX_RUN_ERROR
113 self.start_time = time.time()
114 if not os.path.isdir(self.res_dir):
115 os.makedirs(self.res_dir)
116 stream = six.BytesIO()
117 result = SubunitTestRunner(
118 stream=stream, verbosity=2).run(self.suite).decorated
119 self.generate_stats(stream)
120 self.generate_xunit(stream)
121 with open('{}/subunit_stream'.format(self.res_dir), 'wb') as subfd:
123 shutil.copyfileobj(stream, subfd)
124 self.generate_html('{}/subunit_stream'.format(self.res_dir))
125 self.stop_time = time.time()
127 "testsRun": result.testsRun,
128 "failures": len(result.failures),
129 "errors": len(result.errors)}
130 self.result = 100 * (
131 (result.testsRun - (len(result.failures) +
132 len(result.errors))) /
134 return testcase.TestCase.EX_OK
135 except AssertionError:
136 self.__logger.error("No suite is defined")
137 return testcase.TestCase.EX_RUN_ERROR
138 except ZeroDivisionError:
139 self.__logger.error("No test has been run")
140 return testcase.TestCase.EX_RUN_ERROR
141 except Exception: # pylint: disable=broad-except
142 self.__logger.exception("something wrong occurs")
143 return testcase.TestCase.EX_RUN_ERROR