main.py

"""
This is the cli file for orion, tool to detect regressions using hunter
"""

# pylint: disable = import-error, line-too-long, no-member
import json
import logging
import os
from pathlib import Path
import re
import sys
import warnings
from typing import Any
import xml.etree.ElementTree as ET
import xml.dom.minidom
import click
from tabulate import tabulate
from orion.logger import SingletonLogger
from orion.run_test import run, TestResults
from orion.utils import get_output_extension
from orion import constants as cnsts
from orion.config import load_config, load_ack, merge_ack_files, auto_detect_ack_file_with_vars
from orion.visualization import generate_test_html
from version import __version__

warnings.filterwarnings("ignore", message="Unverified HTTPS request.*")
warnings.filterwarnings(
    "ignore", category=UserWarning, message=".*Connecting to.*verify_certs=False.*"
)

class Dictionary(click.ParamType):
    """Class to define a custom click type for dictionaries

    Args:
        click (ParamType):
    """
    name = "dictionary"
    def convert(self, value: Any, param: Any, ctx: Any) -> dict:
        return json.loads(value)

class List(click.ParamType):
    """Class to define a custom click type for lists

    Args:
        click (ParamType):
    """
    name = "list"
    def convert(self, value: Any, param: Any, ctx: Any) -> list:
        if isinstance(value, list):
            return value
        return value.split(",") if value else []

class MutuallyExclusiveOption(click.Option):
    """Class to implement mutual exclusivity between options in click

    Args:
        click (Option): _description_
    """

    def __init__(self, *args: tuple, **kwargs: dict[str, dict]) -> None:
        self.mutually_exclusive = set(kwargs.pop("mutually_exclusive", []))
        help = kwargs.get("help", "")  # pylint: disable=redefined-builtin
        if self.mutually_exclusive:
            ex_str = ", ".join(self.mutually_exclusive)
            kwargs["help"] = help + (
                " NOTE: This argument is mutually exclusive with "
                " arguments: [" + ex_str + "]."
            )
        super().__init__(*args, **kwargs)

    def handle_parse_result(self, ctx, opts, args):
        if self.mutually_exclusive.intersection(opts) and self.name in opts:
            raise click.UsageError(
                f"Illegal usage: `{self.name}` is mutually exclusive with "
                f"arguments `{', '.join(self.mutually_exclusive)}`."
            )
        return super().handle_parse_result(ctx, opts, args)


def validate_anomaly_options(ctx, param, value: Any) -> Any: # pylint: disable = W0613
    """ validate options so that can only be used with certain flags
    """
    if value or (
        ctx.params.get("anomaly_window") or ctx.params.get("min_anomaly_percent")
    ):
        if not ctx.params.get("anomaly_detection"):
            raise click.UsageError(
                "`--anomaly-window` and `--min-anomaly-percent` can only be used when `--anomaly-detection` is enabled."
            )
    return value


# pylint: disable=too-many-locals
@click.version_option(version=__version__, message="%(prog)s %(version)s")
@click.command(context_settings={"show_default": True, "max_content_width": 180})
@click.option(
    "--cmr",
    is_flag=True,
    help="Generate percent difference in comparison",
    cls=MutuallyExclusiveOption,
    mutually_exclusive=["anomaly_detection","hunter_analyze"],
)
@click.option("--filter", is_flag=True, help="Generate percent difference in comparison")
@click.option("--config", help="Path to the configuration file", required=True)
@click.option("--ack", default="", help="Optional ack YAML to ack known regressions (can specify multiple files separated by comma)")
@click.option("--no-default-ack", is_flag=True, default=False, help="Disable automatic default ACK file detection and loading (manual --ack files are still loaded)")
@click.option(
    "--save-data-path", default="data.csv", help="Path to save the output file"
)
@click.option(
    "--github-repos",
    type=List(),
    default=[""],
    help="List of GitHub repositories (owner/repo) to enrich changepoint output with release and commit info",
)
@click.option("--sippy-pr-search", is_flag=True, help="Search for PRs in sippy")
@click.option("--debug", default=False, is_flag=True, help="log level")
@click.option(
    "--hunter-analyze",
    is_flag=True,
    help="run hunter analyze",
    cls=MutuallyExclusiveOption,
    mutually_exclusive=["anomaly_detection","cmr"],
)
@click.option("--anomaly-window", type=int, callback=validate_anomaly_options, help="set window size for moving average for anomaly-detection")
@click.option("--min-anomaly-percent", type=int, callback=validate_anomaly_options, help="set minimum percentage difference from moving average for data point to be detected as anomaly")
@click.option(
    "--anomaly-detection",
    is_flag=True,
    help="run anomaly detection algorithm powered by isolation forest",
    cls=MutuallyExclusiveOption,
    mutually_exclusive=["hunter_analyze","cmr"],
)
@click.option(
    "-o",
    "--output-format",
    type=click.Choice([cnsts.JSON, cnsts.TEXT, cnsts.JUNIT]),
    default=cnsts.TEXT,
    help="Choose output format (json, text or junit)",
)
@click.option("--save-output-path", default="output.txt", help="path to save output file with regressions")
@click.option("--column-group-size", type=int, default=5, help="Number of metrics per column group in text report")
@click.option("--uuid", default="", help="UUID to use as base for comparisons")
@click.option(
    "--baseline", default="", help="Baseline UUID(s) to to compare against uuid"
)
@click.option("--lookback", help="Get data from last X days and Y hours. Format in XdYh")
@click.option("--since", help="End date to bound the time range. When used with --lookback, creates a time window ending at this date. Format: YYYY-MM-DD")
@click.option("--convert-tinyurl", is_flag=True, help="Convert buildUrls to tiny url format for better formatting")
@click.option("--collapse", is_flag=True, help="For text output: only print regression summary to stdout (full table always saved to file). For JSON output: only include changepoint context rows.")
@click.option("--node-count", default=False, help="Match any node iterations count")
@click.option("--lookback-size", type=int, default=10000, help="Maximum number of entries to be looked back")
@click.option("--es-server", type=str, envvar="ES_SERVER", help="Elasticsearch endpoint where test data is stored, can be set via env var ES_SERVER", default="")
@click.option("--benchmark-index", type=str, envvar="es_benchmark_index",  help="Index where test data is stored, can be set via env var es_benchmark_index", default="")
@click.option("--metadata-index", type=str, envvar="es_metadata_index",  help="Index where metadata is stored, can be set via env var es_metadata_index", default="")
@click.option("--input-vars", type=Dictionary(), default="{}", help='Arbitrary input variables to use in the config template, for example: {"version": "4.18"}')
@click.option("--display", type=List(), default=["buildUrl"], help="Add metadata field as a column in the output (e.g. ocpVirt, upstreamJob)")
@click.option("--pr-analysis", is_flag=True, help="Analyze PRs for regressions", default=False)
@click.option("--viz", is_flag=True, default=False, help="Generate interactive HTML visualizations alongside output")
def main(**kwargs):
    """
    Orion runs on command line mode, and helps in detecting regressions
    """
    level = logging.DEBUG if kwargs["debug"] else logging.INFO
    if kwargs['output_format'] == cnsts.JSON :
        level = logging.ERROR
    logger = SingletonLogger(debug=level, name="Orion")
    logger.info("🏹 Starting Orion in command-line mode")

    # Load config first (needed for auto-detection)
    kwargs["config"] = load_config(kwargs["config"], kwargs["input_vars"])

    # Handle ACK file loading
    # Logic: Auto-load ack/all_ack.yaml unless --no-default-ack. Manual --ack files are always loaded and merged.
    ack_maps = []

    # Step 1: Auto-detect and load consolidated ACK file (skipped when --no-default-ack)
    if kwargs["no_default_ack"]:
        logger.info("Automatic default ACK loading disabled (--no-default-ack flag)")
    else:
        auto_ack_file = auto_detect_ack_file_with_vars(
            kwargs["config"],
            kwargs["input_vars"],
            ack_dir="ack"
        )
        if auto_ack_file:
            # Extract version and test type from config for filtering
            version = None
            test_type = None

            if "tests" in kwargs["config"] and len(kwargs["config"]["tests"]) > 0:
                test = kwargs["config"]["tests"][0]
                metadata = test.get("metadata", {})

                # Resolve version
                version_field = test.get("version_field", "ocpVersion")
                version_value = metadata.get(version_field, "")
                if version_value and "{{" not in str(version_value):
                    version = str(version_value).strip('"')
                elif version_value and "{{" in str(version_value):
                    match = re.search(r'\{\{\s*(\w+)\s*\}\}', str(version_value))
                    if match:
                        var_name = match.group(1)
                        version = kwargs["input_vars"].get(var_name) or kwargs["input_vars"].get(var_name.lower())

                # Resolve test type
                test_type_value = metadata.get("benchmark.keyword", "")
                if test_type_value and "{{" not in str(test_type_value):
                    test_type = str(test_type_value).strip()
                elif test_type_value and "{{" in str(test_type_value):
                    match = re.search(r'\{\{\s*(\w+)\s*\}\}', str(test_type_value))
                    if match:
                        var_name = match.group(1)
                        test_type = kwargs["input_vars"].get(var_name) or kwargs["input_vars"].get(var_name.lower())

            # Load auto-detected ACK file with filtering
            auto_ack_map = load_ack(auto_ack_file, version=version, test_type=test_type)
            if auto_ack_map and auto_ack_map.get("ack") and len(auto_ack_map.get("ack", [])) > 0:
                ack_maps.append(auto_ack_map)
                logger.info("✓ Auto-loaded ACK file: %s (version=%s, test_type=%s, matching entries=%d)",
                           auto_ack_file, version or "all", test_type or "all",
                           len(auto_ack_map.get("ack", [])))
            elif auto_ack_file:
                logger.info("✓ Loaded ACK file: %s (version=%s, test_type=%s, no matching entries found)",
                           auto_ack_file, version or "all", test_type or "all")

    # Load manually specified ACK files (if any) — always runs even with --no-default-ack
    if kwargs["ack"]:
        # Support multiple ACK files separated by comma
        ack_files = [f.strip() for f in kwargs["ack"].split(",") if f.strip()]
        for ack_file in ack_files:
            if len(ack_file) > 0:
                manual_ack_map = load_ack(ack_file)
                if manual_ack_map and manual_ack_map.get("ack"):
                    ack_maps.append(manual_ack_map)
                    logger.info("✓ Loaded manual ACK file: %s (entries=%d)",
                               ack_file, len(manual_ack_map.get("ack", [])))

    # Merge all ACK maps (auto-loaded + manual)
    if ack_maps:
        kwargs["ackMap"] = merge_ack_files(ack_maps)
        total_entries = len(kwargs["ackMap"].get("ack", []))
        logger.info("✓ Total ACK entries loaded: %d", total_entries)
    else:
        kwargs["ackMap"] = None
        logger.debug("No ACK entries loaded")

    if not kwargs["metadata_index"] or not kwargs["es_server"]:
        logger.error("metadata-index and es-server flags must be provided")
        sys.exit(1)
    if kwargs["pr_analysis"]:
        input_vars = kwargs["input_vars"]
        missing_vars = []
        if "jobtype" not in input_vars:
            missing_vars.append("jobtype")
        if "pull_number" not in input_vars:
            missing_vars.append("pull_number")
        if "organization" not in input_vars:
            missing_vars.append("organization")
        if "repository" not in input_vars:
            missing_vars.append("repository")
        if missing_vars:
            logger.error("Missing required input variables: %s", ", ".join(missing_vars))
            sys.exit(1)
    results, results_pull = run(**kwargs)
    is_pull = False
    if results_pull.output:
        is_pull = True
    if kwargs['output_format'] == cnsts.JSON:
        has_regression = print_json(logger, kwargs, results, results_pull, is_pull)
    elif kwargs['output_format'] == cnsts.JUNIT:
        has_regression = print_junit(logger, kwargs, results, results_pull, is_pull)
    else:
        has_regression = print_output(logger, kwargs, results, is_pull)
        if is_pull:
            print_output(logger, kwargs, results_pull, is_pull)
    if kwargs.get("viz"):
        try:
            output_base_path = str(Path(kwargs['save_output_path']).with_suffix(''))
            all_viz_data = results.viz_data
            for viz_data in all_viz_data:
                generate_test_html(viz_data, output_base_path)
        except Exception as e:  # pylint: disable=broad-except
            logger.warning("Visualization generation failed: %s", e)

    if has_regression:
        sys.exit(2) ## regression detected

def print_regression_summary(regression_data) -> None:
    """Print regression summary: affected metrics, PRs, and GitHub context tables."""
    print("Regression(s) found :")
    for regression in regression_data:
        print("-" * 50)
        print(f"Test: {regression.get('test_name')}:")
        print(f"{'Changepoint at:':<20} {regression['bad_ver']}")
        print(f"{'Previous version:':<20} {regression['prev_ver']}")
        print("\nAffected Metrics")
        if regression['metrics_with_change']:
            table = [
                [m['name'], m['value'], f"{m['percentage_change']:.2f}%", m.get('labels', '')]
                for m in regression['metrics_with_change']
            ]
            print(tabulate(table, headers=["Metric", "Value", "Percentage change", "Labels"], tablefmt="outline"))

        if "prs" in regression:
            formatted_prs = "\n".join([f"- {pr}" for pr in regression["prs"]])
        else:
            formatted_prs = "N/A"
        print("\nRelated PRs:")
        print(formatted_prs)
        if "github_context" in regression and regression["github_context"] is not None:
            print("\nGitHub context:")
            ctx = regression["github_context"]
            repos = ctx.get("repositories") or None
            if repos is None or len(repos) == 0:
                print("No GitHub context found")
                return
            for repo_name, repo_data in repos.items():
                commits = repo_data.get("commits") or {}
                releases = repo_data.get("releases") or {}
                if (commits.get("count") or 0) > 0 or (releases.get("count") or 0) > 0:
                    print(f"\nRepository: {repo_name}")
                    rows = []
                    for item in (commits.get("items") or []):
                        date = item.get("commit_timestamp", "")
                        msg = item.get("message", "")
                        email = (item.get("commit_author") or {}).get("email", "")
                        url = item.get("html_url", "")
                        rows.append([date, msg.split("\n")[0], email, url])
                    if len(rows) > 0:
                        print("Commits:")
                        print(tabulate(rows,
                            headers=["Date", "Message", "Author email", "URL"],
                            tablefmt="outline"))
                    rows = []
                    for item in (releases.get("items") or []):
                        date = item.get("published_at") or item.get("timestamp") or item.get("date") or ""
                        msg = item.get("body") or item.get("message") or item.get("name") or ""
                        email = (item.get("author") or item.get("commit_author") or {})
                        if isinstance(email, dict):
                            email = email.get("email", "")
                        else:
                            email = str(email)
                        url = item.get("html_url", "")
                        rows.append([date, msg.split("\n")[0], email, url])
                    if len(rows) > 0:
                        print("Releases:")
                        print(tabulate(rows,
                            headers=["Date", "Message", "Author email", "URL"],
                            tablefmt="outline"))


def save_text_table(test_name, result_table, save_output_path):
    """Save the text table to a file."""
    output_file_name = f"{os.path.splitext(save_output_path)[0]}_table_{test_name}.txt"
    with open(output_file_name, 'w', encoding="utf-8") as file:
        file.write(str(result_table))


def print_output(
        logger,
        kwargs,
        results: TestResults,
        is_pull: bool = False) -> bool:
    """
    Print the output of the tests

    Args:
        logger: logger object
        kwargs: keyword arguments
        results: results of the tests
        is_pull: whether the tests are pull requests
    """
    output = results.output
    regression_flag = results.regression_flag
    regression_data = results.regression_data
    average_values = results.average_values
    pr = results.pr if is_pull else 0
    if not output:
        logger.error("Terminating test")
        sys.exit(0)
    for test_name, result_table in output.items():
        save_text_table(test_name, result_table, kwargs['save_output_path'])
        if not kwargs['collapse']:
            text = test_name
            if pr > 0:
                text = test_name + " | Pull Request #" + str(pr)
            print(text)
            print("=" * len(text))
            print(result_table)
            if is_pull and pr < 1:
                text = test_name + " | Average of above Periodic runs"
                print("\n" + text)
                print("=" * len(text))
                print(average_values)
    if regression_flag:
        print_regression_summary(regression_data)
        if not is_pull:
            return True
    else:
        print("No regressions found")
    return False


def print_json(logger, kwargs, results: TestResults, results_pull: TestResults, is_pull):
    """
    Print the output of the tests in json format
    """
    logger.info("Printing json output")
    output = results.output
    regression_flag = results.regression_flag
    average_values = results.average_values
    output_pull = []
    if not output:
        logger.error("Terminating test")
        sys.exit(0)
    if is_pull and results_pull.pr:
        output_pull = results_pull.output
    for test_name, result_table in output.items():
        output_file_name = f"{os.path.splitext(kwargs['save_output_path'])[0]}_{test_name}.{get_output_extension(kwargs['output_format'])}"
        if is_pull:
            results_json = {
                "periodic": json.loads(result_table),
                "periodic_avg": json.loads(average_values),
                "pull": json.loads(output_pull.get(test_name)),
            }
            print(json.dumps(results_json, indent=2))
            with open(output_file_name, 'w', encoding="utf-8") as file:
                file.write(json.dumps(results_json, indent=2))
        else:
            print(result_table)
            with open(output_file_name, 'w', encoding="utf-8") as file:
                file.write(str(result_table))
        logger.info("Output saved to %s", output_file_name)
        if regression_flag:
            return True
    return False

def print_junit(logger, kwargs, results: TestResults, results_pull: TestResults, is_pull):
    """
    Print the output of the tests in junit format
    """
    logger.info("Printing junit output")
    output = results.output
    regression_flag = results.regression_flag
    average_values = results.average_values
    output_pull = []
    if not output:
        logger.error("Terminating test")
        sys.exit(0)
    if is_pull and results_pull.pr:
        output_pull = results_pull.output
    testsuites = ET.Element("testsuites")
    for test_name, result_table in output.items():
        if not is_pull:
            testsuites.append(result_table)
        else:
            testsuites.append(result_table)
            average_values.tag = "periodic_avg"
            testsuites.append(average_values)
            output_pull.get(test_name).tag = "pull"
            testsuites.append(output_pull.get(test_name))
        xml_str = ET.tostring(testsuites, encoding="utf8", method="xml").decode()
        dom = xml.dom.minidom.parseString(xml_str)
        pretty_xml_as_string = dom.toprettyxml()
        print(pretty_xml_as_string)
        output_file_name = f"{os.path.splitext(kwargs['save_output_path'])[0]}.{get_output_extension(kwargs['output_format'])}"
        with open(output_file_name, 'w', encoding="utf-8") as file:
            file.write(str(pretty_xml_as_string))
        logger.info("Output saved to %s", output_file_name)
        if regression_flag:
            return True
    return False