revcache-job-summary

#!/usr/bin/env python3

import json
import os
import requests
import sys
import yaml

from argparse import ArgumentParser
from collections import defaultdict


class RevcacheResults():
    resultmap = {
        "pass": ".",
        "fail": "F",
        "skip": "s"
    }

    def __init__(self, name):
        """Class to hold results history data from revcache
        :param name:
            String identifier for the data, usually the job name
        """
        self.data = defaultdict(list)
        self.new_fails_or_skips = False
        self.last_fails = 0
        self.last_passes = 0
        self.last_skips = 0
        self.last_total = 0
        revcache_server = os.environ.get(
            "REVCACHE_SERVER", "https://testflinger-revcache.canonical.com")
        revcache_url = "{}/v1/results/{}".format(revcache_server, name)
        try:
            req = requests.get(revcache_url)
            run_list = json.loads(req.text)
        except requests.exceptions.ConnectionError as exc:
            print(exc)
            run_list = []
        except json.decoder.JSONDecodeError:
            print('Unexpected response from revcache server:')
            print(req.text)
        # Create a dict to hold all results
        # For each individual result ID, this will hold a list of prior results
        #     ex. {"example/testname": ["pass", "fail", "skip", "pass"]}
        if run_list:
            for results_list in run_list:
                results_list = json.loads(results_list).get('results', [])
                fails = 0
                passes = 0
                skips = 0
                total = len(results_list)
                for result in results_list:
                    id = result.get('id')
                    if not id:
                        continue
                    status = result.get('status')
                    if status == 'fail':
                        fails += 1
                    elif status == 'pass':
                        passes += 1
                    elif status == 'skip':
                        skips += 1
                    self.data[id].append(status)
            self.last_fails = fails
            self.last_passes = passes
            self.last_skips = skips
            self.last_total = total

    def get_summary(self, result_id):
        """Get a string that summarizes the list of previous results_all
        :param result_id:
            The name (id) of the result to get results on
        :return:
            String representation of the previous results ex: FF.s..FFFF
        """
        summary = ""
        for result in self.data.get(result_id):
            summary += self.resultmap.get(result)
        return summary

    def get_unique_summary(self, result_id, status):
        """Get a summary of the prior results, only if it was ever different
        :param result_id:
            The name (id) of the result to get results on
        :param status:
            The current test status to compare it to (pass, fail, skip)
        :return:
            String representation of the previous results ex: FF.s..FFFF
            or "" if all previous results are the same as status
        """
        result_history = self.data.get(result_id, [])
        if not result_history:
            return self.resultmap.get(status)
        for old_result in result_history:
            if old_result != status:
                return self.get_summary(result_id) + self.resultmap.get(status)
        return ""

    def get_unique_results(self, new_results, status):
        """Generate a report of prior results, only if it was ever different
        :param result_id:
            The name (id) of the result to get results on
        :param status:
            The current test status to compare it to (pass, fail, skip)
        :return:
            String with the prior results history and the name of the test
            if it was ever different from the current result, or "" if all
            previous results are the same as status
        """
        report = ""
        for result_id in new_results:
            summary = self.get_unique_summary(result_id, status)
            if summary:
                report += "[{}] {}\n".format(summary.rjust(11), result_id)
                # Detect if there are new failed or skipped tests
                if status == 'skip':
                    # Fewer than 2 means this is the first time we've seen
                    # this test and it skipped, so mark it as new
                    if len(summary) < 2 or summary[-1] != summary[-2]:
                        self.new_fails_or_skips = True
                if status == 'fail':
                    # For fails, always declare new failures unless we have
                    # A full history of 10 previous runs, and all failed
                    if summary != 'F'*11:
                        self.new_fails_or_skips = True
        return report


def get_test_fail_hints(fail_list, known_fails):
    if not known_fails:
        # Only generate the detailed list if known_fails data is provided
        return fail_list
    detailed_fail_list = list()
    for failed_test in fail_list:
        detailed_fail_list.append('[{}]'.format(failed_test))
        reason = known_fails.get(failed_test, 'Unknown reason for failure')
        detailed_fail_list.append('  - {}'.format(reason))
        detailed_fail_list.append('')
    return detailed_fail_list


def main():
    parser = ArgumentParser()
    parser.add_argument('-f', '--faildata',
                        help='YAML data file for known failures')
    parser.add_argument('result_file', help='JSON file with results')
    parser.add_argument('revcache_key',
                        help='Key or job name to search for in revcache')
    args = parser.parse_args()

    if not os.path.exists(args.result_file):
        print('No results file found!')
        print('Usually this means that the run failed to complete. ',
              'Check the output in the jenkins job for more details')
        return 0
    history = RevcacheResults(args.revcache_key)

    with open(args.result_file) as result_file:
        result_data = json.load(result_file)

    try:
        with open(args.faildata) as f:
            fail_hints = yaml.safe_load(f)
    except Exception:
        # If anything goes wrong, it's better to return a a summary without
        # details than nothing at all
        fail_hints = dict()

    results = result_data.get('results')
    fails = [x.get('id') for x in results if x.get('status') == 'fail']
    passes = [x.get('id') for x in results if x.get('status') == 'pass']
    skips = [x.get('id') for x in results if x.get('status') == 'skip']
    total = len(results)

    if os.path.exists('c3link'):
        print('\n')
        with open('c3link') as c3link:
            print('Full results at: {}'.format(c3link.read()))

    print('Summary')
    print('-------')
    print('pass:\t{}\t\t({:+d})'.format(
        len(passes), len(passes)-history.last_passes))
    print('fail:\t{}\t\t({:+d})'.format(
        len(fails), len(fails)-history.last_fails))
    print('skip:\t{}\t\t({:+d})'.format(
        len(skips), len(skips)-history.last_skips))
    print('total:\t{}\t\t({:+d})'.format(total, total-history.last_total))
    print('\n')

    if len(fails) == 0:
        print('All tests passed!')
    else:
        fails_details = get_test_fail_hints(fails, fail_hints)
        print('Failed tests')
        print('------------')
        print('\n'.join(fails_details))
    print()
    if total <= 1:
        print("WARNING: Very small number of total tests!")
    print()
    print('-' * 80)
    print()

    report = history.get_unique_results(fails, "fail")
    if report:
        print('Unstable failed tests:')
        print(report)
    report = history.get_unique_results(skips, "skip")
    if report:
        print('Unstable skipped tests:')
        print(report)
    report = history.get_unique_results(passes, "pass")
    if report:
        print('Unstable passing tests:')
        print(report)

    if not history.new_fails_or_skips:
        print('No new failed or skipped tests')


if __name__ == "__main__":
    sys.exit(main())