Simulated product dispenser

Dependencies:   HTS221

Fork of mbed-cloud-workshop-connect-HTS221 by Jim Carver

Embed: (wiki syntax)

« Back to documentation index

Show/hide line numbers junit_cmd_wrapper.py Source File

junit_cmd_wrapper.py

00001 #!/usr/bin/env python
00002 import argparse
00003 import logging
00004 import string
00005 import subprocess
00006 import sys
00007 import time
00008 import xml.etree.cElementTree as ElementTree
00009 from io import BytesIO
00010 from threading import Thread
00011 
00012 logger = logging.getLogger('ta-junit-wrapper')
00013 
00014 
00015 class TeeBytesIO (BytesIO):
00016     """duplicate each write command to an additional file object"""
00017     def __init__(self, tee_fh):
00018         self.tee_fh  = tee_fh
00019         super(TeeBytesIO, self).__init__()
00020 
00021     def write(self, s):
00022         self.tee_fh .write(s)
00023         BytesIO.write(self, s)
00024 
00025 
00026 def get_parser():
00027     parser = argparse.ArgumentParser(description='JUNIT wrapper')
00028 
00029     parser.add_argument(
00030         '-o',
00031         '--output-file',
00032         metavar='FILE',
00033         type=argparse.FileType('w'),
00034         help='output JUNIT XML file name',
00035         required=True
00036     )
00037 
00038     parser.add_argument(
00039         '-s',
00040         '--test-suite',
00041         metavar='NAME',
00042         help='test suite name',
00043         required=True
00044     )
00045 
00046     parser.add_argument(
00047         '-t',
00048         '--test-case',
00049         metavar='NAME',
00050         help='test case name',
00051         required=True
00052     )
00053 
00054     parser.add_argument(
00055         '-v',
00056         '--verbose',
00057         action='store_true',
00058         help='verbose - duplicate command output to STDOUT'
00059     )
00060 
00061     parser.add_argument(
00062         '--validate',
00063         action='store_true',
00064         help='validate generated XML against Jenkins XSD. Requires "requests" and "lxml" libraries'
00065     )
00066 
00067     parser.add_argument(
00068         '--command',
00069         nargs=argparse.REMAINDER,
00070         help='command to be executed'
00071     )
00072     return parser
00073 
00074 
00075 def get_file_copy_worker(infile, outfile):
00076     def do_work(_infile, _outfile):
00077         for line in iter(_infile.readline, ''):
00078             _outfile.write(line)
00079         _infile.close()
00080     thread = Thread(target=do_work, args=(infile, outfile))
00081     thread.daemon = True
00082     thread.start()
00083     return thread
00084 
00085 
00086 def generate_junit_xml(test_suite, test_case, out_fh, stdout, stderr, return_code, duration_in_sec, command):
00087     test_suite_root_element = ElementTree.Element(
00088         'testsuite',
00089         tests='1',
00090         name=test_suite.replace(' ', '_'),
00091         failures=str(1 if return_code != 0 else 0),
00092         time=str(duration_in_sec)
00093     )
00094     test_case_element = ElementTree.SubElement(
00095         test_suite_root_element,
00096         'testcase',
00097         time=str(duration_in_sec),
00098         name=test_case.replace(' ', '_')
00099     )
00100     ElementTree.SubElement(test_case_element, 'system-out').text = filter(
00101         lambda x: x in string.printable, stdout.getvalue()
00102     )
00103     ElementTree.SubElement(test_case_element, 'system-err').text = filter(
00104         lambda x: x in string.printable, stderr.getvalue()
00105     )
00106 
00107     if return_code != 0:
00108         failure_msg = 'Command "{cmd}" returned {ret}'.format(cmd=command, ret=return_code)
00109         ElementTree.SubElement(
00110             test_case_element,
00111             'failure',
00112             type='Non-Zero return code',
00113             message=failure_msg)
00114 
00115     ElementTree.ElementTree(test_suite_root_element).write(out_fh)
00116 
00117 
00118 def main():
00119     parser = get_parser()
00120     args = parser.parse_args()
00121     logging.basicConfig(
00122         level=logging.DEBUG if args.verbose else logging.INFO,
00123         format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
00124         stream=sys.stdout
00125     )
00126 
00127     stdout = TeeBytesIO(sys.stdout) if args.verbose else BytesIO()
00128     stderr = TeeBytesIO(stdout)
00129     logger.debug('Executing + ' + ' '.join(args.command))
00130     start_time = time.time()
00131     process = subprocess.Popen(args.command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
00132     threads = [
00133         get_file_copy_worker(process.stdout, stdout),
00134         get_file_copy_worker(process.stderr, stderr)
00135     ]
00136     for t in threads:  # wait for IO completion
00137         t.join()
00138     return_code = process.wait()
00139     logger.debug('Wrapped process return code ' + str(return_code))
00140     duration_in_sec = time.time() - start_time
00141 
00142     with args.output_file as fh:  # insure file object is closed - since it will be read in do_validate()
00143         generate_junit_xml(
00144             args.test_suite,
00145             args.test_case,
00146             fh,
00147             stdout,
00148             stderr,
00149             return_code,
00150             duration_in_sec,
00151             args.command
00152         )
00153     logger.debug('Generated JUNIT report file ' + args.output_file.name)
00154 
00155     if args.validate:
00156         do_validate(args.output_file.name)
00157 
00158     raise SystemExit(return_code)
00159 
00160 if __name__ == '__main__':
00161     main()