galaxy/scripts/tests_markdown.py

#!/usr/bin/env python
import argparse
import json
import os
import statistics
import sys

import jinja2
from mir import html_report


DESCRIPTION = "Script to generate (potentially merged) HTML summary of Galaxy Test Performance"
templateLoader = jinja2.FileSystemLoader(searchpath="./scripts")
template_env = jinja2.Environment(loader=templateLoader)
TEMPLATE_FILE = "tests_markdown.tpl"
TEMPLATE_COMPARE_FILE = "tests_markdown_compare.tpl"
LINKS = [
    {"href": "https://github.com/galaxyproject/galaxy", "title": "Galaxy"}
]


def main(argv=None):
    if argv is None:
        argv = sys.argv[1:]

    args = _parser().parse_args(argv)
    common_env = {
        "include_raw_metrics": args.include_raw_metrics,
    }
    if len(args.input_path) == 1:
        markdown_output = _summarize_one(args.input_path[0], common_env)
    else:
        markdown_output = _summarize_many(args.input_path, common_env)

    html_output = html_report(markdown_output, args.title, links=LINKS)
    with open(args.output_path, "w") as f:
        f.write(html_output)


def _summarize_one(input_path, common_env):
    environment = _prepare_raw_data(input_path)
    environment.update(common_env)
    template = template_env.get_template(TEMPLATE_FILE)
    output = template.render(environment)
    return output


def _summarize_many(input_paths, common_env):
    raw_data_dicts = []
    for arg in input_paths:
        raw_data_dicts.append(_prepare_raw_data(arg))
    environment = _merge_summarizes(raw_data_dicts)
    environment.update(common_env)
    template = template_env.get_template(TEMPLATE_COMPARE_FILE)
    output = template.render(environment)
    return output


def _empty_statistics():
    return {"count": 0, "sum": "n/a", "median": "n/a", "stdev": "n/a", "mean": "n/a"}


def _merge_summarizes(raw_data_dicts):
    tests = {}
    api_endpoints = {}
    internals = {}

    all_tests = set()
    all_api_endpoints = set()
    all_internal_metrics = set()
    all_labels = set()

    for raw_data_dict in raw_data_dicts:
        these_api_endpoints = raw_data_dict['raw_data']['api_endpoint_metrics']
        for api_endpoint in these_api_endpoints.keys():
            all_api_endpoints.add(api_endpoint)
        these_internal_endpoints = raw_data_dict['raw_data']['internals_metrics']
        for internal_endpoint in these_internal_endpoints.keys():
            all_internal_metrics.add(internal_endpoint)
        these_tests = raw_data_dict['raw_data']['tests']
        for test in these_tests:
            all_tests.add(test['nodeid'])
        all_labels.add(raw_data_dict["label"])

    for label in all_labels:
        for api_endpoint in all_api_endpoints:
            _ensure_has_dict_at_key(api_endpoints, api_endpoint,
                total_time={},
                sql_time={},
                sql_queries={},
            )
            api_endpoints[api_endpoint]["total_time"][label] = _empty_statistics()
            api_endpoints[api_endpoint]["sql_time"][label] = _empty_statistics()
        for internal_endpoint in all_internal_metrics:
            _ensure_has_dict_at_key(internals, internal_endpoint,
                total_time={},
            )
            internals[internal_endpoint]["total_time"][label] = _empty_statistics()
        for test in all_tests:
            if test not in tests:
                tests[test] = {}
            tests[test][label] = {"outcome": "absent"}

    for raw_data_dict in raw_data_dicts:
        ab_label = raw_data_dict['label']
        these_api_endpoints = raw_data_dict['raw_data']['api_endpoint_metrics']
        these_internals = raw_data_dict['raw_data']['internals_metrics']
        these_tests = raw_data_dict['raw_data']['tests']

        for api_endpoint, endpoint_metrics in these_api_endpoints.items():
            api_endpoints[api_endpoint]["label"] = endpoint_metrics["label"]
            api_endpoints[api_endpoint]["total_time"][ab_label].update(endpoint_metrics["total_time"])
            api_endpoints[api_endpoint]["sql_time"][ab_label].update(endpoint_metrics["sql_time"])

        for endpoint, endpoint_metrics in these_internals.items():
            internals[endpoint]["label"] = endpoint_metrics["label"]
            internals[endpoint]['total_time'][ab_label].update(endpoint_metrics["total_time"])

        for test in these_tests:
            tests[test["nodeid"]][ab_label]["outcome"] = test.get('outcome')

    return {"raw_data": {
        "api_endpoint_metrics": api_endpoints,
        "internals_metrics": internals,
        "tests": tests,
    }}


def _prepare_raw_data(path):
    with open(path) as f:
        inp_dict = json.load(f)
    environment = dict(raw_data=inp_dict, label=os.path.basename(path))
    __inject_summary(environment)
    __inject_api_timing_summary_environment(environment)
    __inject_api_timing_summary_across_tests(environment)
    __inject_raw_timings(environment)
    return environment


def __inject_api_timing_summary_environment(environment):
    for test in environment['raw_data']['tests']:
        if 'metadata' not in test:
            continue

        if 'local_metrics' in test['metadata']:
            __inject_api_timing_summary_test(test)


def __inject_api_timing_summary_across_tests(environment):
    api_endpoints = {}
    internals = {}

    for test in environment['raw_data']['tests']:
        if 'metadata' not in test:
            continue

        test_endpoints = test['api_endpoint_metrics']
        for api_endpoint, endpoint_metrics in test_endpoints.items():
            _ensure_has_dict_at_key(api_endpoints, api_endpoint,
                total_time={'raw': []},
                sql_time={'raw': []},
                sql_queries={'raw': []},
            )
            api_endpoints[api_endpoint]['label'] = endpoint_metrics['label']
            api_endpoints[api_endpoint]['total_time']['raw'].extend(endpoint_metrics['total_time']['raw'])
            api_endpoints[api_endpoint]['sql_time']['raw'].extend(endpoint_metrics['sql_time']['raw'])
            api_endpoints[api_endpoint]['sql_queries']['raw'].extend(endpoint_metrics['sql_queries']['raw'])

        test_endpoints = test['internals_metrics']
        for api_endpoint, endpoint_metrics in test_endpoints.items():
            _ensure_has_dict_at_key(internals, api_endpoint,
                total_time={'raw': []},
            )
            internals[api_endpoint]['label'] = endpoint_metrics['label']
            internals[api_endpoint]['total_time']['raw'].extend(endpoint_metrics['total_time']['raw'])

    for endpoint_metrics in api_endpoints.values():
        __inject_statistics(endpoint_metrics['total_time'])
        __inject_statistics(endpoint_metrics['sql_time'])
        __inject_statistics(endpoint_metrics['sql_queries'])

    for endpoint_metrics in internals.values():
        __inject_statistics(endpoint_metrics['total_time'])

    environment['raw_data']['api_endpoint_metrics'] = api_endpoints
    environment['raw_data']['internals_metrics'] = internals


def __inject_raw_timings(environment):
    all_timings = []

    for test in environment['raw_data']['tests']:
        if 'metadata' not in test or 'local_metrics' not in test['metadata']:
            continue

        metrics = test['metadata']['local_metrics']
        timing = metrics['timing']

        for endpoint, timings in timing.items():
            recording = timings[0].copy()
            recording['endpoint'] = endpoint
            all_timings.append(recording)

    environment['raw_data']['all_timings'] = all_timings


def _ensure_has_dict_at_key(the_dict, key, **kwd):
    if key not in the_dict:
        the_dict[key] = dict(**kwd)


def __inject_api_timing_summary_test(test):
    metrics = test['metadata']['local_metrics']
    timing = metrics['timing']
    counter = metrics['counter']
    api_endpoints = {}
    internal_timings = {}

    def summarize_times(timings):
        times = list(map(lambda t: t['time'], timings))
        return __inject_statistics({
            'raw': times,
        })

    def summarize_counter(c):
        counters = list(map(lambda t: t['n'], c))
        return __inject_statistics({
            'raw': counters,
        })

    for endpoint, timings in timing.items():
        if not endpoint.startswith("api"):
            continue

        endpoint_summary = {
            'total_time': summarize_times(timings),
            'label': endpoint[len("api."):]
        }
        sql_times = "sql.%s" % endpoint
        if sql_times in timing:
            endpoint_summary['sql_time'] = summarize_times(timing[sql_times])
        sql_queries = "sqlqueries.%s" % endpoint
        if sql_queries in counter:
            endpoint_summary['sql_queries'] = summarize_counter(counter[sql_queries])
        api_endpoints[endpoint] = endpoint_summary

    for endpoint, timings in timing.items():
        if not endpoint.startswith("internals"):
            continue

        internal_summary = {
            'total_time': summarize_times(timings),
            'label': endpoint[len("internals."):]
        }
        internal_timings[endpoint] = internal_summary

    test['api_endpoint_metrics'] = api_endpoints
    test['internals_metrics'] = internal_timings


def __inject_statistics(from_dict):
    raw_values = from_dict['raw']
    from_dict['sum'] = sum(raw_values)
    from_dict['median'] = f"{statistics.median(raw_values):.2f}"
    if len(raw_values) > 1:
        from_dict['stdev'] = f"{statistics.stdev(raw_values):.4f}"
    else:
        from_dict['stdev'] = "n/a"
    from_dict['mean'] = f"{statistics.mean(raw_values):.2f}"
    from_dict['count'] = len(raw_values)
    return from_dict


def __inject_summary(environment):
    total = 0
    failures = 0
    skips = 0
    for test in environment['raw_data']['tests']:
        total += 1
        status = test.get('outcome')
        if status == 'failed':
            failures += 1
        elif status == 'skipped':
            skips += 1
    environment['raw_data']['results'] = {
        'total': total,
        'failures': failures,
        'skips': skips,
    }


def _parser():
    parser = argparse.ArgumentParser(description=DESCRIPTION)
    parser.add_argument('input_path', metavar='INPUT', type=str, nargs="+",
                        help='structured input path (.json)')
    parser.add_argument('--output_path', type=str, default="test.html",
                        help='output path (.html)')
    parser.add_argument('--title', type=str, default="Test Performance Summary",
                        help='Performance Test Results')
    parser.add_argument('--include_raw_metrics', action="store_true", default=False)

    return parser


if __name__ == "__main__":
    main()