servo/etc/blink-perf-test-runner/main.py

#!/usr/bin/env -S uv run --script

# Copyright 2025 The Servo Project Developers. See the COPYRIGHT
# file at the top-level directory of this distribution.
#
# Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
# http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
# <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
# option. This file may not be copied, modified, or distributed
# except according to those terms.
# /// script
# requires-python = ">=3.12"
# dependencies = ["selenium"]
# ///

import argparse
import json
import os
from pathlib import PurePath
import subprocess
import sys
import time
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.options import ArgOptions
from urllib3.exceptions import ProtocolError, MaxRetryError
from selenium.common.exceptions import NoSuchElementException
from dataclasses import dataclass

from enum import Enum
import re


class AbortReason(Enum):
    NotFound = 1
    Panic = 2


# the time in seconds we wait at most
MAX_WAIT_TIME = 60
# tests that currently do not return results
IGNORE_TESTS = [
    "tall-content-short-columns.html",
    "tall-content-short-columns-realistic.html",
]


def create_driver(webdriver_port: int, timeout: int = 3) -> webdriver.Remote | None:
    """Create the webdriver connection."""
    print("Trying to create driver")
    options = ArgOptions()
    options.set_capability("browserName", "servo")
    driver = None
    start_time = time.time()
    while driver is None and time.time() - start_time < timeout:
        try:
            driver = webdriver.Remote(command_executor=f"http://127.0.0.1:{webdriver_port}", options=options)
        except (ConnectionError, ProtocolError, MaxRetryError):
            time.sleep(0.2)
        except Exception as e:
            print(f"Unexpected exception when creating webdriver: {e}, {type(e)}")
            time.sleep(1)
    print(
        "Established Webdriver connection",
    )
    return driver


def start_servo(webdriver_port: int, servo_path: str) -> webdriver.Remote | None:
    """Start servo and create webdriver"""
    try:
        subprocess.Popen(
            [
                servo_path,
                f"--webdriver={webdriver_port}",
            ]
        )
    except FileNotFoundError:
        print("The servo binary does not exist")
        return sys.exit(1)
    return create_driver(webdriver_port)


def kill_servo():
    subprocess.Popen(["killall", "servoshell"])


REGEX_RESULTS_LOG_ELEMENT = re.compile(
    r"""
Time:\s+
values\s+[0-9.,\s]+(?P<unit>ms|runs\/s)\s+
avg\s+(?P<avg>[0-9.]+)\s+(?P=unit)\s+
median\s+[0-9.]+\s+(?P=unit)\s+
stdev\s+[0-9.]+\s+(?P=unit)\s+
min\s+(?P<min>[0-9.]+)\s+(?P=unit)\s+
max\s+(?P<max>[0-9.]+)\s+(?P=unit)
""",
    re.VERBOSE,
)


@dataclass(frozen=True)
class TestResult:
    value: float
    lower_value: float
    upper_value: float
    unit: str


def test(s: str, driver: webdriver.Remote) -> TestResult | AbortReason:
    """Run a test by loading a website, and returning (avg, min, max).
    This will run for MAX_WAIT_TIME seconds and return as soon as the avg line exists in the log element"""

    print("Running: " + canonical_test_path(s, None))
    try:
        driver.get(s)
        avg_line, min_line, max_line = None, None, None
        for i in range(MAX_WAIT_TIME):
            element = driver.find_element(By.ID, "log")
            text = element.text
            results_log_groups = REGEX_RESULTS_LOG_ELEMENT.search(text)
            if results_log_groups:
                avg_line = float(results_log_groups.group("avg"))
                min_line = float(results_log_groups.group("min"))
                max_line = float(results_log_groups.group("max"))
                unit = results_log_groups.group("unit")
            if avg_line is not None and min_line is not None and max_line is not None and unit is not None:
                return TestResult(value=avg_line, lower_value=min_line, upper_value=max_line, unit=unit)
            time.sleep(1)
    except NoSuchElementException:
        print("Could not find log?")
        return AbortReason.NotFound
    except Exception as e:
        print(f"Some other exception for this test case: {e}")
        return AbortReason.Panic
    print("Wait for valid log element timed out. Could not find valid log.")
    return AbortReason.NotFound


def canonical_test_path(filePath: str, prepend: str | None) -> str:
    """Make the filepath into just the directory and name"""
    p = PurePath(filePath)
    parts = p.parts
    index = parts.index("perf_tests")
    if prepend:
        return prepend + "/" + "/".join(parts[index:])
    else:
        return "/".join(parts[index:])


def test_file(file_name: str) -> bool:
    return ".html" in file_name and (file_name not in IGNORE_TESTS)


def write_file(results):
    with open("results.json", "w") as f:
        json.dump(results, f)


def oswalk_error(error: OSError):
    print(error)
    sys.exit(1)


def main():
    parser = argparse.ArgumentParser(description="Run Blink Perf Tests on Servo Instance.")
    parser.add_argument("servo_path", type=str, help="the servo binary")
    parser.add_argument(
        "-w", "--webdriver", default=7123, type=int, action="store", help="The webdriver port servo will listen on."
    )
    parser.add_argument(
        "-p",
        "--prepend",
        action="store",
        help="A value prepended to all results. Useful to distinguish between profiles.",
    )
    args = parser.parse_args()

    webdriver = start_servo(args.webdriver, args.servo_path)
    final_result = {}
    time.sleep(2)
    if webdriver:
        webdriver.implicitly_wait(30)
        for root, _, files in os.walk("tests/blink_perf_tests/perf_tests/layout", onerror=oswalk_error):
            for file in files:
                if test_file(file):
                    filePath = os.path.join(os.path.abspath(root), file)
                    result = test("file://" + filePath, webdriver)
                    if result == AbortReason.Panic:
                        print("Restarting servo")
                        start_servo(args.webdriver, args.servo_path)
                    elif result == AbortReason.NotFound:
                        pass
                    else:
                        combined_result = {}
                        combined_result["value"] = result.value
                        combined_result["lower_value"] = result.lower_value
                        combined_result["upper_value"] = result.upper_value

                        bencher_unit = None

                        if result.unit == "ms":
                            bencher_unit = "ms"
                        elif result.unit == "runs/s":
                            bencher_unit = "throughput"
                        else:
                            bencher_unit = "other"

                        final_result[canonical_test_path(filePath, args.prepend)] = {bencher_unit: combined_result}

    print(final_result)
    write_file(final_result)
    kill_servo()


if __name__ == "__main__":
    main()