1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
|
#!/usr/bin/env python3
#
# A script to analyse failures in the gitlab pipelines. It requires an
# API key from gitlab with the following permissions:
# - api
# - read_repository
# - read_user
#
import argparse
import gitlab
import os
#
# Arguments
#
class NoneForEmptyStringAction(argparse.Action):
def __call__(self, parser, namespace, value, option_string=None):
if value == '':
setattr(namespace, self.dest, None)
else:
setattr(namespace, self.dest, value)
parser = argparse.ArgumentParser(description="Analyse failed GitLab CI runs.")
parser.add_argument("--gitlab",
default="https://gitlab.com",
help="GitLab instance URL (default: https://gitlab.com).")
parser.add_argument("--id", default=11167699,
type=int,
help="GitLab project id (default: 11167699 for qemu-project/qemu)")
parser.add_argument("--token",
default=os.getenv("GITLAB_TOKEN"),
help="Your personal access token with 'api' scope.")
parser.add_argument("--branch",
type=str,
default="staging",
action=NoneForEmptyStringAction,
help="The name of the branch (default: 'staging')")
parser.add_argument("--status",
type=str,
action=NoneForEmptyStringAction,
default="failed",
help="Filter by branch status (default: 'failed')")
parser.add_argument("--count", type=int,
default=3,
help="The number of failed runs to fetch.")
parser.add_argument("--skip-jobs",
default=False,
action='store_true',
help="Skip dumping the job info")
parser.add_argument("--pipeline", type=int,
nargs="+",
default=None,
help="Explicit pipeline ID(s) to fetch.")
if __name__ == "__main__":
args = parser.parse_args()
gl = gitlab.Gitlab(url=args.gitlab, private_token=args.token)
project = gl.projects.get(args.id)
pipelines_to_process = []
# Use explicit pipeline IDs if provided, otherwise fetch a list
if args.pipeline:
args.count = len(args.pipeline)
for p_id in args.pipeline:
pipelines_to_process.append(project.pipelines.get(p_id))
else:
# Use an iterator to fetch the pipelines
pipe_iter = project.pipelines.list(iterator=True,
status=args.status,
ref=args.branch)
# Check each failed pipeline
pipelines_to_process = [next(pipe_iter) for _ in range(args.count)]
# Check each pipeline
for p in pipelines_to_process:
jobs = p.jobs.list(get_all=True)
failed_jobs = [j for j in jobs if j.status == "failed"]
skipped_jobs = [j for j in jobs if j.status == "skipped"]
manual_jobs = [j for j in jobs if j.status == "manual"]
trs = p.test_report_summary.get()
total = trs.total["count"]
skipped = trs.total["skipped"]
failed = trs.total["failed"]
print(f"{p.status} pipeline {p.id}, total jobs {len(jobs)}, "
f"skipped {len(skipped_jobs)}, "
f"failed {len(failed_jobs)}, ",
f"{total} tests, "
f"{skipped} skipped tests, "
f"{failed} failed tests")
if not args.skip_jobs:
for j in failed_jobs:
print(f" Failed job {j.id}, {j.name}, {j.web_url}")
# It seems we can only extract failing tests from the full
# test report, maybe there is some way to filter it.
if failed > 0:
ftr = p.test_report.get()
failed_suites = [s for s in ftr.test_suites if
s["failed_count"] > 0]
for fs in failed_suites:
name = fs["name"]
tests = fs["test_cases"]
failed_tests = [t for t in tests if t["status"] == 'failed']
for t in failed_tests:
print(f" Failed test {t["classname"]}, {name}, {t["name"]}")
|
