blob: 786c21b609168eb9137a1e5adc1a64e32ef8f8cc [file] [log] [blame]
# Copyright (c) 2017 Mark D. Hill and David A. Wood
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met: redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer;
# redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution;
# neither the name of the copyright holders nor the names of its
# contributors may be used to endorse or promote products derived from
# this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#
# Authors: Sean Wilson
import os
import pickle
import xml.sax.saxutils
from config import config
import helper
import state
import log
def _create_uid_index(iterable):
index = {}
for item in iterable:
assert item.uid not in index
index[item.uid] = item
return index
class _CommonMetadataMixin:
@property
def name(self):
return self._metadata.name
@property
def uid(self):
return self._metadata.uid
@property
def result(self):
return self._metadata.result
@result.setter
def result(self, result):
self._metadata.result = result
@property
def unsuccessful(self):
return self._metadata.result.value != state.Result.Passed
class InternalTestResult(object, _CommonMetadataMixin):
def __init__(self, obj, suite, directory):
self._metadata = obj.metadata
self.suite = suite
self.stderr = os.path.join(
InternalSavedResults.output_path(self.uid, suite.uid),
'stderr'
)
self.stdout = os.path.join(
InternalSavedResults.output_path(self.uid, suite.uid),
'stdout'
)
class InternalSuiteResult(object, _CommonMetadataMixin):
def __init__(self, obj, directory):
self._metadata = obj.metadata
self.directory = directory
self._wrap_tests(obj)
def _wrap_tests(self, obj):
self._tests = [InternalTestResult(test, self, self.directory)
for test in obj]
self._tests_index = _create_uid_index(self._tests)
def get_test(self, uid):
return self._tests_index[uid]
def __iter__(self):
return iter(self._tests)
def get_test_result(self, uid):
return self.get_test(uid)
def aggregate_test_results(self):
results = {}
for test in self:
helper.append_dictlist(results, test.result.value, test)
return results
class InternalLibraryResults(object, _CommonMetadataMixin):
def __init__(self, obj, directory):
self.directory = directory
self._metadata = obj.metadata
self._wrap_suites(obj)
def __iter__(self):
return iter(self._suites)
def _wrap_suites(self, obj):
self._suites = [InternalSuiteResult(suite, self.directory)
for suite in obj]
self._suites_index = _create_uid_index(self._suites)
def add_suite(self, suite):
if suite.uid in self._suites:
raise ValueError('Cannot have duplicate suite UIDs.')
self._suites[suite.uid] = suite
def get_suite_result(self, suite_uid):
return self._suites_index[suite_uid]
def get_test_result(self, test_uid, suite_uid):
return self.get_suite_result(suite_uid).get_test_result(test_uid)
def aggregate_test_results(self):
results = {}
for suite in self._suites:
for test in suite:
helper.append_dictlist(results, test.result.value, test)
return results
class InternalSavedResults:
@staticmethod
def output_path(test_uid, suite_uid, base=None):
'''
Return the path which results for a specific test case should be
stored.
'''
if base is None:
base = config.result_path
return os.path.join(
base,
str(suite_uid).replace(os.path.sep, '-'),
str(test_uid).replace(os.path.sep, '-'))
@staticmethod
def save(results, path, protocol=pickle.HIGHEST_PROTOCOL):
if not os.path.exists(os.path.dirname(path)):
try:
os.makedirs(os.path.dirname(path))
except OSError as exc: # Guard against race condition
if exc.errno != errno.EEXIST:
raise
with open(path, 'w') as f:
pickle.dump(results, f, protocol)
@staticmethod
def load(path):
with open(path, 'r') as f:
return pickle.load(f)
class XMLElement(object):
def write(self, file_):
self.begin(file_)
self.end(file_)
def begin(self, file_):
file_.write('<')
file_.write(self.name)
for attr in self.attributes:
file_.write(' ')
attr.write(file_)
file_.write('>')
self.body(file_)
def body(self, file_):
for elem in self.elements:
file_.write('\n')
elem.write(file_)
file_.write('\n')
def end(self, file_):
file_.write('</%s>' % self.name)
class XMLAttribute(object):
def __init__(self, name, value):
self.name = name
self.value = value
def write(self, file_):
file_.write('%s=%s' % (self.name,
xml.sax.saxutils.quoteattr(self.value)))
class JUnitTestSuites(XMLElement):
name = 'testsuites'
result_map = {
state.Result.Errored: 'errors',
state.Result.Failed: 'failures',
state.Result.Passed: 'tests'
}
def __init__(self, internal_results):
results = internal_results.aggregate_test_results()
self.attributes = []
for result, tests in results.items():
self.attributes.append(self.result_attribute(result,
str(len(tests))))
self.elements = []
for suite in internal_results:
self.elements.append(JUnitTestSuite(suite))
def result_attribute(self, result, count):
return XMLAttribute(self.result_map[result], count)
class JUnitTestSuite(JUnitTestSuites):
name = 'testsuite'
result_map = {
state.Result.Errored: 'errors',
state.Result.Failed: 'failures',
state.Result.Passed: 'tests',
state.Result.Skipped: 'skipped'
}
def __init__(self, suite_result):
results = suite_result.aggregate_test_results()
self.attributes = [
XMLAttribute('name', suite_result.name)
]
for result, tests in results.items():
self.attributes.append(self.result_attribute(result,
str(len(tests))))
self.elements = []
for test in suite_result:
self.elements.append(JUnitTestCase(test))
def result_attribute(self, result, count):
return XMLAttribute(self.result_map[result], count)
class JUnitTestCase(XMLElement):
name = 'testcase'
def __init__(self, test_result):
self.attributes = [
XMLAttribute('name', test_result.name),
# TODO JUnit expects class of test.. add as test metadata.
XMLAttribute('classname', str(test_result.uid)),
XMLAttribute('status', str(test_result.result)),
]
# TODO JUnit expects a message for the reason a test was
# skipped or errored, save this with the test metadata.
# http://llg.cubic.org/docs/junit/
self.elements = [
LargeFileElement('system-err', test_result.stderr),
LargeFileElement('system-out', test_result.stdout),
]
if str(test_result.result) == 'Failed':
self.elements.append(JUnitFailure('Test failed', 'ERROR'))
class JUnitFailure(XMLElement):
name = 'failure'
def __init__(self, message, fail_type):
self.attributes = [
XMLAttribute('message', message),
XMLAttribute('type', fail_type),
]
self.elements = []
class LargeFileElement(XMLElement):
def __init__(self, name, filename):
self.name = name
self.filename = filename
self.attributes = []
def body(self, file_):
try:
with open(self.filename, 'r') as f:
for line in f:
file_.write(xml.sax.saxutils.escape(line))
except IOError:
# TODO Better error logic, this is sometimes O.K.
# if there was no stdout/stderr captured for the test
#
# TODO If that was the case, the file should still be made and it
# should just be empty instead of not existing.
pass
class JUnitSavedResults:
@staticmethod
def save(results, path):
'''
Compile the internal results into JUnit format writting it to the
given file.
'''
results = JUnitTestSuites(results)
with open(path, 'w') as f:
results.write(f)