ext/testlib/result.py

   1 # Copyright (c) 2017 Mark D. Hill and David A. Wood
   2 # All rights reserved.
   3 #
   4 # Redistribution and use in source and binary forms, with or without
   5 # modification, are permitted provided that the following conditions are
   6 # met: redistributions of source code must retain the above copyright
   7 # notice, this list of conditions and the following disclaimer;
   8 # redistributions in binary form must reproduce the above copyright
   9 # notice, this list of conditions and the following disclaimer in the
  10 # documentation and/or other materials provided with the distribution;
  11 # neither the name of the copyright holders nor the names of its
  12 # contributors may be used to endorse or promote products derived from
  13 # this software without specific prior written permission.
  14 #
  15 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  16 # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  17 # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  18 # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  19 # OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  20 # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  21 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  22 # DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  23 # THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  24 # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  25 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  26 #
  27 # Authors: Sean Wilson
  28
  29 import os
  30 import pickle
  31 import xml.sax.saxutils
  32
  33 from testlib.configuration import config
  34 import testlib.helper as helper
  35 import testlib.state as state
  36
  37 def _create_uid_index(iterable):
  38     index = {}
  39     for item in iterable:
  40         assert item.uid not in index
  41         index[item.uid] = item
  42     return index
  43
  44
  45 class _CommonMetadataMixin:
  46     @property
  47     def name(self):
  48         return self._metadata.name
  49     @property
  50     def uid(self):
  51         return self._metadata.uid
  52     @property
  53     def result(self):
  54         return self._metadata.result
  55     @result.setter
  56     def result(self, result):
  57         self._metadata.result = result
  58
  59     @property
  60     def unsuccessful(self):
  61         return self._metadata.result.value != state.Result.Passed
  62
  63
  64 class InternalTestResult(_CommonMetadataMixin):
  65     def __init__(self, obj, suite, directory):
  66         self._metadata = obj.metadata
  67         self.suite = suite
  68
  69         self.stderr = os.path.join(
  70             InternalSavedResults.output_path(self.uid, suite.uid),
  71             'stderr'
  72         )
  73         self.stdout = os.path.join(
  74             InternalSavedResults.output_path(self.uid, suite.uid),
  75             'stdout'
  76         )
  77
  78
  79 class InternalSuiteResult(_CommonMetadataMixin):
  80     def __init__(self, obj, directory):
  81         self._metadata = obj.metadata
  82         self.directory = directory
  83         self._wrap_tests(obj)
  84
  85     def _wrap_tests(self, obj):
  86         self._tests = [InternalTestResult(test, self, self.directory)
  87                        for test in obj]
  88         self._tests_index = _create_uid_index(self._tests)
  89
  90     def get_test(self, uid):
  91         return self._tests_index[uid]
  92
  93     def __iter__(self):
  94         return iter(self._tests)
  95
  96     def get_test_result(self, uid):
  97         return self.get_test(uid)
  98
  99     def aggregate_test_results(self):
 100         results = {}
 101         for test in self:
 102             helper.append_dictlist(results, test.result.value, test)
 103         return results
 104
 105
 106 class InternalLibraryResults(_CommonMetadataMixin):
 107     def __init__(self, obj, directory):
 108         self.directory = directory
 109         self._metadata = obj.metadata
 110         self._wrap_suites(obj)
 111
 112     def __iter__(self):
 113         return iter(self._suites)
 114
 115     def _wrap_suites(self, obj):
 116         self._suites = [InternalSuiteResult(suite, self.directory)
 117                         for suite in obj]
 118         self._suites_index = _create_uid_index(self._suites)
 119
 120     def add_suite(self, suite):
 121         if suite.uid in self._suites:
 122             raise ValueError('Cannot have duplicate suite UIDs.')
 123         self._suites[suite.uid] = suite
 124
 125     def get_suite_result(self, suite_uid):
 126         return self._suites_index[suite_uid]
 127
 128     def get_test_result(self, test_uid, suite_uid):
 129         return self.get_suite_result(suite_uid).get_test_result(test_uid)
 130
 131     def aggregate_test_results(self):
 132         results = {}
 133         for suite in self._suites:
 134             for test in suite:
 135                 helper.append_dictlist(results, test.result.value, test)
 136         return results
 137
 138 class InternalSavedResults:
 139     @staticmethod
 140     def output_path(test_uid, suite_uid, base=None):
 141         '''
 142         Return the path which results for a specific test case should be
 143         stored.
 144         '''
 145         if base is None:
 146             base = config.result_path
 147         return os.path.join(
 148                 base,
 149                 str(suite_uid).replace(os.path.sep, '-'),
 150                 str(test_uid).replace(os.path.sep, '-'))
 151
 152     @staticmethod
 153     def save(results, path, protocol=pickle.HIGHEST_PROTOCOL):
 154         if not os.path.exists(os.path.dirname(path)):
 155            try:
 156                os.makedirs(os.path.dirname(path))
 157            except OSError as exc: # Guard against race condition
 158                if exc.errno != errno.EEXIST:
 159                    raise
 160
 161         with open(path, 'wb') as f:
 162             pickle.dump(results, f, protocol)
 163
 164     @staticmethod
 165     def load(path):
 166         with open(path, 'rb') as f:
 167             return pickle.load(f)
 168
 169
 170 class XMLElement(object):
 171     def write(self, file_):
 172         self.begin(file_)
 173         self.end(file_)
 174
 175     def begin(self, file_):
 176         file_.write('<')
 177         file_.write(self.name)
 178         for attr in self.attributes:
 179             file_.write(' ')
 180             attr.write(file_)
 181         file_.write('>')
 182
 183         self.body(file_)
 184
 185     def body(self, file_):
 186         for elem in self.elements:
 187             file_.write('\n')
 188             elem.write(file_)
 189         file_.write('\n')
 190
 191     def end(self, file_):
 192         file_.write('</%s>' % self.name)
 193
 194 class XMLAttribute(object):
 195     def __init__(self, name, value):
 196         self.name = name
 197         self.value = value
 198
 199     def write(self, file_):
 200         file_.write('%s=%s' % (self.name,
 201                 xml.sax.saxutils.quoteattr(self.value)))
 202
 203
 204 class JUnitTestSuites(XMLElement):
 205     name = 'testsuites'
 206     result_map = {
 207         state.Result.Errored: 'errors',
 208         state.Result.Failed: 'failures',
 209         state.Result.Passed: 'tests'
 210     }
 211
 212     def __init__(self, internal_results):
 213         results = internal_results.aggregate_test_results()
 214
 215         self.attributes = []
 216         for result, tests in results.items():
 217             self.attributes.append(self.result_attribute(result,
 218                     str(len(tests))))
 219
 220         self.elements = []
 221         for suite in internal_results:
 222             self.elements.append(JUnitTestSuite(suite))
 223
 224     def result_attribute(self, result, count):
 225         return XMLAttribute(self.result_map[result], count)
 226
 227 class JUnitTestSuite(JUnitTestSuites):
 228     name = 'testsuite'
 229     result_map = {
 230         state.Result.Errored: 'errors',
 231         state.Result.Failed: 'failures',
 232         state.Result.Passed: 'tests',
 233         state.Result.Skipped: 'skipped'
 234     }
 235
 236     def __init__(self, suite_result):
 237         results = suite_result.aggregate_test_results()
 238
 239         self.attributes = [
 240             XMLAttribute('name', suite_result.name)
 241         ]
 242         for result, tests in results.items():
 243             self.attributes.append(self.result_attribute(result,
 244                     str(len(tests))))
 245
 246         self.elements = []
 247         for test in suite_result:
 248             self.elements.append(JUnitTestCase(test))
 249
 250     def result_attribute(self, result, count):
 251         return XMLAttribute(self.result_map[result], count)
 252
 253 class JUnitTestCase(XMLElement):
 254     name = 'testcase'
 255     def __init__(self, test_result):
 256         self.attributes = [
 257             XMLAttribute('name', test_result.name),
 258              # TODO JUnit expects class of test.. add as test metadata.
 259             XMLAttribute('classname', str(test_result.uid)),
 260             XMLAttribute('status', str(test_result.result)),
 261         ]
 262
 263         # TODO JUnit expects a message for the reason a test was
 264         # skipped or errored, save this with the test metadata.
 265         # http://llg.cubic.org/docs/junit/
 266         self.elements = [
 267             LargeFileElement('system-err', test_result.stderr),
 268             LargeFileElement('system-out', test_result.stdout),
 269         ]
 270
 271         if str(test_result.result) == 'Failed':
 272             self.elements.append(JUnitFailure('Test failed', 'ERROR'))
 273
 274
 275 class JUnitFailure(XMLElement):
 276     name = 'failure'
 277     def __init__(self, message, fail_type):
 278         self.attributes = [
 279             XMLAttribute('message', message),
 280             XMLAttribute('type', fail_type),
 281         ]
 282         self.elements = []
 283
 284
 285 class LargeFileElement(XMLElement):
 286     def __init__(self, name, filename):
 287         self.name = name
 288         self.filename = filename
 289         self.attributes = []
 290
 291     def body(self, file_):
 292         try:
 293             with open(self.filename, 'r') as f:
 294                 for line in f:
 295                     file_.write(xml.sax.saxutils.escape(line))
 296         except IOError:
 297             # TODO Better error logic, this is sometimes O.K.
 298             # if there was no stdout/stderr captured for the test
 299             #
 300             # TODO If that was the case, the file should still be made and it
 301             # should just be empty instead of not existing.
 302             pass
 303
 304
 305
 306 class JUnitSavedResults:
 307     @staticmethod
 308     def save(results, path):
 309         '''
 310         Compile the internal results into JUnit format writting it to the
 311         given file.
 312         '''
 313         results = JUnitTestSuites(results)
 314         with open(path, 'w') as f:
 315             results.write(f)
 316