report_test_results.py

#!/usr/bin/env python
# SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import argparse
import json
import logging
import os
import sys
import typing
import xml.etree.ElementTree as ET
from datetime import date

from slack_sdk import WebClient

MAX_TEXT_LENGTH = 3000  # Slack message text limit
BLOCK_LIMIT = 20  # Slack block limit -- actual limit is 50, but we will use a smaller limit to be safe

logger = logging.getLogger()


class ReportMessages(typing.NamedTuple):
    plain_text: list[str]
    blocks: list[dict]
    failure_text: list[str] | None
    failure_blocks: list[dict] | None


def get_testcase_name(testcase: ET.Element) -> str:
    return f"{testcase.attrib.get('classname', 'Unknown')}::{testcase.attrib.get('name', 'Unknown')}"


def parse_junit(junit_file: str) -> dict[str, typing.Any]:
    tree = ET.parse(junit_file)
    root = tree.getroot()

    total_tests = 0
    total_failures = 0
    total_errors = 0
    total_skipped = 0

    failed_tests = []
    for testsuite in root.findall('testsuite'):
        total_tests += int(testsuite.attrib.get('tests', 0))
        num_failures = int(testsuite.attrib.get('failures', 0))
        num_errors = int(testsuite.attrib.get('errors', 0))
        total_failures += num_failures
        total_errors += num_errors
        total_skipped += int(testsuite.attrib.get('skipped', 0))

        if (num_failures + num_errors) > 0:
            for testcase in testsuite.findall('testcase'):
                failure = testcase.find('failure')
                error = testcase.find('error')

                for failed_test_tag in (failure, error):
                    if failed_test_tag is not None:
                        failed_info = {
                            "test_name": get_testcase_name(testcase),
                            "message": failed_test_tag.attrib.get('message', '').strip()
                        }
                        failed_tests.append(failed_info)

    return {
        "num_tests": total_tests,
        "num_failures": total_failures,
        "num_errors": total_errors,
        "num_skipped": total_skipped,
        "failed_tests": failed_tests
    }


def parse_coverage(coverage_file: str) -> str:
    tree = ET.parse(coverage_file)
    root = tree.getroot()
    coverage = root.attrib.get('line-rate', '0')
    return f"{float(coverage) * 100:.2f}%"


def get_error_string(num_errors: int, error_type: str) -> str:
    error_message = f"{error_type}: {num_errors}"
    if num_errors > 0:
        error_message = f"*{error_message}* :x:"
    return error_message


def text_to_block(text: str) -> dict:
    return {"type": "section", "text": {"type": "mrkdwn", "text": text}}


def add_text(text: str, blocks: list[dict], plain_text: list[str]) -> None:
    if len(text) > MAX_TEXT_LENGTH:
        text = text[:(MAX_TEXT_LENGTH - 3)] + "..."

    blocks.append(text_to_block(text))
    plain_text.append(text)


def chunk_items(items: list[typing.Any], chunk_size: int) -> list[list[typing.Any]]:
    return [items[index:index + chunk_size] for index in range(0, len(items), chunk_size)]


def build_messages(junit_data: dict[str, typing.Any], coverage_data: str) -> ReportMessages:
    branch_name = os.environ.get("CI_COMMIT_BRANCH", "unknown")
    num_errors = junit_data['num_errors']
    num_failures = junit_data['num_failures']

    # We need to create both a plain text message and a formatted message with blocks, the plain text message is used
    # for push notifications and accessibility purposes.
    plain_text = []
    blocks = []

    summary_line = f"Nightly CI/CD Test Results for `{branch_name}` - {date.today()}"
    plain_text.append(summary_line + "\n")

    num_errors_and_failures = num_errors + num_failures
    if num_errors_and_failures > 0:
        formatted_summary_line = f"@nat-core-devs :rotating_light: {summary_line}"
    else:
        formatted_summary_line = summary_line

    blocks.append(text_to_block(formatted_summary_line))
    test_results = "\n".join([
        get_error_string(num_failures, "Failures"),
        get_error_string(num_errors, "Errors"),
        f"Skipped: {junit_data['num_skipped']}",
        f"Total Tests: {junit_data['num_tests']}",
        f"Coverage: {coverage_data}"
    ])
    add_text(test_results, blocks, plain_text)

    failure_blocks = None
    failure_text = None
    if num_errors_and_failures > 0:
        failure_blocks = []
        failure_text = []
        add_text(f"*Failed Tests ({num_errors_and_failures}):*", failure_blocks, failure_text)

        failed_tests = junit_data['failed_tests']
        for failed_test in failed_tests:
            test_name = failed_test['test_name']
            message = failed_test['message']
            add_text(f"`{test_name}`\n```\n{message}\n```", failure_blocks, failure_text)
            failure_text.append("---\n")
            failure_blocks.append({"type": "divider"})

        job_url = os.environ.get("CI_JOB_URL")
        if job_url is not None:
            add_text(f"Full details available at: {job_url}", failure_blocks, failure_text)

    return ReportMessages(plain_text=plain_text,
                          blocks=blocks,
                          failure_text=failure_text,
                          failure_blocks=failure_blocks)


def parse_model_health(json_path: str) -> dict[str, typing.Any]:
    """Read the structured JSON output from model_health_check.py."""
    with open(json_path, encoding="utf-8") as f:
        return json.load(f)


def build_model_health_messages(health_data: dict[str, typing.Any]) -> ReportMessages:
    """Build Slack messages from model health check results."""
    branch_name = os.environ.get("CI_COMMIT_BRANCH", "unknown")

    removed = health_data.get("removed", [])
    down = health_data.get("down", [])
    deprecated = health_data.get("deprecated", [])
    ok = health_data.get("ok", [])
    num_failures = len(removed) + len(down) + len(deprecated)

    plain_text: list[str] = []
    blocks: list[dict] = []

    summary_line = f"Model Health Check for `{branch_name}` - {date.today()}"
    plain_text.append(summary_line + "\n")

    if num_failures > 0:
        formatted_summary_line = f"@nat-core-devs :rotating_light: {summary_line}"
    else:
        formatted_summary_line = summary_line

    blocks.append(text_to_block(formatted_summary_line))
    stats = "\n".join([
        get_error_string(len(removed), "Removed"),
        get_error_string(len(down), "Down"),
        get_error_string(len(deprecated), "Deprecated"),
        f"OK: {len(ok)}",
        f"Total models: {len(removed) + len(down) + len(deprecated) + len(ok)}",
    ])
    add_text(stats, blocks, plain_text)

    failure_blocks: list[dict] | None = None
    failure_text: list[str] | None = None

    if num_failures > 0:
        failure_blocks = []
        failure_text = []

        if removed:
            add_text(f"*Removed from catalog ({len(removed)}):*", failure_blocks, failure_text)
            for entry in removed:
                configs = "\n".join(f"  - {c}" for c in entry.get("configs", []))
                add_text(f"`{entry.get('model', 'unknown')}` ({entry.get('type', 'unknown')})\n{configs}",
                         failure_blocks,
                         failure_text)
            failure_blocks.append({"type": "divider"})

        if down:
            add_text(f"*Down ({len(down)}):*", failure_blocks, failure_text)
            for entry in down:
                status = entry.get("status", "?")
                detail = entry.get("detail", "")
                configs = "\n".join(f"  - {c}" for c in entry.get("configs", []))
                model_name = entry.get('model', 'unknown')
                model_type = entry.get('type', 'unknown')
                msg = f"`{model_name}` ({model_type}) HTTP {status}: {detail}\n{configs}"
                add_text(msg, failure_blocks, failure_text)

        if deprecated:
            add_text(f"*Deprecated ({len(deprecated)}):*", failure_blocks, failure_text)
            for entry in deprecated:
                detail = entry.get("detail", "")
                configs = "\n".join(f"  - {c}" for c in entry.get("configs", []))
                model_name = entry.get('model', 'unknown')
                model_type = entry.get('type', 'unknown')
                msg = f"`{model_name}` ({model_type}) Deprecated: {detail}\n{configs}"
                add_text(msg, failure_blocks, failure_text)

        job_url = os.environ.get("CI_JOB_URL")
        if job_url is not None:
            failure_blocks.append({"type": "divider"})
            add_text(f"Full details available at: {job_url}", failure_blocks, failure_text)

    return ReportMessages(plain_text=plain_text,
                          blocks=blocks,
                          failure_text=failure_text,
                          failure_blocks=failure_blocks)


def main():
    parser = argparse.ArgumentParser(description='Report test or model health status to slack channel')
    parser.add_argument('junit_file', nargs='?', default=None, type=str, help='JUnit XML file to parse')
    parser.add_argument('coverage_file', nargs='?', default=None, type=str, help='Coverage report file to parse')
    parser.add_argument('--model-health-json', type=str, default=None, help='Model health check JSON results file')

    logging.basicConfig(level=logging.INFO)

    try:
        slack_token = os.environ["SLACK_TOKEN"]
        slack_channel = os.environ["SLACK_CHANNEL"]
    except KeyError:
        logger.error('Error: Set environment variables SLACK_TOKEN and SLACK_CHANNEL to post to slack.')
        return 1

    args = parser.parse_args()

    has_test_args = args.junit_file is not None and args.coverage_file is not None
    has_health_arg = args.model_health_json is not None

    if not has_test_args and not has_health_arg:
        parser.error("Provide either junit_file + coverage_file, or --model-health-json")

    return_code = 0
    try:
        if has_health_arg:
            health_data = parse_model_health(args.model_health_json)
            report_messages = build_model_health_messages(health_data)
        else:
            junit_data = parse_junit(args.junit_file)
            coverage_data = parse_coverage(args.coverage_file)
            report_messages = build_messages(junit_data, coverage_data)
    except Exception as e:
        msg = f"Error: Failed to parse report data: {e}"
        logger.error(msg)
        plain_text = []
        blocks = []
        add_text(msg, blocks, plain_text)

        report_messages = ReportMessages(plain_text=plain_text, blocks=blocks, failure_text=None, failure_blocks=None)
        return_code = 1

    client = WebClient(token=slack_token)
    response = client.chat_postMessage(channel=slack_channel,
                                       text="\n".join(report_messages.plain_text),
                                       blocks=report_messages.blocks,
                                       link_names=report_messages.failure_text is not None)

    if report_messages.failure_text is not None:
        # Since potentially a large number of failures could occur, we will post them in a thread to the original
        # message to avoid spamming the channel.
        blocks_chunks = chunk_items(report_messages.failure_blocks or [], BLOCK_LIMIT)
        text_chunks = chunk_items(report_messages.failure_text or [], BLOCK_LIMIT)

        for blocks_chunk, text_chunk in zip(blocks_chunks, text_chunks):
            client.chat_postMessage(channel=slack_channel,
                                    text="\n".join(text_chunk),
                                    blocks=blocks_chunk,
                                    thread_ts=response["ts"])

    return return_code


if __name__ == '__main__':
    sys.exit(main())