(self, state, context, parent: Build)
| 56 | domain_models = [devops.CICDTask] |
| 57 | |
| 58 | def collect(self, state, context, parent: Build) -> Iterable[tuple[object, dict]]: |
| 59 | repo: GitRepository = context.scope |
| 60 | api = AzureDevOpsAPI(context.connection) |
| 61 | try: |
| 62 | response = api.jobs(repo.org_id, repo.project_id, parent.id) |
| 63 | except APIException as e: |
| 64 | # Asking for the timeline of a deleted build returns a 204. |
| 65 | # But a "deleted" build may be "cleaned" (i.e. deleted for real) |
| 66 | # after some time. In this case, the timeline endpoint returns a |
| 67 | # 404 instead. |
| 68 | if e.response.status == HTTPStatus.NOT_FOUND: |
| 69 | return |
| 70 | raise |
| 71 | # If a build has failed before any jobs have started, e.g. due to a |
| 72 | # bad YAML file, then the jobs endpoint will return a 204 NO CONTENT. |
| 73 | if response.status == HTTPStatus.NO_CONTENT: |
| 74 | return |
| 75 | for raw_job in response.json["records"]: |
| 76 | # Collect both Job and Stage records to support environment detection from stages |
| 77 | if raw_job["type"] in ("Job", "Stage"): |
| 78 | raw_job["build_id"] = parent.domain_id() |
| 79 | raw_job["x_request_url"] = response.get_url_with_query_string() |
| 80 | raw_job["x_request_input"] = { |
| 81 | "OrgId": repo.org_id, |
| 82 | "ProjectId": repo.project_id, |
| 83 | "BuildId": parent.id, |
| 84 | } |
| 85 | yield raw_job, state |
| 86 | |
| 87 | def convert(self, j: Job, ctx: Context) -> Iterable[devops.CICDPipeline]: |
| 88 | if not j.start_time: |
nothing calls this directly
no test coverage detected