| #!/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"]}") |