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 config import config
  34 import helper
  35 import state
  36 import log
  37
  38 def _create_uid_index(iterable):
  39     index = {}
  40     for item in iterable:
  41         assert item.uid not in index
  42         index[item.uid] = item
  43     return index
  44
  45
  46 class _CommonMetadataMixin:
  47     @property
  48     def name(self):
  49         return self._metadata.name
  50     @property
  51     def uid(self):
  52         return self._metadata.uid
  53     @property
  54     def result(self):
  55         return self._metadata.result
  56     @result.setter
  57     def result(self, result):
  58         self._metadata.result = result
  59
  60     @property
  61     def unsucessful(self):
  62         return self._metadata.result.value != state.Result.Passed
  63
  64
  65 class InternalTestResult(object, _CommonMetadataMixin):
  66     def __init__(self, obj, suite, directory):
  67         self._metadata = obj.metadata
  68         self.suite = suite
  69
  70         self.stderr = os.path.join(
  71             InternalSavedResults.output_path(self.uid, suite.uid),
  72             'stderr'
  73         )
  74         self.stdout = os.path.join(
  75             InternalSavedResults.output_path(self.uid, suite.uid),
  76             'stdout'
  77         )
  78
  79
  80 class InternalSuiteResult(object, _CommonMetadataMixin):
  81     def __init__(self, obj, directory):
  82         self._metadata = obj.metadata
  83         self.directory = directory
  84         self._wrap_tests(obj)
  85
  86     def _wrap_tests(self, obj):
  87         self._tests = [InternalTestResult(test, self, self.directory)
  88                        for test in obj]
  89         self._tests_index = _create_uid_index(self._tests)
  90
  91     def get_test(self, uid):
  92         return self._tests_index[uid]
  93
  94     def __iter__(self):
  95         return iter(self._tests)
  96
  97     def get_test_result(self, uid):
  98         return self.get_test(uid)
  99
 100     def aggregate_test_results(self):
 101         results = {}
 102         for test in self:
 103             helper.append_dictlist(results, test.result.value, test)
 104         return results
 105
 106
 107 class InternalLibraryResults(object, _CommonMetadataMixin):
 108     def __init__(self, obj, directory):
 109         self.directory = directory
 110         self._metadata = obj.metadata
 111         self._wrap_suites(obj)
 112
 113     def __iter__(self):
 114         return iter(self._suites)
 115
 116     def _wrap_suites(self, obj):
 117         self._suites = [InternalSuiteResult(suite, self.directory)
 118                         for suite in obj]
 119         self._suites_index = _create_uid_index(self._suites)
 120
 121     def add_suite(self, suite):
 122         if suite.uid in self._suites:
 123             raise ValueError('Cannot have duplicate suite UIDs.')
 124         self._suites[suite.uid] = suite
 125
 126     def get_suite_result(self, suite_uid):
 127         return self._suites_index[suite_uid]
 128
 129     def get_test_result(self, test_uid, suite_uid):
 130         return self.get_suite_result(suite_uid).get_test_result(test_uid)
 131
 132     def aggregate_test_results(self):
 133         results = {}
 134         for suite in self._suites:
 135             for test in suite:
 136                 helper.append_dictlist(results, test.result.value, test)
 137         return results
 138
 139 class InternalSavedResults:
 140     @staticmethod
 141     def output_path(test_uid, suite_uid, base=None):
 142         '''
 143         Return the path which results for a specific test case should be
 144         stored.
 145         '''
 146         if base is None:
 147             base = config.result_path
 148         return os.path.join(
 149                 base,
 150                 str(suite_uid).replace(os.path.sep, '-'),
 151                 str(test_uid).replace(os.path.sep, '-'))
 152
 153     @staticmethod
 154     def save(results, path, protocol=pickle.HIGHEST_PROTOCOL):
 155         if not os.path.exists(os.path.dirname(path)):
 156            try:
 157                os.makedirs(os.path.dirname(path))
 158            except OSError as exc: # Guard against race condition
 159                if exc.errno != errno.EEXIST:
 160                    raise
 161
 162         with open(path, 'w') as f:
 163             pickle.dump(results, f, protocol)
 164
 165     @staticmethod
 166     def load(path):
 167         with open(path, 'r') as f:
 168             return pickle.load(f)
 169
 170
 171 class XMLElement(object):
 172     def write(self, file_):
 173         self.begin(file_)
 174         self.end(file_)
 175
 176     def begin(self, file_):
 177         file_.write('<')
 178         file_.write(self.name)
 179         for attr in self.attributes:
 180             file_.write(' ')
 181             attr.write(file_)
 182         file_.write('>')
 183
 184         self.body(file_)
 185
 186     def body(self, file_):
 187         for elem in self.elements:
 188             file_.write('\n')
 189             elem.write(file_)
 190         file_.write('\n')
 191
 192     def end(self, file_):
 193         file_.write('</%s>' % self.name)
 194
 195 class XMLAttribute(object):
 196     def __init__(self, name, value):
 197         self.name = name
 198         self.value = value
 199
 200     def write(self, file_):
 201         file_.write('%s=%s' % (self.name,
 202                 xml.sax.saxutils.quoteattr(self.value)))
 203
 204
 205 class JUnitTestSuites(XMLElement):
 206     name = 'testsuites'
 207     result_map = {
 208         state.Result.Errored: 'errors',
 209         state.Result.Failed: 'failures',
 210         state.Result.Passed: 'tests'
 211     }
 212
 213     def __init__(self, internal_results):
 214         results = internal_results.aggregate_test_results()
 215
 216         self.attributes = []
 217         for result, tests in results.items():
 218             self.attributes.append(self.result_attribute(result,
 219                     str(len(tests))))
 220
 221         self.elements = []
 222         for suite in internal_results:
 223             self.elements.append(JUnitTestSuite(suite))
 224
 225     def result_attribute(self, result, count):
 226         return XMLAttribute(self.result_map[result], count)
 227
 228 class JUnitTestSuite(JUnitTestSuites):
 229     name = 'testsuite'
 230     result_map = {
 231         state.Result.Errored: 'errors',
 232         state.Result.Failed: 'failures',
 233         state.Result.Passed: 'tests',
 234         state.Result.Skipped: 'skipped'
 235     }
 236
 237     def __init__(self, suite_result):
 238         results = suite_result.aggregate_test_results()
 239
 240         self.attributes = [
 241             XMLAttribute('name', suite_result.name)
 242         ]
 243         for result, tests in results.items():
 244             self.attributes.append(self.result_attribute(result,
 245                     str(len(tests))))
 246
 247         self.elements = []
 248         for test in suite_result:
 249             self.elements.append(JUnitTestCase(test))
 250
 251     def result_attribute(self, result, count):
 252         return XMLAttribute(self.result_map[result], count)
 253
 254 class JUnitTestCase(XMLElement):
 255     name = 'testcase'
 256     def __init__(self, test_result):
 257         self.attributes = [
 258             XMLAttribute('name', test_result.name),
 259              # TODO JUnit expects class of test.. add as test metadata.
 260             XMLAttribute('classname', str(test_result.uid)),
 261             XMLAttribute('status', str(test_result.result)),
 262         ]
 263
 264         # TODO JUnit expects a message for the reason a test was
 265         # skipped or errored, save this with the test metadata.
 266         # http://llg.cubic.org/docs/junit/
 267         self.elements = [
 268             LargeFileElement('system-err', test_result.stderr),
 269             LargeFileElement('system-out', test_result.stdout),
 270         ]
 271
 272 class LargeFileElement(XMLElement):
 273     def __init__(self, name, filename):
 274         self.name = name
 275         self.filename = filename
 276         self.attributes = []
 277
 278     def body(self, file_):
 279         try:
 280             with open(self.filename, 'r') as f:
 281                 for line in f:
 282                     file_.write(xml.sax.saxutils.escape(line))
 283         except IOError:
 284             # TODO Better error logic, this is sometimes O.K.
 285             # if there was no stdout/stderr captured for the test
 286             #
 287             # TODO If that was the case, the file should still be made and it
 288             # should just be empty instead of not existing.
 289             pass
 290
 291
 292
 293 class JUnitSavedResults:
 294     @staticmethod
 295     def save(results, path):
 296         '''
 297         Compile the internal results into JUnit format writting it to the
 298         given file.
 299         '''
 300         results = JUnitTestSuites(results)
 301         with open(path, 'w') as f:
 302             results.write(f)
 303