Simple interface for Mbed Cloud Client
Embed:
(wiki syntax)
Show/hide line numbers
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()
Generated on Tue Jul 12 2022 19:01:34 by 1.7.2