dokku/tests/shellcheck-to-junit

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import argparse
import collections
import datetime
import re
import socket
import sys

from xml.etree import ElementTree


def CDATA(text=None):
    element = ElementTree.Element('![CDATA[')
    element.text = text
    return element


def _serialize_xml(write, elem, qnames, namespaces,short_empty_elements, **kwargs):

    if elem.tag == '![CDATA[':
        write("\n<{}{}]]>\n".format(elem.tag, elem.text))
        if elem.tail:
            write(ElementTree._escape_cdata(elem.tail))
    else:
        return ElementTree._original_serialize_xml(write, elem, qnames, namespaces,short_empty_elements, **kwargs)


ElementTree._original_serialize_xml = ElementTree._serialize_xml
ElementTree._serialize_xml = ElementTree._serialize['xml'] = _serialize_xml


def read_in():
    lines = sys.stdin.readlines()
    for i in range(len(lines)):
        lines[i] = lines[i].rstrip()
    return lines


def process_lines(lines):
    files = {}
    current_file = None
    previous_line = None
    line_no = None
    new_issues = []
    code = None

    RE_VIOLATION = re.compile(r"\^(-)+(\^)? (SC[\w]+) \((\w+)\): (.*)")
    HTTP_VIOLATION = re.compile(r"^https?://")
    line_num = 0
    skip_next_line = False

    for line in lines:
        # start a new block
        if skip_next_line:
            skip_next_line = False
            continue
        elif line == '':
            if current_file is not None:
                file_data = files.get(current_file, {})
                files[current_file] = file_data

                issue_data = file_data.get(line_no, {})
                issue_data['code'] = code
                files[current_file][line_no] = issue_data

                issues = issue_data.get('issues', [])
                issues.extend(new_issues)
                issue_data['issues'] = issues

                files[current_file][line_no] = issue_data

            code = None
            current_file = None
            line_no = None
        elif line.startswith('In ./') and not previous_line:
            current_file = line.split(' ')[1].replace('./', '')
            line_no = line.split(' ')[3]
            new_issues = []
            code = None
        elif code is None and len(new_issues) == 0:
            code = line
        elif line.startswith('Did you mean:') or line.startswith('For more information:') or HTTP_VIOLATION.match(line.strip()):
            skip_next_line = True
            pass
        else:
            match = RE_VIOLATION.match(line.strip())
            if not match:
                print('Error: Issue parsing line "{0}"'.format(line.strip()))
            else:
                next_line_num = line_num + 2
                if next_line_num < len(lines):
                    next_line = lines[next_line_num]
                    if next_line.startswith('Did you mean:'):
                        line += '\n' + lines[next_line_num] + '\n' + lines[next_line_num+1]
                new_issues.append({
                    'shellcheck_id': match.group(1),
                    'message': match.group(2),
                    'original_message': line
                })

        previous_line = line
        line_num += 1
    return files


def output_junit(files, args):
    timestamp = datetime.datetime.now().replace(microsecond=0).isoformat()
    failures = 0
    for file, data in files.items():
        for line, issue_data in data.items():
            code = issue_data.get('code')
            for issue in issue_data.get('issues', []):
                failures += 1

    tests = 0
    if args.files:
        with open(args.files, 'r') as f:
            tests = len(f.readlines())

    if tests < failures:
          tests = failures

    root = ElementTree.Element("testsuite",
                               name="shellcheck",
                               tests="{0}".format(tests),
                               failures="{0}".format(failures),
                               errors="0",
                               skipped="0",
                               timestamp=timestamp,
                               time="0",
                               hostname=socket.gethostname())

    properties = ElementTree.SubElement(root, "properties")
    if args.exclude:
        ElementTree.SubElement(properties,
                               "property",
                               name="exclude",
                               value=args.exclude)

    if args.files:
        with open(args.files, 'r') as f:
            lines = f.readlines()
            for i in range(len(lines)):
                file = lines[i].rstrip().replace('./', '')
                data = files.get(file, None)
                if data:
                    for line, issue_data in data.items():
                        code = issue_data.get('code')
                        for issue in issue_data.get('issues', []):
                            testcase = ElementTree.SubElement(root,
                                                              "testcase",
                                                              classname=file,
                                                              name=file,
                                                              time="0")
                            shellcheck_id = issue.get('shellcheck_id')
                            message = 'line {0}: {1}'.format(
                              line, issue.get('message'))
                            original_message = issue.get('original_message')
                            e = ElementTree.Element("failure",
                                                    type=shellcheck_id,
                                                    message=message)
                            cdata = CDATA("\n".join([code, original_message]))
                            e.append(cdata)
                            testcase.append(e)
                ElementTree.SubElement(root,
                                       "testcase",
                                       classname=file,
                                       name=file,
                                       time="0")

    ElementTree.SubElement(root, "system-out")
    ElementTree.SubElement(root, "system-err")

    content = ElementTree.tostring(root, encoding='UTF-8', method='xml')
    if args.output:
        with open(args.output, 'w') as f:
            f.write(content.decode("utf-8"))


def main():
    parser = argparse.ArgumentParser(
        description='Process shellcheck output to junit.')
    parser.add_argument('--output',
                        dest='output',
                        action='store',
                        default=None,
                        help='file to write shellcheck output')
    parser.add_argument('--files',
                        dest='files',
                        action='store',
                        default=None,
                        help='a file containing a list of all files processed by shellcheck')
    parser.add_argument('--exclude',
                        dest='exclude',
                        action='store',
                        default=None,
                        help='a comma-separated list of rules being excluded by shellcheck')
    args = parser.parse_args()

    lines = read_in()
    files = process_lines(lines)
    files = collections.OrderedDict(sorted(files.items()))
    output_junit(files, args)
    for line in lines:
        print(line)


if __name__ == '__main__':
    main()