Set utf-8 in decode and encode calls
[functest-xtesting.git] / 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
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 import six
23
24 from xtesting.core import testcase
25
26 __author__ = ("Steven Pisarski <s.pisarski@cablelabs.com>, "
27               "Cedric Ollivier <cedric.ollivier@orange.com>")
28
29
30 class Suite(testcase.TestCase):
31     """Base model for running unittest.TestSuite."""
32
33     __logger = logging.getLogger(__name__)
34
35     def __init__(self, **kwargs):
36         super(Suite, self).__init__(**kwargs)
37         self.res_dir = "/var/lib/xtesting/results/{}".format(self.case_name)
38         self.suite = None
39
40     @classmethod
41     def generate_stats(cls, stream):
42         """Generate stats from subunit stream
43
44         Raises:
45             Exception
46         """
47         stream.seek(0)
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"))
52
53     def generate_xunit(self, stream):
54         """Generate junit report from subunit stream
55
56         Raises:
57             Exception
58         """
59         stream.seek(0)
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"))
66
67     def generate_html(self, stream):
68         """Generate html report from subunit stream
69
70         Raises:
71             CalledProcessError
72         """
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)
76
77     def run(self, **kwargs):
78         """Run the test suite.
79
80         It allows running any unittest.TestSuite and getting its
81         execution status.
82
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().
86
87         It sets the following attributes required to push the results
88         to DB:
89
90             * result,
91             * start_time,
92             * stop_time,
93             * details.
94
95         Args:
96             kwargs: Arbitrary keyword arguments.
97
98         Return:
99             TestCase.EX_OK if any TestSuite has been run
100             TestCase.EX_RUN_ERROR otherwise.
101         """
102         try:
103             name = kwargs["name"]
104             try:
105                 self.suite = unittest.TestLoader().loadTestsFromName(name)
106             except ImportError:
107                 self.__logger.error("Can not import %s", name)
108                 return testcase.TestCase.EX_RUN_ERROR
109         except KeyError:
110             pass
111         try:
112             assert self.suite
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:
122                 stream.seek(0)
123                 shutil.copyfileobj(stream, subfd)
124             self.generate_html('{}/subunit_stream'.format(self.res_dir))
125             self.stop_time = time.time()
126             self.details = {
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))) /
133                 result.testsRun)
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