audit-labs/audit-tools

 1"""Collect CI/CD pipeline history for every project in the group."""
 2
 3import sys
 4
 5import requests
 6
 7from .api import paginate
 8
 9
10def pipelines(_group, cfg, projects):
11    rows = []
12    for p in projects:
13        try:
14            project_pipelines = paginate(
15                f"{cfg['base_url']}/projects/{p['id']}/pipelines", cfg
16            )
17        except requests.HTTPError as e:
18            if e.response is not None and e.response.status_code in (403, 404):
19                print(
20                    f"  Skipping {p.get('path_with_namespace', p['id'])}: "
21                    f"pipelines returned {e.response.status_code}",
22                    file=sys.stderr,
23                )
24                continue
25            raise
26        for pipe in project_pipelines:
27            rows.append(
28                {
29                    "project": p.get("path_with_namespace", ""),
30                    "pipeline_id": pipe.get("id"),
31                    "status": pipe.get("status", ""),
32                    "ref": pipe.get("ref", ""),
33                    "source": pipe.get("source", ""),
34                    "created_at": pipe.get("created_at", ""),
35                    "web_url": pipe.get("web_url", ""),
36                }
37            )
38    return rows