Merge branch 'ci/print_retry_job_message_optionally' into 'master'

ci: add condition to print retry job message in dynamic pipeline report if any job has failed

See merge request espressif/esp-idf!33211
This commit is contained in:
Aleksei Apaseev 2024-09-03 18:35:22 +08:00
commit 4f399061e7
2 changed files with 46 additions and 24 deletions

View File

@ -230,10 +230,9 @@ class ReportGenerator:
def _get_report_str(self) -> str: def _get_report_str(self) -> str:
raise NotImplementedError raise NotImplementedError
def post_report(self, print_report_path: bool = True) -> None: def _generate_comment(self, print_report_path: bool) -> str:
# report in html format, otherwise will exceed the limit # Report in HTML format to avoid exceeding length limits
comment = f'#### {self.title}\n' comment = f'#### {self.title}\n'
report_str = self._get_report_str() report_str = self._get_report_str()
if self.additional_info: if self.additional_info:
@ -241,33 +240,56 @@ class ReportGenerator:
report_url_path = self.write_report_to_file(report_str, self.job_id, self.output_filepath) report_url_path = self.write_report_to_file(report_str, self.job_id, self.output_filepath)
if print_report_path and report_url_path: if print_report_path and report_url_path:
comment += dedent(f""" comment += dedent(
Full {self.title} here: {report_url_path} (with commit {self.commit_id[:8]} f"""
Full {self.title} here: {report_url_path} (with commit {self.commit_id[:8]})
"""
)
return comment
def _update_mr_comment(self, comment: str, print_retry_jobs_message: bool) -> None:
retry_job_picture_comment = (f'{RETRY_JOB_TITLE}\n\n' f'{RETRY_JOB_PICTURE_LINK}').format(
pic_url=get_repository_file_url(RETRY_JOB_PICTURE_PATH)
)
del_retry_job_pic_pattern = re.escape(RETRY_JOB_TITLE) + r'.*?' + re.escape(f'{RETRY_JOB_PICTURE_PATH})')
for note in self.mr.notes.list(iterator=True):
if note.body.startswith(COMMENT_START_MARKER):
updated_str = self._get_updated_comment(note.body, comment)
# Add retry job message only if any job has failed
if print_retry_jobs_message:
updated_str = re.sub(del_retry_job_pic_pattern, '', updated_str, flags=re.DOTALL)
updated_str += retry_job_picture_comment
note.body = updated_str
note.save()
break
else:
# Create a new comment if no existing comment is found
new_comment = f'{COMMENT_START_MARKER}\n\n{comment}'
if print_retry_jobs_message:
new_comment += retry_job_picture_comment
self.mr.notes.create({'body': new_comment})
def _get_updated_comment(self, existing_comment: str, new_comment: str) -> str:
updated_str = re.sub(self.REGEX_PATTERN.format(self.title), new_comment, existing_comment)
if updated_str == existing_comment:
updated_str = f'{existing_comment.strip()}\n\n{new_comment}'
return updated_str
def post_report(self, print_report_path: bool = True, print_retry_jobs_message: bool = False) -> None:
comment = self._generate_comment(print_report_path)
""")
print(comment) print(comment)
if self.mr is None: if self.mr is None:
print('No MR found, skip posting comment') print('No MR found, skip posting comment')
return return
retry_job_picture_comment = (f'{RETRY_JOB_TITLE}\n\n'
f'{RETRY_JOB_PICTURE_LINK}').format(pic_url=get_repository_file_url(RETRY_JOB_PICTURE_PATH))
del_retry_job_pic_pattern = re.escape(RETRY_JOB_TITLE) + r'.*?' + re.escape(f'{RETRY_JOB_PICTURE_PATH})')
for note in self.mr.notes.list(iterator=True):
if note.body.startswith(COMMENT_START_MARKER):
updated_str = re.sub(self.REGEX_PATTERN.format(self.title), comment, note.body)
if updated_str == note.body: # not updated
updated_str = f'{note.body.strip()}\n\n{comment}'
updated_str = re.sub(del_retry_job_pic_pattern, '', updated_str, flags=re.DOTALL) self._update_mr_comment(comment, print_retry_jobs_message=print_retry_jobs_message)
note.body = updated_str + retry_job_picture_comment
note.save()
break
else:
new_comment = f"""{COMMENT_START_MARKER}
{comment}{retry_job_picture_comment}"""
self.mr.notes.create({'body': new_comment})
class BuildReportGenerator(ReportGenerator): class BuildReportGenerator(ReportGenerator):

View File

@ -94,7 +94,7 @@ def generate_jobs_report(args: argparse.Namespace) -> None:
return return
report_generator = JobReportGenerator(args.project_id, args.mr_iid, args.pipeline_id, args.job_id, args.commit_id, jobs=jobs) report_generator = JobReportGenerator(args.project_id, args.mr_iid, args.pipeline_id, args.job_id, args.commit_id, jobs=jobs)
report_generator.post_report(print_report_path=False) report_generator.post_report(print_report_path=False, print_retry_jobs_message=any(job.is_failed for job in jobs))
if __name__ == '__main__': if __name__ == '__main__':