| #!/usr/bin/env vpython3 |
| # Copyright 2026 The Chromium Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| """Fetches tryjob build statuses, per-test failure logs, text diffs, and |
| screenshot artifacts (actual_image, expected_image, image_diff) for a Gerrit CL. |
| |
| Reuses ResultDB and Buildbucket helpers from scripts/tools/update_goldens.py. |
| """ |
| |
| import argparse |
| from concurrent.futures import ThreadPoolExecutor |
| from html.parser import HTMLParser |
| import json |
| import os |
| import re |
| import subprocess |
| import sys |
| |
| TOOLS_DIR = os.path.dirname(os.path.abspath(__file__)) |
| if TOOLS_DIR not in sys.path: |
| sys.path.insert(0, TOOLS_DIR) |
| |
| import update_goldens # pylint: disable=wrong-import-position |
| |
| IMAGE_ARTIFACT_IDS = { |
| 'actual_image', |
| 'expected_image', |
| 'image_diff', |
| 'generated', |
| } |
| |
| TEXT_ARTIFACT_IDS = { |
| 'stderr', |
| 'stdout', |
| 'crash_log', |
| 'command', |
| 'snippet', |
| 'actual_text', |
| 'expected_text', |
| 'text_diff', |
| 'pretty_text_diff', |
| } |
| |
| |
| class _HTMLStripper(HTMLParser): |
| """Strips HTML tags from ResultDB summaryHtml while preserving line breaks.""" |
| |
| def __init__(self): |
| super().__init__() |
| self._parts = [] |
| |
| def handle_starttag(self, tag, attrs): |
| if tag in ('br', 'p', 'div', 'pre', 'li', 'tr'): |
| self._parts.append('\n') |
| |
| def handle_data(self, data): |
| self._parts.append(data) |
| |
| def get_text(self): |
| return ''.join(self._parts).strip() |
| |
| def error(self, message): |
| pass |
| |
| |
| def strip_html(html_text): |
| """Converts ResultDB summaryHtml into readable plain text.""" |
| if not html_text: |
| return '' |
| parser = _HTMLStripper() |
| parser.feed(html_text) |
| return parser.get_text() |
| |
| |
| def sanitize_filename(name, max_len=120): |
| """Sanitizes a test ID or step name for safe filesystem directory names.""" |
| sanitized = re.sub(r'[^A-Za-z0-9._-]+', '_', name).strip('_') |
| if len(sanitized) > max_len: |
| sanitized = sanitized[:max_len] |
| return sanitized or 'unknown' |
| |
| |
| def get_artifact_extension(artifact): |
| """Determines the file extension for a ResultDB artifact.""" |
| artifact_id = artifact.get('artifactId', '') |
| content_type = artifact.get('contentType', '') |
| if artifact_id in IMAGE_ARTIFACT_IDS or content_type.startswith('image/'): |
| return '.png' |
| if content_type == 'text/html' or artifact_id.endswith('_html'): |
| return '.html' |
| return '.txt' |
| |
| |
| def should_download_artifact(artifact): |
| """Returns True if an artifact is an image, diff, or test failure log.""" |
| artifact_id = artifact.get('artifactId', '') |
| content_type = artifact.get('contentType', '') |
| if artifact_id in IMAGE_ARTIFACT_IDS or artifact_id in TEXT_ARTIFACT_IDS: |
| return True |
| if content_type.startswith('image/') or content_type.startswith('text/'): |
| return True |
| return False |
| |
| |
| def format_failure_log(detail): |
| """Formats a comprehensive plain-text failure log from a ResultDB test result.""" |
| lines = [ |
| f"Test ID: {detail.get('testId', '')}", |
| f"Result Name: {detail.get('name', '')}", |
| f"Status: {detail.get('status', '')}", |
| f"Expected: {detail.get('expected', False)}", |
| f"Duration: {detail.get('duration', '')}", |
| ] |
| tags = detail.get('tags', []) |
| if tags: |
| lines.append('\nTags:') |
| for tag in tags: |
| lines.append(f" {tag.get('key')}: {tag.get('value')}") |
| |
| failure_reason = detail.get('failureReason') or {} |
| primary_error = failure_reason.get('primaryErrorMessage') |
| if primary_error: |
| lines.append('\nPrimary Error Message:') |
| lines.append(primary_error) |
| |
| errors = failure_reason.get('errors') or [] |
| if len(errors) > 1: |
| lines.append('\nAdditional Errors:') |
| for err in errors: |
| msg = err.get('message') |
| if msg: |
| lines.append(msg) |
| |
| summary_html = detail.get('summaryHtml') |
| if summary_html: |
| lines.append('\nSummary Output:') |
| lines.append(strip_html(summary_html)) |
| |
| return '\n'.join(lines) + '\n' |
| |
| |
| def fetch_failed_step_logs(build_id, builder_dir): |
| """Fetches failed step logs via `bb get` and `bb log` when no RDB results exist.""" |
| step_logs = {} |
| try: |
| stdout = subprocess.check_output( |
| ['bb', 'get', str(build_id), '-steps', '-json'], text=True) |
| build_data = json.loads(stdout) |
| except Exception: # pylint: disable=broad-except |
| return step_logs |
| |
| for step in build_data.get('steps', []): |
| if step.get('status') not in ('FAILURE', 'INFRA_FAILURE'): |
| continue |
| step_name = step.get('name', '') |
| if not step_name: |
| continue |
| safe_step = sanitize_filename(step_name) |
| log_path = os.path.join(builder_dir, f'step_{safe_step}.log') |
| try: |
| log_output = subprocess.check_output( |
| ['bb', 'log', str(build_id), step_name], |
| text=True, |
| stderr=subprocess.STDOUT) |
| os.makedirs(builder_dir, exist_ok=True) |
| with open(log_path, 'w', encoding='utf-8') as log_file: |
| log_file.write(log_output) |
| step_logs[step_name] = log_path |
| except Exception: # pylint: disable=broad-except |
| continue |
| return step_logs |
| |
| |
| def process_failed_test(detail, artifacts_for_test, builder, builder_dir, |
| test_filter_re, apply_goldens): |
| """Writes failure_log.txt and downloads pre-fetched artifacts for a test.""" |
| test_id = detail.get('testId', '') |
| screenshot_path = update_goldens.get_screenshot_path(detail) |
| if test_filter_re and not ( |
| test_filter_re.search(test_id) |
| or test_filter_re.search(detail.get('name', '')) or |
| (screenshot_path and test_filter_re.search(screenshot_path))): |
| return None |
| safe_test_id = sanitize_filename(test_id) |
| test_dir = os.path.join(builder_dir, safe_test_id) |
| os.makedirs(test_dir, exist_ok=True) |
| |
| failure_log_path = os.path.join(test_dir, 'failure_log.txt') |
| with open(failure_log_path, 'w', encoding='utf-8') as log_file: |
| log_file.write(format_failure_log(detail)) |
| |
| downloaded_artifacts = {} |
| applied_golden_path = None |
| |
| for artifact in artifacts_for_test: |
| if not should_download_artifact(artifact): |
| continue |
| artifact_id = artifact['artifactId'] |
| fetch_url = artifact.get('fetchUrl') |
| if not fetch_url: |
| continue |
| ext = get_artifact_extension(artifact) |
| filename = artifact_id if artifact_id.endswith( |
| ext) else f'{artifact_id}{ext}' |
| dest_path = os.path.join(test_dir, filename) |
| update_goldens.download_individual_screenshot(dest_path, fetch_url) |
| downloaded_artifacts[artifact_id] = dest_path |
| |
| if (apply_goldens and screenshot_path and 'linux' in builder |
| and artifact_id in ('actual_image', 'generated')): |
| update_goldens.download_individual_screenshot( |
| screenshot_path, fetch_url) |
| applied_golden_path = screenshot_path |
| |
| failure_reason = detail.get('failureReason') or {} |
| return { |
| 'testId': test_id, |
| 'status': detail.get('status'), |
| 'primaryErrorMessage': failure_reason.get('primaryErrorMessage', ''), |
| 'screenshotPath': screenshot_path, |
| 'failureLogPath': failure_log_path, |
| 'artifacts': downloaded_artifacts, |
| 'appliedGoldenPath': applied_golden_path, |
| } |
| |
| |
| def collect_cl_results(cl=None, |
| patchset=None, |
| builder_pattern=None, |
| test_filter=None, |
| output_dir=None, |
| apply_goldens=False): |
| """Collects build statuses, failure logs, and artifacts for a CL/patchset.""" |
| raw_builds = update_goldens.read_try_results(patchset=patchset, cl=cl) |
| latest_builds = update_goldens.filter_last_results(raw_builds) |
| |
| builder_re = re.compile(builder_pattern) if builder_pattern else None |
| test_filter_re = re.compile(test_filter) if test_filter else None |
| |
| if not output_dir: |
| output_dir = os.path.abspath('cl_test_results') |
| os.makedirs(output_dir, exist_ok=True) |
| |
| summary = { |
| 'cl': cl, |
| 'patchset': patchset, |
| 'outputDir': output_dir, |
| 'builders': {}, |
| } |
| |
| for builder, build_info in sorted(latest_builds.items()): |
| if builder_re and not builder_re.search(builder): |
| continue |
| if summary['cl'] is None: |
| summary['cl'] = build_info.get('cl') |
| if summary['patchset'] is None: |
| summary['patchset'] = build_info.get('patch') |
| |
| builder_entry = { |
| 'buildId': build_info['id'], |
| 'status': build_info['status'], |
| 'patch': build_info['patch'], |
| 'summaryMarkdown': build_info.get('summaryMarkdown', ''), |
| 'failedTests': [], |
| 'failedStepLogs': {}, |
| } |
| |
| if build_info['status'] not in ('SUCCESS', 'STARTED', 'SCHEDULED'): |
| builder_dir = os.path.join(output_dir, sanitize_filename(builder)) |
| unexpected = update_goldens.get_unexpected_results( |
| build_info['id']) |
| artifacts_by_result = update_goldens.get_unexpected_artifacts( |
| build_info['id']) |
| |
| unique_failures = [] |
| seen_test_ids = set() |
| for result_entry in unexpected.get('testResults', []): |
| if result_entry.get('expected'): |
| continue |
| if result_entry.get('status') not in ('FAIL', 'CRASH', |
| 'ABORT'): |
| continue |
| test_id = result_entry.get('testId', '') |
| if test_id in seen_test_ids: |
| continue |
| seen_test_ids.add(test_id) |
| unique_failures.append(result_entry) |
| |
| if unique_failures: |
| with ThreadPoolExecutor(max_workers=16) as executor: |
| futures = [ |
| executor.submit( |
| process_failed_test, result_entry, |
| artifacts_by_result.get(result_entry['name'], |
| []), builder, builder_dir, |
| test_filter_re, apply_goldens) |
| for result_entry in unique_failures |
| ] |
| for future in futures: |
| test_report = future.result() |
| if test_report: |
| builder_entry['failedTests'].append(test_report) |
| |
| if not builder_entry['failedTests'] and not test_filter_re: |
| builder_entry['failedStepLogs'] = fetch_failed_step_logs( |
| build_info['id'], builder_dir) |
| |
| summary['builders'][builder] = builder_entry |
| |
| summary_path = os.path.join(output_dir, 'summary.json') |
| with open(summary_path, 'w', encoding='utf-8') as summary_file: |
| json.dump(summary, summary_file, indent=2) |
| summary['summaryJsonPath'] = summary_path |
| return summary |
| |
| |
| def print_human_summary(summary): |
| """Prints a concise human- and agent-readable summary of CL results.""" |
| print(f"CL {summary['cl']} (Patchset {summary['patchset']})") |
| print(f"Artifacts & logs directory: {summary['outputDir']}") |
| print(f"Summary JSON: {summary['summaryJsonPath']}\n") |
| |
| for builder, info in summary['builders'].items(): |
| print(f"[{info['status']}] {builder} (build {info['buildId']})") |
| for test in info['failedTests']: |
| print(f" FAILED TEST: {test['testId']}") |
| if test['primaryErrorMessage']: |
| first_line = test['primaryErrorMessage'].splitlines()[0] |
| print(f" Error: {first_line}") |
| if test['screenshotPath']: |
| print(f" Screenshot Golden Path: {test['screenshotPath']}") |
| if test['appliedGoldenPath']: |
| print(f" Applied Golden: {test['appliedGoldenPath']}") |
| print(f" Failure Log: {test['failureLogPath']}") |
| for art_id, art_path in sorted(test['artifacts'].items()): |
| print(f" Artifact [{art_id}]: {art_path}") |
| for step_name, log_path in info['failedStepLogs'].items(): |
| print(f" FAILED STEP [{step_name}]: {log_path}") |
| |
| |
| def build_parser(): |
| """Builds the CLI argument parser.""" |
| parser = argparse.ArgumentParser( |
| description=__doc__, |
| formatter_class=argparse.RawDescriptionHelpFormatter) |
| parser.add_argument('--cl', |
| type=int, |
| help='Gerrit CL number (auto-detected if omitted).') |
| parser.add_argument('--patchset', |
| type=int, |
| help='Patchset number (latest if omitted).') |
| parser.add_argument('--builder', |
| help='Regex to filter builders (e.g. "dtf_linux.*").') |
| parser.add_argument( |
| '--test-filter', |
| help='Regex to filter failing test IDs (e.g. "DOMTreeWidget").') |
| parser.add_argument( |
| '--output-dir', |
| default='cl_test_results', |
| help='Directory to store failure logs and downloaded artifacts.') |
| parser.add_argument( |
| '--apply-goldens', |
| action='store_true', |
| help='Also copy Linux actual_image artifacts directly to test/goldens/.' |
| ) |
| parser.add_argument('--json', |
| action='store_true', |
| help='Emit full JSON report to stdout.') |
| return parser |
| |
| |
| def main(argv=None): |
| parser = build_parser() |
| options = parser.parse_args(argv) |
| summary = collect_cl_results(cl=options.cl, |
| patchset=options.patchset, |
| builder_pattern=options.builder, |
| test_filter=options.test_filter, |
| output_dir=options.output_dir, |
| apply_goldens=options.apply_goldens) |
| if options.json: |
| print(json.dumps(summary, indent=2)) |
| else: |
| print_human_summary(summary) |
| |
| |
| if __name__ == '__main__': |
| main() |
