Jenkins blueocean api average test package time

api.py

import requests

jenkins_url = ""
pipeline_name = ""
username = ""
token = ""
base_url = f"{ jenkins_url }/blue/rest/organizations/jenkins/pipelines/{ pipeline_name }/branches/"

auth = (username, token)

def get_branches():
    branches_url = base_url
    response = requests.get(branches_url, auth=auth)
    if response.status_code == 200:
        branches = response.json()
        return branches
    else:
        print(f"Error to get branches: status { response.status_code } - { response.text }")
        return []

def get_runs(branch, only_success=False):
    runs_url = f"{ base_url }/{ branch }/runs/"
    response = requests.get(runs_url, auth=auth)
    if response.status_code == 200:
        runs = response.json()
        return [run for run in runs if run['result'] == 'SUCCESS'] if only_success else runs
    else:
        print(f"Error to get runs: status { response.status_code } - { response.text }")
        return []

def get_packages(branch, run, only_success=False):
    nodes_url = f"{ base_url }/{ branch }/runs/{ run }/nodes/"
    response = requests.get(nodes_url, auth=auth)
    if response.status_code == 200:
        nodes = response.json()
        return [node for node in nodes if node['result'] == 'SUCCESS'] if only_success else nodes
    else:
        print(f"Error to get packages: status { response.status_code } - { response.text }")
        return []
    
def get_steps(branch, run, package, only_success=False):
    steps_url = f"{ base_url }/{ branch }/runs/{ run }/nodes/{ package }/steps/"
    response = requests.get(steps_url, auth=auth)
    if response.status_code == 200:
        steps = response.json()
        return [step for step in steps if step['result'] == 'SUCCESS'] if only_success else steps
    else:
        print(f"Error to get steps: status { response.status_code } - { response.text }")
        return []

get_packages_run.py

import sys
import api

tests = {}

def main():
    branches = []
    if sys.argv[1:]:
        branches = [{ 'name': sys.argv[1] }]

    if len(branches) == 0:
        branches = api.get_branches()
    
    if not branches:
        return
    
    for branch in branches:
        branch_name = branch['name']
        runs = []
        if sys.argv[2:]:
            runs = [{ 'id': sys.argv[2] }]
        if len(runs) == 0:
            runs = api.get_runs(branch_name, only_success=True)
        for run in runs:
            run_id = run['id']
            packages = api.get_packages(branch_name, run_id, only_success=True)
            for package in packages:
                package_name = package['displayName']
                if package_name not in tests: 
                    tests[package_name] = []
                if package['durationInMillis'] > 7200000: continue
                steps = api.get_steps(branch_name, run_id, package['id'], only_success=True)
                time = 0
                for step in steps:
                    time += step['durationInMillis']
                tests[package_name] += [time]

    averages = {}
    for k in tests.keys():
        averages[k] = sum(tests[k]) / len(tests[k])
        
    for k, v in sorted(averages.items(), key=lambda item: item[1], reverse=True):
        print(f"{ k } - { int(v / (1000*60)) }min")

if __name__ == "__main__":
    main()

requirements.txt

requests

Run:

python3 get_packages_run.py {branch_name:optional} {run_id:optional}