Tools/Scripts/webkit-test-results

#!/usr/bin/env python3
#
# Copyright (C) 2021 Igalia S.L.
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this program; if not, write to the
# Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
# Boston, MA 02110-1301, USA.
# pylint: disable=missing-docstring,invalid-name

"""Command-line interface to the Webkit Results Database"""

import argparse
from datetime import datetime
import functools
import json
import logging
import os
import pathlib
import sys
import textwrap
from typing import Optional
from urllib.parse import urljoin, urlencode, parse_qs
from urllib.request import urlopen
from urllib.error import HTTPError

BASE_URL = "https://results.webkit.org/"


def red(msg):
    """Returns a shell-aware red string"""
    return "\033[91m" + msg + "\033[0m"


def green(msg):
    """Returns a shell-aware green string"""
    return "\033[92m" + msg + "\033[0m"


# Roughly ordered from the most desired to the most unwanted.
OUTCOMES = [
    "PASS",
    "TEXT",
    "IMAGE",
    "SKIP",
    "FAIL",
    "ERROR",
    "TIMEOUT",  # Put timeout as more severe than ERROR as they often are harder to debug
    "CRASH",
]


def is_improvement(baseline, current):
    """Returns true if the current outcome is more favourable."""
    best_baseline = min(OUTCOMES.index(baseline_token) for baseline_token in baseline.split())
    return best_baseline > OUTCOMES.index(current)


def is_regression(baseline, current):
    """Returns true if the current outcome downgraded the result."""
    worst_baseline = max(OUTCOMES.index(baseline_token) for baseline_token in baseline.split())
    return worst_baseline < OUTCOMES.index(current)


def decorate(msg, baseline, current):
    """Colors the message if the status change represents an improvement or regression."""
    if is_improvement(baseline, current):
        msg = green(msg)
    elif is_regression(baseline, current):
        msg = red(msg)
    return msg


class Query:
    """Wrapper around the most common query parameters for a given request"""

    def __init__(self, **kwargs):
        """Just stores the passed kwargs as parameters to be forwarded to the actual query"""
        self._params = kwargs.copy()

    @classmethod
    def from_query_string(cls, query_str):
        fields = parse_qs(query_str)
        # parse_qs forces the values to be lists, even when they are single-valued
        for k, v in fields.items():
            if len(v) == 1:
                fields[k] = v[0]
        return cls(**fields)

    def as_query_string(self):
        """Escape and build the actual query string"""
        return urlencode(self._params)

    def add_param(self, **kwargs):
        """Extends the current set of parameters.

        Beware that it'll overwrite existing values."""
        self._params.update(kwargs)


BOTS = {
    "gtk-release": Query(platform="GTK", style="release", version_name="Xvfb"),
    "gtk-debug": Query(platform="GTK", style="debug", version_name="Xvfb"),
    "wpe-release": Query(platform="WPE", style="release", flavor="wk2"),
    "wpe-debug": Query(platform="WPE", style="debug", flavor="wk2"),
}


def get_commit_cache_filename():
    """Returns the cache filename, ensuring it's parent folder exists"""
    app_name = "webkit-test-results"
    xdg_config_home = os.getenv("XDG_CACHE_HOME")
    if xdg_config_home:
        directory = os.path.join(xdg_config_home, app_name)
    else:
        directory = os.path.join(os.path.expanduser("~"), ".cache", app_name)

    if not os.path.isdir(directory):
        pathlib.Path(directory).mkdir(parents=True, exist_ok=True)

    return os.path.join(directory, "commits.json")


def load_commit_cache(force_reset=False):
    """Loads the stored commits, fetching again if needed."""
    filename = get_commit_cache_filename()
    if force_reset or not os.path.isfile(filename):
        return reset_commit_cache()

    logging.info("Opening commit cache %s", filename)
    with open(filename, encoding="utf-8") as handle:
        return json.load(handle)


def reset_commit_cache():
    """Fetches the last 1000 commits and store their info to be reused in later calls."""
    # TODO Append to existing instead of resetting
    logging.info("Resetting commit cache")
    limit = 1000
    command = "/api/commits"

    query = {"limit": limit, "branch": "main"}

    url = urljoin(BASE_URL, command) + "?" + urlencode(query)

    filename = get_commit_cache_filename()
    logging.info("Fetching commits from %s", url)
    with urlopen(url) as response:
        raw_data = json.load(response)

    uuids = {}
    for commit in raw_data:
        # Order is usually zero for single commits pushed to the repo. For example,
        # commits in SVN. When moving to git, branchs with multiple commits will
        # make use of it.
        uuid = str(commit["timestamp"] * 100 + commit["order"])
        commit["uuid"] = uuid
        uuids[uuid] = commit

    logging.info("Saving commits to %s", filename)
    with open(filename, "w", encoding="utf-8") as output:
        logging.info("Saving downloaded cache")
        json.dump(uuids, output)

    return uuids


def last_run(args):
    """Shows data about the last test run registered for the selected bot"""
    endpoint = "api/results/layout-tests"
    if args.only_changes:
        logging.info("--only-changes ignored in this command")
    if args.only_unexpected:
        logging.info("--only-unexpected ignored in this command")

    configuration = BOTS.get(args.bot, Query.from_query_string(args.bot))
    configuration.add_param(limit=1)
    query = configuration.as_query_string()

    url = urljoin(BASE_URL, endpoint) + "?" + query

    logging.info("Loading test data from %s", url)
    try:
        with urlopen(url) as response:
            data = json.load(response)
        print(json.dumps(data, sort_keys=True, indent=4, separators=(",", ": ")))
    except (HTTPError, json.JSONDecodeError) as e:
        print(e)
        return 1

    return 0


def get_latest_commit(commits):
    """Returns the newest commit from a set of commits"""

    latest = max(commits.values(), key=lambda x: x["uuid"])

    return latest["identifier"], latest["timestamp"]


@functools.cache
def version_string(platform: str, version_number: int, version_name: Optional[str]) -> str:
    """Translates the version number into a version string, following the resultsdb rules"""

    if platform.lower() not in ('wpe', 'gtk') and version_name:
        return version_name

    if not version_number:
        return "0.0.0"

    micro = version_number % 1000
    minor = (version_number // 1000) % 1000
    major = version_number // 1000_000

    return f"{major}.{minor}.{micro}"


def report_test(args):
    """Reports the test history for a single testcase in a single configuration"""
    endpoint = "api/results/layout-tests"
    configuration = BOTS.get(args.bot, Query.from_query_string(args.bot))
    if args.limit > 0:
        configuration.add_param(limit=args.limit)
    query = configuration.as_query_string()

    url = urljoin(BASE_URL, endpoint + "/" + args.test) + "?" + query

    logging.info("Loading test data from %s", url)
    with urlopen(url) as response:
        data = json.load(response)

    if not data:
        logging.error("No results returned. Exiting.")
        return 1

    # Until early 2026 the linux bots used the kernel x.y.z version as the test result version,
    # resulting in multiple configurations in a short timeframe, so we need to flatten them.
    all_results = []
    for configuration_result in data:
        config_data = configuration_result["configuration"]
        for result in configuration_result["results"]:
            result.update(config_data)
            result["version"] = version_string(result['platform'], result.get("version", 0), result.get('version_name'))
            all_results.append(result)
    results = sorted(all_results, key=lambda result: result["uuid"])

    if args.limit > 0:
        results = results[-args.limit :]

    commits = load_commit_cache(force_reset=args.reset_cache)
    logging.info("Found %d cached commits", len(commits))

    previous = None

    matched = False

    for result in results:
        actual = result["actual"]
        expected = result["expected"]
        start_time = datetime.fromtimestamp(result["start_time"])
        uuid = result["uuid"]

        config_str = (
            "{platform} {version} {flavor} {style} with {architecture}".format(**result)
        )

        try:
            commit = commits[str(uuid)]
            matched = True
        except KeyError:
            logging.info("Could not find commit with uuid %s", uuid)
            continue

        if args.only_changes:
            if previous == actual:
                continue
        elif args.only_unexpected:
            if actual == expected:
                continue

        msg = "{} '{}' {}: expected {} actual {} ".format(
            commit.get("identifier", 'UNKNOWN'), config_str, start_time, expected, actual
        )

        if args.color:
            print(decorate(msg, expected, actual))
        else:
            print(msg)
        previous = actual

    if not matched:
        latest_revision, latest_timestamp = get_latest_commit(commits)
        latest_date = datetime.fromtimestamp(latest_timestamp)
        print(
            f"No commits matched. The latest commit in cache is {latest_revision} from {latest_date}",
            file=sys.stderr,
        )
        return 1

    return 0


def parse_args():
    """Parse command line arguments and switches"""
    parser = argparse.ArgumentParser(
        description=textwrap.dedent(
            """
                Command line tool to query https://results.webkit.org for test history.

                Passing a test case in the command line will show one entry for each
                test run registered, alongside commit information, expected and actual outcomes,
                and timestamp of the run. For example:

                ./Tools/Scripts/webkit-test-results --bot wpe-debug css3/filters/drop-shadow-current-color.html --limit=5
                305904@main 'WPE 2.51.0 wk2 debug with x86' 2026-01-20 21:30:13: expected PASS actual CRASH
                305907@main 'WPE 2.51.0 wk2 debug with x86' 2026-01-21 00:24:09: expected PASS actual CRASH
                305913@main 'WPE 2.51.0 wk2 debug with x86' 2026-01-21 03:16:45: expected PASS actual CRASH
                305923@main 'WPE 2.51.0 wk2 debug with x86' 2026-01-21 06:01:24: expected PASS actual CRASH
                305928@main 'WPE 2.51.0 wk2 debug with x86' 2026-01-21 08:44:56: expected PASS actual CRASH

                To avoid querying the commit data at each invocation, the last 1000 commits
                are cached (by default, to $XDG_CACHE_HOME/wk-gardening-tools/commits.json).

                Currently, only GLIB-based bots and layout-test suite are supported.
            """
        ),
        formatter_class=argparse.RawTextHelpFormatter,
    )

    bots_group = parser.add_mutually_exclusive_group()
    bots_group.add_argument("-b", "--bot", help="Bot to query")
    bots_group.add_argument("--list-bots", action="store_true", help="List predefined bots")

    # Common options
    parser.add_argument("-v", "--verbose", action="store_true", help="Verbose output")
    parser.add_argument(
        "-r",
        "--reset-cache",
        action="store_true",
        help="Reset the commit cache if needed",
    )
    parser.add_argument(
        "-c",
        "--color",
        action="store_true",
        help="Use colors to highlight unexpected results",
    )

    parser.add_argument(
        "--last-run",
        action="store_true",
        help="Show status of the last registered run and exit.",
    )

    parser.add_argument(
        "--limit",
        type=int,
        default=100,
        help="""Limit the number of results. Defaults to 100, use -1 to ask the server for
                its default value. Use together with --test""",
    )


    # Output options
    output_group = parser.add_mutually_exclusive_group()
    output_group.add_argument(
        "--only-changes",
        action="store_true",
        help="Display only revisions where the state changed. Use with --test",
    )
    output_group.add_argument(
        "--only-unexpected",
        action="store_true",
        help="Display only unexpected results. Use with --test",
    )

    parser.add_argument(
        "test", help="Test case to be searched", nargs="?", default=None
    )

    return parser.parse_args()


def main():
    """Main entry point"""
    args = parse_args()

    if args.verbose:
        logging.basicConfig(level=logging.DEBUG)

    if args.list_bots:
        for bot in sorted(BOTS.keys()):
            print(bot)
    elif args.last_run:
        sys.exit(last_run(args))
    elif args.test:
        if not args.bot:
            print("Test history requires a bot to be specified with -b or --bot. Exiting.")
            sys.exit(1)
        sys.exit(report_test(args))
    else:
        print("No test case provided. Did you forget to pass it?")
        sys.exit(1)


if __name__ == "__main__":
    main()