Jenkins failed tests statistics

failed-tests.py

#!/bin/env python
#
# Usage: failed_tests.py [BRANCH] [NUMBER_OF_BUILDS]
#
# Default:
#       BRANCH              - develop
#       NUMBER_OF_BUILDS    - 10
#

import collections
import getpass
import os
import requests
import sys

USER_ENV = 'JUSER'
PASS_ENV = 'JPASS'

FAILED_STATUS = ('FAILED', 'REGRESSION')

def create_session():
    session = requests.Session()

    if USER_ENV in os.environ:
        user = os.environ[USER_ENV]
    else:
        user = input('Username: ')

    if PASS_ENV in os.environ:
        password = os.environ[PASS_ENV]
    else:
        password = getpass.getpass()

    session.auth = (user, password)

    return session

def get_branch():
    return sys.argv[1] if len(sys.argv) > 1 else 'develop'

def get_job_url(branch):
    return f'https://ci.zeebe.camunda.cloud/job/zeebe-io/job/zeebe/job/{branch}'

def get_last_build_number(session, job_url):
    response = session.get(f'{job_url}/api/json').json()
    return int(response['lastCompletedBuild']['number'])

def get_build_numbers(session, job_url):
    number_of_builds = int(sys.argv[2]) if len(sys.argv) > 2 else 10
    last_build = get_last_build_number(session, job_url)
    first_build = max(1, last_build - number_of_builds + 1)
    return (first_build, last_build)

def check_job(session, job_url, first_build, last_build):
    failed_tests = collections.defaultdict(list)
    for build_id in range(first_build, last_build + 1):
        check_build(session, job_url, build_id, failed_tests)
    return failed_tests

def check_build(session, job_url, build_id, errors):
    try:
        response = session.get(f'{job_url}/{build_id}/testReport/api/json').json()
        print(f"Checking build {build_id} (passed: {response['passCount']}, failed: {response['failCount']}, skipped: {response['skipCount']})")
        if response['failCount'] > 0:
            for suite in response['suites']:
                for case in suite['cases']:
                    if case['status'] in FAILED_STATUS:
                        errors[f"{case['className']}#{case['name']}"].append(build_id)
    except:
        pass # skip non existing builds

if __name__ == '__main__':
    session = create_session()
    branch = get_branch()
    job_url = get_job_url(branch)
    (first_build, last_build) = get_build_numbers(session, job_url)

    print(f'Checking branch {branch} builds {first_build} to {last_build} for failed tests')
    failed_tests = check_job(session, job_url, first_build, last_build)

    count_label = "Count"
    count_width = max(max([len(str(len(failed_test[1]))) for failed_test in failed_tests.items()]), len(count_label))
    name_label = "Test Name"
    name_width = max([len(test_name) for test_name in failed_tests])
    builds_label = "Failed Builds"
    print_row = lambda count, name, builds: print(f'{count:{count_width}} | {name:{name_width}} | {builds}')

    print()
    print_row(count_label, name_label, builds_label)

    for (test_name, build_ids) in sorted(failed_tests.items(), key=lambda test: len(test[1]), reverse=True):
        print_row(len(build_ids), test_name, build_ids)