ci: add dut_log_url column to failed testcases report

Introduced changes:
- add xml attribute "dut_log_url" to pytest report
- add column "dut_log_url" to failed testcases table of dynamic pipeline report
- make the table header sticky
- add permalinks to the Table Titles
- split target test report by testcase type for better clarity
- fix the logic of finding the testcases failed on cur branch / other branches
This commit is contained in:
Aleksei Apaseev
2024-07-02 17:38:41 +08:00
parent 7d3ac1abe4
commit cd59d96ff4
12 changed files with 574 additions and 245 deletions

View File

@@ -1,11 +1,13 @@
# SPDX-FileCopyrightText: 2024 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
import abc
import copy
import fnmatch
import html
import os
import re
import typing as t
from textwrap import dedent
import yaml
from artifacts_handler import ArtifactType
@@ -21,20 +23,24 @@ from .constants import TEST_RELATED_APPS_DOWNLOAD_URLS_FILENAME
from .models import GitlabJob
from .models import TestCase
from .utils import fetch_failed_testcases_failure_ratio
from .utils import format_permalink
from .utils import get_report_url
from .utils import is_url
from .utils import load_known_failure_cases
class ReportGenerator:
REGEX_PATTERN = '#### {}[^####]+'
REGEX_PATTERN = r'#### {}\n[\s\S]*?(?=\n#### |$)'
def __init__(self, project_id: int, mr_iid: int, pipeline_id: int, *, title: str):
def __init__(self, project_id: int, mr_iid: int, pipeline_id: int, job_id: int, commit_id: str, *, title: str):
gl_project = Gitlab(project_id).project
if mr_iid is not None:
self.mr = gl_project.mergerequests.get(mr_iid)
else:
self.mr = None
self.pipeline_id = pipeline_id
self.job_id = job_id
self.commit_id = commit_id
self.title = title
self.output_filepath = self.title.lower().replace(' ', '_') + '.html'
@@ -47,10 +53,30 @@ class ReportGenerator:
return ''
def write_report_to_file(self, report_str: str, job_id: int, output_filepath: str) -> t.Optional[str]:
"""
Writes the report to a file and constructs a modified URL based on environment settings.
:param report_str: The report content to be written to the file.
:param job_id: The job identifier used to construct the URL.
:param output_filepath: The path to the output file.
:return: The modified URL pointing to the job's artifacts.
"""
if not report_str:
return None
with open(output_filepath, 'w') as file:
file.write(report_str)
# for example, {URL}/-/esp-idf/-/jobs/{id}/artifacts/list_job_84.txt
# CI_PAGES_URL is {URL}/esp-idf, which missed one `-`
report_url: str = get_report_url(job_id, output_filepath)
return report_url
def generate_html_report(self, table_str: str) -> str:
# we're using bootstrap table
table_str = table_str.replace('<table>', '<table data-toggle="table" data-search="true">')
table_str = table_str.replace(
'<table>', '<table data-toggle="table" data-search="true" data-sticky-header="true">'
)
with open(REPORT_TEMPLATE_FILEPATH) as fr:
template = fr.read()
@@ -62,19 +88,16 @@ class ReportGenerator:
def create_table_section(
self,
report_sections: list,
title: str,
items: list,
headers: list,
row_attrs: list,
value_functions: t.Optional[list] = None,
) -> None:
) -> t.List:
"""
Appends a formatted section to a report based on the provided items. This section includes
a header and a table constructed from the items list with specified headers and attributes.
:param report_sections: List where the HTML report sections are collected. This list is
modified in-place by appending new sections.
:param title: Title for the report section. This title is used as a header above the table.
:param items: List of item objects to include in the table. Each item should have attributes
that correspond to the row_attrs and value_functions specified.
@@ -86,17 +109,34 @@ class ReportGenerator:
a function that takes an item and returns a string. This is used for
generating dynamic columns based on item data.
:return: None. The function modifies the 'report_sections' list by appending new HTML sections.
:return: List with appended HTML sections.
"""
if not items:
return
return []
report_sections.append(f'<h2>{title}</h2>')
report_sections.append(
report_sections = [
f"""<h2 id="{format_permalink(title)}">{title}<i class="fas fa-link copy-link-icon"
onclick="copyPermalink('#{format_permalink(title)}')"></i></h2>""",
self._create_table_for_items(
items=items, headers=headers, row_attrs=row_attrs, value_functions=value_functions or []
)
)
),
]
return report_sections
@staticmethod
def generate_additional_info_section(title: str, count: int, report_url: t.Optional[str] = None) -> str:
"""
Generate a section for the additional info string.
:param title: The title of the section.
:param count: The count of test cases.
:param report_url: The URL of the report. If count = 0, only the count will be included.
:return: The formatted additional info section string.
"""
if count != 0 and report_url:
return f'- **{title}:** [{count}]({report_url}/#{format_permalink(title)})\n'
else:
return f'- **{title}:** {count}\n'
def _create_table_for_items(
self,
@@ -185,7 +225,7 @@ class ReportGenerator:
def _get_report_str(self) -> str:
raise NotImplementedError
def post_report(self, job_id: int, commit_id: str) -> None:
def post_report(self, print_report_path: bool = True) -> None:
# report in html format, otherwise will exceed the limit
comment = f'#### {self.title}\n'
@@ -194,18 +234,12 @@ class ReportGenerator:
if self.additional_info:
comment += f'{self.additional_info}\n'
if report_str:
with open(self.output_filepath, 'w') as fw:
fw.write(report_str)
report_url_path = self.write_report_to_file(report_str, self.job_id, self.output_filepath)
if print_report_path and report_url_path:
comment += dedent(f"""
Full {self.title} here: {report_url_path} (with commit {self.commit_id[:8]}
# for example, {URL}/-/esp-idf/-/jobs/{id}/artifacts/list_job_84.txt
# CI_PAGES_URL is {URL}/esp-idf, which missed one `-`
url = os.getenv('CI_PAGES_URL', '').replace('esp-idf', '-/esp-idf')
comment += f"""
Full {self.title} here: {url}/-/jobs/{job_id}/artifacts/{self.output_filepath} (with commit {commit_id[:8]})
"""
""")
print(comment)
if self.mr is None:
@@ -234,11 +268,13 @@ class BuildReportGenerator(ReportGenerator):
project_id: int,
mr_iid: int,
pipeline_id: int,
job_id: int,
commit_id: str,
*,
title: str = 'Build Report',
apps: t.List[App],
):
super().__init__(project_id, mr_iid, pipeline_id, title=title)
super().__init__(project_id, mr_iid, pipeline_id, job_id, commit_id, title=title)
self.apps = apps
self.apps_presigned_url_filepath = TEST_RELATED_APPS_DOWNLOAD_URLS_FILENAME
@@ -365,14 +401,26 @@ class TargetTestReportGenerator(ReportGenerator):
project_id: int,
mr_iid: int,
pipeline_id: int,
job_id: int,
commit_id: str,
*,
title: str = 'Target Test Report',
test_cases: t.List[TestCase],
):
super().__init__(project_id, mr_iid, pipeline_id, title=title)
super().__init__(project_id, mr_iid, pipeline_id, job_id, commit_id, title=title)
self.test_cases = test_cases
self._known_failure_cases_set = None
self.report_titles_map = {
'failed_yours': 'Failed Test Cases on Your branch (Excludes Known Failure Cases)',
'failed_others': 'Failed Test Cases on Other branches (Excludes Known Failure Cases)',
'failed_known': 'Known Failure Cases',
'skipped': 'Skipped Test Cases',
'succeeded': 'Succeeded Test Cases',
}
self.skipped_test_cases_report_file = 'skipped_cases.html'
self.succeeded_cases_report_file = 'succeeded_cases.html'
self.failed_cases_report_file = 'failed_cases.html'
@property
def known_failure_cases_set(self) -> t.Optional[t.Set[str]]:
@@ -382,6 +430,10 @@ class TargetTestReportGenerator(ReportGenerator):
return self._known_failure_cases_set
def get_known_failure_cases(self) -> t.List[TestCase]:
"""
Retrieve the known failure test cases.
:return: A list of known failure test cases.
"""
if self.known_failure_cases_set is None:
return []
matched_cases = [
@@ -392,109 +444,187 @@ class TargetTestReportGenerator(ReportGenerator):
]
return matched_cases
@staticmethod
def filter_test_cases(
cur_branch_failures: t.List[TestCase],
other_branch_failures: t.List[TestCase],
) -> t.Tuple[t.List[TestCase], t.List[TestCase]]:
"""
Filter the test cases into current branch failures and other branch failures.
:param cur_branch_failures: List of failed test cases on the current branch.
:param other_branch_failures: List of failed test cases on other branches.
:return: A tuple containing two lists:
- failed_test_cases_cur_branch_only: Test cases that have failed only on the current branch.
- failed_test_cases_other_branch_exclude_cur_branch: Test cases that have failed on other branches
excluding the current branch.
"""
cur_branch_unique_failures = []
other_branch_failure_map = {tc.name: tc for tc in other_branch_failures}
for cur_tc in cur_branch_failures:
if cur_tc.latest_failed_count > 0 and (
cur_tc.name not in other_branch_failure_map
or other_branch_failure_map[cur_tc.name].latest_failed_count == 0
):
cur_branch_unique_failures.append(cur_tc)
uniq_fail_names = {cur_tc.name for cur_tc in cur_branch_unique_failures}
other_branch_exclusive_failures = [tc for tc in other_branch_failures if tc.name not in uniq_fail_names]
return cur_branch_unique_failures, other_branch_exclusive_failures
def get_failed_cases_report_parts(self) -> t.List[str]:
"""
Generate the report parts for failed test cases and update the additional info section.
:return: A list of strings representing the table sections for the failed test cases.
"""
known_failures = self.get_known_failure_cases()
failed_test_cases = self._filter_items(
self.test_cases, lambda tc: tc.is_failure and tc.name not in {case.name for case in known_failures}
)
failed_test_cases_cur_branch = self._sort_items(
fetch_failed_testcases_failure_ratio(
copy.deepcopy(failed_test_cases),
branches_filter={'include_branches': [os.getenv('CI_MERGE_REQUEST_SOURCE_BRANCH_NAME', '')]},
),
key='latest_failed_count',
)
failed_test_cases_other_branch = self._sort_items(
fetch_failed_testcases_failure_ratio(
copy.deepcopy(failed_test_cases),
branches_filter={'exclude_branches': [os.getenv('CI_MERGE_REQUEST_SOURCE_BRANCH_NAME', '')]},
),
key='latest_failed_count',
)
failed_test_cases_cur_branch, failed_test_cases_other_branch = self.filter_test_cases(
failed_test_cases_cur_branch, failed_test_cases_other_branch
)
cur_branch_cases_table_section = self.create_table_section(
title=self.report_titles_map['failed_yours'],
items=failed_test_cases_cur_branch,
headers=[
'Test Case',
'Test Script File Path',
'Failure Reason',
f'Failures on your branch (40 latest testcases)',
'Dut Log URL',
'Job URL',
'Grafana URL',
],
row_attrs=['name', 'file', 'failure', 'dut_log_url', 'ci_job_url', 'ci_dashboard_url'],
value_functions=[
(
'Failures on your branch (40 latest testcases)',
lambda item: f"{getattr(item, 'latest_failed_count', '')} / {getattr(item, 'latest_total_count', '')}",
)
],
)
other_branch_cases_table_section = self.create_table_section(
title=self.report_titles_map['failed_others'],
items=failed_test_cases_other_branch,
headers=[
'Test Case',
'Test Script File Path',
'Failure Reason',
'Failures across all other branches (40 latest testcases)',
'Dut Log URL',
'Job URL',
'Grafana URL',
],
row_attrs=['name', 'file', 'failure', 'dut_log_url', 'ci_job_url', 'ci_dashboard_url'],
value_functions=[
(
'Failures across all other branches (40 latest testcases)',
lambda item: f"{getattr(item, 'latest_failed_count', '')} / {getattr(item, 'latest_total_count', '')}",
)
],
)
known_failures_cases_table_section = self.create_table_section(
title=self.report_titles_map['failed_known'],
items=known_failures,
headers=['Test Case', 'Test Script File Path', 'Failure Reason', 'Job URL', 'Grafana URL'],
row_attrs=['name', 'file', 'failure', 'ci_job_url', 'ci_dashboard_url'],
)
failed_cases_report_url = self.write_report_to_file(
self.generate_html_report(
''.join(
cur_branch_cases_table_section
+ other_branch_cases_table_section
+ known_failures_cases_table_section
)
),
self.job_id,
self.failed_cases_report_file,
)
self.additional_info += self.generate_additional_info_section(
self.report_titles_map['failed_yours'], len(failed_test_cases_cur_branch), failed_cases_report_url
)
self.additional_info += self.generate_additional_info_section(
self.report_titles_map['failed_others'], len(failed_test_cases_other_branch), failed_cases_report_url
)
self.additional_info += self.generate_additional_info_section(
self.report_titles_map['failed_known'], len(known_failures), failed_cases_report_url
)
return cur_branch_cases_table_section + other_branch_cases_table_section + known_failures_cases_table_section
def get_skipped_cases_report_parts(self) -> t.List[str]:
"""
Generate the report parts for skipped test cases and update the additional info section.
:return: A list of strings representing the table sections for the skipped test cases.
"""
skipped_test_cases = self._filter_items(self.test_cases, lambda tc: tc.is_skipped)
skipped_cases_table_section = self.create_table_section(
title=self.report_titles_map['skipped'],
items=skipped_test_cases,
headers=['Test Case', 'Test Script File Path', 'Skipped Reason', 'Grafana URL'],
row_attrs=['name', 'file', 'skipped', 'ci_dashboard_url'],
)
skipped_cases_report_url = self.write_report_to_file(
self.generate_html_report(''.join(skipped_cases_table_section)),
self.job_id,
self.skipped_test_cases_report_file,
)
self.additional_info += self.generate_additional_info_section(
self.report_titles_map['skipped'], len(skipped_test_cases), skipped_cases_report_url
)
return skipped_cases_table_section
def get_succeeded_cases_report_parts(self) -> t.List[str]:
"""
Generate the report parts for succeeded test cases and update the additional info section.
:return: A list of strings representing the table sections for the succeeded test cases.
"""
succeeded_test_cases = self._filter_items(self.test_cases, lambda tc: tc.is_success)
succeeded_cases_table_section = self.create_table_section(
title=self.report_titles_map['succeeded'],
items=succeeded_test_cases,
headers=['Test Case', 'Test Script File Path', 'Job URL', 'Grafana URL'],
row_attrs=['name', 'file', 'ci_job_url', 'ci_dashboard_url'],
)
succeeded_cases_report_url = self.write_report_to_file(
self.generate_html_report(''.join(succeeded_cases_table_section)),
self.job_id,
self.succeeded_cases_report_file,
)
self.additional_info += self.generate_additional_info_section(
self.report_titles_map['succeeded'], len(succeeded_test_cases), succeeded_cases_report_url
)
self.additional_info += '\n'
return succeeded_cases_table_section
def _get_report_str(self) -> str:
"""
Generate a complete HTML report string by processing test cases.
:return: Complete HTML report string.
"""
report_parts: list = []
self.additional_info = f'**Test Case Summary (with commit {self.commit_id[:8]}):**\n'
failed_cases_report_parts = self.get_failed_cases_report_parts()
skipped_cases_report_parts = self.get_skipped_cases_report_parts()
succeeded_cases_report_parts = self.get_succeeded_cases_report_parts()
known_failures = self.get_known_failure_cases()
known_failure_case_names = {case.name for case in known_failures}
failed_test_cases = self._filter_items(
self.test_cases, lambda tc: tc.is_failure and tc.name not in known_failure_case_names
return self.generate_html_report(
''.join(failed_cases_report_parts + skipped_cases_report_parts + succeeded_cases_report_parts)
)
failed_test_cases_with_ratio = self._sort_items(
fetch_failed_testcases_failure_ratio(failed_test_cases), key='latest_failed_count'
)
skipped_test_cases = self._filter_items(self.test_cases, lambda tc: tc.is_skipped)
successful_test_cases = self._filter_items(self.test_cases, lambda tc: tc.is_success)
current_branch_failures = self._sort_items(
self._filter_items(failed_test_cases_with_ratio, lambda tc: tc.latest_failed_count == 0),
key='latest_failed_count',
)
other_branch_failures = self._sort_items(
self._filter_items(
failed_test_cases_with_ratio, lambda tc: tc.name not in [t.name for t in current_branch_failures]
),
key='latest_failed_count',
)
self.create_table_section(
report_sections=report_parts,
title='Failed Test Cases on Your branch (Excludes Known Failure Cases)',
items=current_branch_failures,
headers=[
'Test Case',
'Test Script File Path',
'Failure Reason',
'Failures across all other branches (20 latest testcases)',
'Job URL',
'Grafana URL',
],
row_attrs=['name', 'file', 'failure', 'ci_job_url', 'ci_dashboard_url'],
value_functions=[
(
'Failures across all other branches (20 latest testcases)',
lambda item: f"{getattr(item, 'latest_failed_count', '')} / {getattr(item, 'latest_total_count', '')}",
)
],
)
self.create_table_section(
report_sections=report_parts,
title='Failed Test Cases on Other branches (Excludes Known Failure Cases)',
items=other_branch_failures,
headers=[
'Test Case',
'Test Script File Path',
'Failure Reason',
'Failures across all other branches (20 latest testcases)',
'Job URL',
'Grafana URL',
],
row_attrs=['name', 'file', 'failure', 'ci_job_url', 'ci_dashboard_url'],
value_functions=[
(
'Failures across all other branches (20 latest testcases)',
lambda item: f"{getattr(item, 'latest_failed_count', '')} / {getattr(item, 'latest_total_count', '')}",
)
],
)
self.create_table_section(
report_sections=report_parts,
title='Known Failure Cases',
items=known_failures,
headers=['Test Case', 'Test Script File Path', 'Failure Reason', 'Job URL', 'Grafana URL'],
row_attrs=['name', 'file', 'failure', 'ci_job_url', 'ci_dashboard_url'],
)
self.create_table_section(
report_sections=report_parts,
title='Skipped Test Cases',
items=skipped_test_cases,
headers=['Test Case', 'Test Script File Path', 'Skipped Reason', 'Grafana URL'],
row_attrs=['name', 'file', 'skipped', 'ci_dashboard_url'],
)
self.create_table_section(
report_sections=report_parts,
title='Succeeded Test Cases',
items=successful_test_cases,
headers=['Test Case', 'Test Script File Path', 'Job URL', 'Grafana URL'],
row_attrs=['name', 'file', 'ci_job_url', 'ci_dashboard_url'],
)
self.additional_info = (
'**Test Case Summary:**\n'
f'- **Failed Test Cases on Your Branch (Excludes Known Failure Cases):** {len(current_branch_failures)}.\n'
f'- **Failed Test Cases on Other Branches (Excludes Known Failure Cases):** {len(other_branch_failures)}.\n'
f'- **Known Failures:** {len(known_failures)}\n'
f'- **Skipped Test Cases:** {len(skipped_test_cases)}\n'
f'- **Succeeded Test Cases:** {len(successful_test_cases)}\n\n'
'Please check report below for more information.\n\n'
)
return self.generate_html_report(''.join(report_parts))
class JobReportGenerator(ReportGenerator):
@@ -503,12 +633,19 @@ class JobReportGenerator(ReportGenerator):
project_id: int,
mr_iid: int,
pipeline_id: int,
job_id: int,
commit_id: str,
*,
title: str = 'Job Report',
jobs: t.List[GitlabJob],
):
super().__init__(project_id, mr_iid, pipeline_id, title=title)
super().__init__(project_id, mr_iid, pipeline_id, job_id, commit_id, title=title)
self.jobs = jobs
self.report_titles_map = {
'failed_jobs': 'Failed Jobs (Excludes "integration_test" and "target_test" jobs)',
'succeeded': 'Succeeded Jobs',
}
self.failed_jobs_report_file = 'job_report.html'
def _get_report_str(self) -> str:
"""
@@ -516,7 +653,6 @@ class JobReportGenerator(ReportGenerator):
:return: Complete HTML report string.
"""
report_str: str = ''
report_parts: list = []
if not self.jobs:
print('No jobs found, skip generating job report')
@@ -530,34 +666,41 @@ class JobReportGenerator(ReportGenerator):
)
succeeded_jobs = self._filter_items(self.jobs, lambda job: job.is_success)
self.additional_info = (
'**Job Summary:**\n'
f'- **Failed Jobs (Excludes "integration_test" and "target_test" jobs):** {len(relevant_failed_jobs)}\n'
f'- **Succeeded Jobs:** {len(succeeded_jobs)}\n\n'
self.additional_info = f'**Job Summary (with commit {self.commit_id[:8]}):**\n'
self.additional_info += self.generate_additional_info_section(
self.report_titles_map['succeeded'], len(succeeded_jobs)
)
if relevant_failed_jobs:
self.create_table_section(
report_sections=report_parts,
title='Failed Jobs (Excludes "integration_test" and "target_test" jobs)',
items=relevant_failed_jobs,
headers=[
'Job Name',
'Failure Reason',
'Failure Log',
'Failures across all other branches (10 latest jobs)',
'URL',
'CI Dashboard URL',
],
row_attrs=['name', 'failure_reason', 'failure_log', 'url', 'ci_dashboard_url'],
value_functions=[
(
'Failures across all other branches (10 latest jobs)',
lambda item: f"{getattr(item, 'latest_failed_count', '')} / {getattr(item, 'latest_total_count', '')}",
)
],
if not relevant_failed_jobs:
self.additional_info += self.generate_additional_info_section(
self.report_titles_map['failed_jobs'], len(relevant_failed_jobs)
)
self.additional_info += f'Please check report below for more information.\n\n'
report_str = self.generate_html_report(''.join(report_parts))
return report_str
report_sections = self.create_table_section(
title='Failed Jobs (Excludes "integration_test" and "target_test" jobs)',
items=relevant_failed_jobs,
headers=[
'Job Name',
'Failure Reason',
'Failure Log',
'Failures across all other branches (10 latest jobs)',
'URL',
'CI Dashboard URL',
],
row_attrs=['name', 'failure_reason', 'failure_log', 'url', 'ci_dashboard_url'],
value_functions=[
(
'Failures across all other branches (10 latest jobs)',
lambda item: f"{getattr(item, 'latest_failed_count', '')} / {getattr(item, 'latest_total_count', '')}",
)
],
)
relevant_failed_jobs_report_url = get_report_url(self.job_id, self.failed_jobs_report_file)
self.additional_info += self.generate_additional_info_section(
self.report_titles_map['failed_jobs'], len(relevant_failed_jobs), relevant_failed_jobs_report_url
)
report_str = self.generate_html_report(''.join(report_sections))
return report_str