#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
A test runner for Jpp.
This script traverses the tests directory and reports a summary.
"""
from __future__ import print_function
from collections import defaultdict
from glob import glob, iglob
import os
from os.path import join, basename
from subprocess import Popen, PIPE
import sys
from time import time
import re
import xml.etree.ElementTree as ET
import xml.dom.minidom
if not len(sys.argv) == 2:
print("Usage: run_tests.py PATH_TO_TESTS")
raise SystemExit
os.environ["TEST_DEBUG"] = "1"
PY2 = sys.version_info < (3, 0, 0)
try:
unichr
except NameError:
unichr = chr
__author__ = "Tamas Gal"
__credits__ = "Brian Beyer"
__license__ = "MIT"
__email__ = "tgal@km3net.de"
__status__ = "Development"
TESTS_DIR = sys.argv[1]
JUNIT_XML = 'out/junit_{}.xml'.format(os.path.basename(TESTS_DIR))
if hasattr(sys.stdout, 'isatty') and sys.stdout.isatty():
INFO = '\033[94m' # blue
OK = '\033[92m' # green
FAIL = '\033[91m' # red
RST = '\033[0m' # colour reset
BOLD = '\033[1m'
else:
INFO, OK, FAIL, RST, BOLD = ('', ) * 5
def main():
test_results = run_tests(TESTS_DIR)
n_tests = len(test_results)
n_failed_tests = sum(1 for r in test_results.values() if r[0] > 0)
total_time = sum(r[1] for r in test_results.values())
print("\n{}"
"Test summary\n"
"============\n{}"
"Total number of tests: {}\n{}"
"Failed tests: {}{}\n"
"Elapsed time: {:.1f}s\n".format(
INFO, RST, n_tests, BOLD + FAIL if n_failed_tests > 0 else OK,
n_failed_tests, RST, total_time))
write_junit_xml(test_results)
if n_failed_tests > 0:
print_captured_output(test_results)
exit(1)
else:
exit(0)
def write_junit_xml(test_results):
"""Generate XML file according to JUnit specs"""
test_cases = []
for test_script, (exit_code, t, stdout, stderr) in test_results.items():
test_case = TestCase(test_script,
elapsed_sec=t,
stdout=stdout,
stderr=stderr)
if exit_code > 0:
test_case.add_error_info('non-zero exit-code: %d' % exit_code)
test_cases.append(test_case)
test_suite = TestSuite("Jpp Test Suite", test_cases)
with open(JUNIT_XML, 'w') as f:
TestSuite.to_file(f, [test_suite])
def print_captured_output(test_results):
"""Prints the STDOUT and STDERR of failing test scripts"""
print("{}"
"Captured output of failing tests\n"
"================================\n{}".format(INFO, RST))
for test_script, (exit_code, t, stdout, stderr) in test_results.items():
if exit_code > 0:
print("{}\n{}\n".format(test_script, len(test_script) * '-'))
print('{}stdout:{}\n{}\n{}stderr:{}\n{}'.format(
OK + BOLD, RST, stdout.decode('utf-8'), FAIL + BOLD, RST,
stderr.decode('utf-8')))
def run_tests(tests_dir):
"""Runs each script in the tests directory and returns the results.
Parameters
----------
tests_dir: str
The path to the test dir, containing the test scripts (`*.sh`).
Returns
-------
dict: key = script path, value = (exit_code, elapsed_time, stdout, stderr)
"""
test_results = {}
for subdir in sorted(glob(join(tests_dir, '*'))):
component_group = basename(subdir)
print("\n{}{}\n{}{}".format(INFO, component_group,
len(component_group) * '=', RST))
for test_script in sorted(glob(join(subdir, '*.sh'))):
print("+ {}".format(test_script), end=' => ')
sys.stdout.flush()
start_time = time()
proc = Popen(test_script, stdout=PIPE, stderr=PIPE)
out, err = proc.communicate()
exit_code = proc.wait()
delta_t = time() - start_time
test_results[test_script] = (exit_code, delta_t, out, err)
print(" ({:.2f} s) ".format(delta_t), end='')
sys.stdout.flush()
if exit_code > 0:
print("{}FAILED (exit code {}){}".format(FAIL, exit_code, RST))
sys.stdout.flush()
else:
print("{}OK{}".format(OK, RST))
sys.stdout.flush()
return test_results
def decode(var, encoding):
"""
If not already unicode, decode it.
"""
if PY2:
if isinstance(var, unicode):
ret = var
elif isinstance(var, str):
if encoding:
ret = var.decode(encoding)
else:
ret = unicode(var)
else:
ret = unicode(var)
else:
ret = str(var)
return ret
class TestSuite(object):
"""
Suite of test cases.
Can handle unicode strings or binary strings if their encoding is provided.
"""
def __init__(self,
name,
test_cases=None,
hostname=None,
id=None,
package=None,
timestamp=None,
properties=None,
file=None,
log=None,
url=None,
stdout=None,
stderr=None):
self.name = name
if not test_cases:
test_cases = []
try:
iter(test_cases)
except TypeError:
raise Exception('test_cases must be a list of test cases')
self.test_cases = test_cases
self.timestamp = timestamp
self.hostname = hostname
self.id = id
self.package = package
self.file = file
self.log = log
self.url = url
self.stdout = stdout
self.stderr = stderr
self.properties = properties
def build_xml_doc(self, encoding=None):
"""
Builds the XML document for the JUnit test suite.
Produces clean unicode strings and decodes non-unicode with the help of encoding.
@param encoding: Used to decode encoded strings.
@return: XML document with unicode string elements
"""
# build the test suite element
test_suite_attributes = dict()
test_suite_attributes['name'] = decode(self.name, encoding)
if any(c.assertions for c in self.test_cases):
test_suite_attributes['assertions'] = \
str(sum([int(c.assertions) for c in self.test_cases if c.assertions]))
test_suite_attributes['disabled'] = \
str(len([c for c in self.test_cases if not c.is_enabled]))
test_suite_attributes['failures'] = \
str(len([c for c in self.test_cases if c.is_failure()]))
test_suite_attributes['errors'] = \
str(len([c for c in self.test_cases if c.is_error()]))
test_suite_attributes['skipped'] = \
str(len([c for c in self.test_cases if c.is_skipped()]))
test_suite_attributes['time'] = \
str(sum(c.elapsed_sec for c in self.test_cases if c.elapsed_sec))
test_suite_attributes['tests'] = str(len(self.test_cases))
if self.hostname:
test_suite_attributes['hostname'] = decode(self.hostname, encoding)
if self.id:
test_suite_attributes['id'] = decode(self.id, encoding)
if self.package:
test_suite_attributes['package'] = decode(self.package, encoding)
if self.timestamp:
test_suite_attributes['timestamp'] = decode(
self.timestamp, encoding)
if self.file:
test_suite_attributes['file'] = decode(self.file, encoding)
if self.log:
test_suite_attributes['log'] = decode(self.log, encoding)
if self.url:
test_suite_attributes['url'] = decode(self.url, encoding)
xml_element = ET.Element("testsuite", test_suite_attributes)
# add any properties
if self.properties:
props_element = ET.SubElement(xml_element, "properties")
for k, v in self.properties.items():
attrs = {
'name': decode(k, encoding),
'value': decode(v, encoding)
}
ET.SubElement(props_element, "property", attrs)
# add test suite stdout
if self.stdout:
stdout_element = ET.SubElement(xml_element, "system-out")
stdout_element.text = decode(self.stdout, encoding)
# add test suite stderr
if self.stderr:
stderr_element = ET.SubElement(xml_element, "system-err")
stderr_element.text = decode(self.stderr, encoding)
# test cases
for case in self.test_cases:
test_case_attributes = dict()
test_case_attributes['name'] = decode(case.name, encoding)
if case.assertions:
# Number of assertions in the test case
test_case_attributes['assertions'] = "%d" % case.assertions
if case.elapsed_sec:
test_case_attributes['time'] = "%f" % case.elapsed_sec
if case.timestamp:
test_case_attributes['timestamp'] = decode(
case.timestamp, encoding)
if case.classname:
test_case_attributes['classname'] = decode(
case.classname, encoding)
if case.status:
test_case_attributes['status'] = decode(case.status, encoding)
if case.category:
test_case_attributes['class'] = decode(case.category, encoding)
if case.file:
test_case_attributes['file'] = decode(case.file, encoding)
if case.line:
test_case_attributes['line'] = decode(case.line, encoding)
if case.log:
test_case_attributes['log'] = decode(case.log, encoding)
if case.url:
test_case_attributes['url'] = decode(case.url, encoding)
test_case_element = ET.SubElement(xml_element, "testcase",
test_case_attributes)
# failures
if case.is_failure():
attrs = {'type': 'failure'}
if case.failure_message:
attrs['message'] = decode(case.failure_message, encoding)
if case.failure_type:
attrs['type'] = decode(case.failure_type, encoding)
failure_element = ET.Element("failure", attrs)
if case.failure_output:
failure_element.text = decode(case.failure_output,
encoding)
test_case_element.append(failure_element)
# errors
if case.is_error():
attrs = {'type': 'error'}
if case.error_message:
attrs['message'] = decode(case.error_message, encoding)
if case.error_type:
attrs['type'] = decode(case.error_type, encoding)
error_element = ET.Element("error", attrs)
if case.error_output:
error_element.text = decode(case.error_output, encoding)
test_case_element.append(error_element)
# skippeds
if case.is_skipped():
attrs = {'type': 'skipped'}
if case.skipped_message:
attrs['message'] = decode(case.skipped_message, encoding)
skipped_element = ET.Element("skipped", attrs)
if case.skipped_output:
skipped_element.text = decode(case.skipped_output,
encoding)
test_case_element.append(skipped_element)
# test stdout
if case.stdout:
stdout_element = ET.Element("system-out")
stdout_element.text = decode(case.stdout, encoding)
test_case_element.append(stdout_element)
# test stderr
if case.stderr:
stderr_element = ET.Element("system-err")
stderr_element.text = decode(case.stderr, encoding)
test_case_element.append(stderr_element)
return xml_element
@staticmethod
def to_xml_string(test_suites, prettyprint=True, encoding=None):
"""
Returns the string representation of the JUnit XML document.
@param encoding: The encoding of the input.
@return: unicode string
"""
try:
iter(test_suites)
except TypeError:
raise Exception('test_suites must be a list of test suites')
xml_element = ET.Element("testsuites")
attributes = defaultdict(int)
for ts in test_suites:
ts_xml = ts.build_xml_doc(encoding=encoding)
for key in ['failures', 'errors', 'tests', 'disabled']:
attributes[key] += int(ts_xml.get(key, 0))
for key in ['time']:
attributes[key] += float(ts_xml.get(key, 0))
xml_element.append(ts_xml)
for key, value in attributes.items():
xml_element.set(key, str(value))
xml_string = ET.tostring(xml_element, encoding=encoding)
# is encoded now
xml_string = TestSuite._clean_illegal_xml_chars(
xml_string.decode(encoding or 'utf-8'))
# is unicode now
if prettyprint:
# minidom.parseString() works just on correctly encoded binary strings
xml_string = xml_string.encode(encoding or 'utf-8')
xml_string = xml.dom.minidom.parseString(xml_string)
# toprettyxml() produces unicode if no encoding is being passed or binary string with an encoding
xml_string = xml_string.toprettyxml(encoding=encoding)
if encoding:
xml_string = xml_string.decode(encoding)
# is unicode now
return xml_string
@staticmethod
def to_file(file_descriptor, test_suites, prettyprint=True, encoding=None):
"""
Writes the JUnit XML document to a file.
"""
xml_string = TestSuite.to_xml_string(test_suites,
prettyprint=prettyprint,
encoding=encoding)
# has problems with encoded str with non-ASCII (non-default-encoding) characters!
file_descriptor.write(xml_string)
@staticmethod
def _clean_illegal_xml_chars(string_to_clean):
"""
Removes any illegal unicode characters from the given XML string.
@see: http://stackoverflow.com/questions/1707890/fast-way-to-filter-illegal-xml-unicode-chars-in-python
"""
illegal_unichrs = [(0x00, 0x08), (0x0B, 0x1F), (0x7F, 0x84),
(0x86, 0x9F), (0xD800, 0xDFFF), (0xFDD0, 0xFDDF),
(0xFFFE, 0xFFFF), (0x1FFFE, 0x1FFFF),
(0x2FFFE, 0x2FFFF), (0x3FFFE, 0x3FFFF),
(0x4FFFE, 0x4FFFF), (0x5FFFE, 0x5FFFF),
(0x6FFFE, 0x6FFFF), (0x7FFFE, 0x7FFFF),
(0x8FFFE, 0x8FFFF), (0x9FFFE, 0x9FFFF),
(0xAFFFE, 0xAFFFF), (0xBFFFE, 0xBFFFF),
(0xCFFFE, 0xCFFFF), (0xDFFFE, 0xDFFFF),
(0xEFFFE, 0xEFFFF), (0xFFFFE, 0xFFFFF),
(0x10FFFE, 0x10FFFF)]
illegal_ranges = [
"%s-%s" % (unichr(low), unichr(high))
for (low, high) in illegal_unichrs if low < sys.maxunicode
]
illegal_xml_re = re.compile('[%s]' % ''.join(illegal_ranges))
return illegal_xml_re.sub('', string_to_clean)
class TestCase(object):
"""A JUnit test case with a result and possibly some stdout or stderr"""
def __init__(self,
name,
classname=None,
elapsed_sec=None,
stdout=None,
stderr=None,
assertions=None,
timestamp=None,
status=None,
category=None,
file=None,
line=None,
log=None,
group=None,
url=None):
self.name = name
self.assertions = assertions
self.elapsed_sec = elapsed_sec
self.timestamp = timestamp
self.classname = classname
self.status = status
self.category = category
self.file = file
self.line = line
self.log = log
self.url = url
self.stdout = stdout
self.stderr = stderr
self.is_enabled = True
self.error_message = None
self.error_output = None
self.error_type = None
self.failure_message = None
self.failure_output = None
self.failure_type = None
self.skipped_message = None
self.skipped_output = None
def add_error_info(self, message=None, output=None, error_type=None):
"""Adds an error message, output, or both to the test case"""
if message:
self.error_message = message
if output:
self.error_output = output
if error_type:
self.error_type = error_type
def add_failure_info(self, message=None, output=None, failure_type=None):
"""Adds a failure message, output, or both to the test case"""
if message:
self.failure_message = message
if output:
self.failure_output = output
if failure_type:
self.failure_type = failure_type
def add_skipped_info(self, message=None, output=None):
"""Adds a skipped message, output, or both to the test case"""
if message:
self.skipped_message = message
if output:
self.skipped_output = output
def is_failure(self):
"""returns true if this test case is a failure"""
return self.failure_output or self.failure_message
def is_error(self):
"""returns true if this test case is an error"""
return self.error_output or self.error_message
def is_skipped(self):
"""returns true if this test case has been skipped"""
return self.skipped_output or self.skipped_message
if __name__ == '__main__':
main()