#!/usr/bin/env vpython3 # Copyright 2025 The Chromium Authors # Use of this source code is governed by a BSD-style license that can be # found in the LICENSE file. """Performance test suite for video playback on a laptop device. This script uses Selenium and Chromedriver to automate performance tests for video playback. It sets up an SSH tunnel to a remote machine, records the video using ffmpeg, and analyzes the output for metrics like dropped frames and smoothness. """ import argparse import logging import os import shutil import subprocess import sys import time from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.chrome.options import Options as ChromeOptions from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as ec import common # pylint: disable=import-error, wrong-import-position from repeating_log import RepeatingLog # pylint: enable=import-error, wrong-import-position CHROME_OPTIONS = [ # Redirects logging output to stderr to better catch automation issues. "--enable-logging=stderr", # Sets the default verbose logging level to 1. "--v=1", # Disables the sandbox, often necessary in automated test environments. "--no-sandbox", # Disables the GPU sandbox, used to prevent issues with GPU crashes. "--disable-gpu-sandbox", # Launches Chrome in fullscreen mode to prevent scrollbar clipping. "--start-fullscreen", ] def connect_to_remote_driver(chrome_options, binary_location): """Attempts to connect to the remote chromedriver via the tunnel.""" logging.info("Attempting connection to %s.", common.REMOTE_URL) # Set the binary location directly on the options object. if binary_location: chrome_options.binary_location = binary_location for _ in range(20): try: driver = webdriver.Remote( command_executor=common.REMOTE_URL, options=chrome_options ) logging.info("Successfully connected!") return driver except Exception as e: #pylint: disable=broad-exception-caught logging.info("Tunnel not yet up. Sleeping ... Error: %s", e) time.sleep(2) raise RuntimeError("Could not connect to the remote chromedriver.") def setup_test_environment(args, chrome_version): """ Sets up the remote chromedriver and SSH tunnel for testing. This function terminates any old Chromedriver processes, starts a new one, waits for it to be ready, and then establishes an SSH tunnel to it. It then connects a WebDriver instance to the tunnel and enables Cast discovery. Returns: tuple: A tuple containing the WebDriver, the tunnel process, and the actual chrome version used. """ if args.sender_os == 'cros': return common.setup_cros_environment( args, chrome_version, CHROME_OPTIONS) common.terminate_old_chromedriver(args) remote_app_path, actual_version = common.install_and_setup_chrome( args, chrome_version) common.wait_for_chromedriver(args) tunnel_proc = common.start_ssh_tunnel(args) chrome_options = ChromeOptions() for option in CHROME_OPTIONS: chrome_options.add_argument(option) binary_path = None if args.sender_os == 'mac': binary_path = (f'{remote_app_path}/Contents/MacOS/Google Chrome for ' 'Testing') logging.info( "Mac OS detected. Setting binary_location to: %s", binary_path) elif args.sender_os == 'win': logging.info( "Windows OS detected. Setting binary_location to: %s", remote_app_path) binary_path = remote_app_path chrome_options.binary_location = binary_path driver = connect_to_remote_driver(chrome_options, binary_path) return driver, tunnel_proc, actual_version # pylint: disable=too-many-locals def run_performance_test(video_file: str, driver: webdriver, args): """ Runs a single video performance test by playing and recording the video. This function navigates to the video player page, starts an ffmpeg recording process, plays a video, and then analyzes the recorded output for various performance metrics. Args: video_file (str): The name of the video file to be tested. driver (webdriver.Remote): The Selenium WebDriver instance. args: The parsed command-line arguments. Returns: subprocess.Popen: The Popen object for the ffmpeg recording process. """ # force video output to mp4 output_file = os.path.join(common.RECORDINGS_DIR, video_file.replace('.webm', '.mp4')) host_recording_cmd = [ 'ffmpeg', # Overwrite output files without asking. '-y', # Set the input format to Video4Linux2. '-f', 'video4linux2', # Force V4L2 capture framerate to 60fps. '-framerate', '60', # Set the input pixel format. '-input_format', 'yuyv422', # Specify the input file (video device). '-i', '/dev/video1', # Set the size of the input buffer to help prevent dropped frames. '-thread_queue_size', '1024', # Set the video codec to libx264 (H.264). '-c:v', 'libx264', # Use the ultrafast preset for real-time encoding. '-preset', 'ultrafast', # Set the Constant Rate Factor for quality (lower is better). '-crf', '28', # Set the output pixel format for compatibility. '-pix_fmt', 'yuv420p', # Set the Group of Pictures (GOP) size for better seeking. '-g', '60', # Set the duration of the recording. '-t', '35', output_file ] wait = WebDriverWait(driver, 30) driver.get(f'http://{common.LOCAL_HOST_IP}:' f'{common.SERVER_PORT}/video.html?file={video_file}') wait.until(ec.presence_of_element_located((By.ID, "video"))) glances_proc = None if args.sender_os == 'win': csv_remote_path = f"C:/Users/Public/glances_{video_file}.csv" else: csv_remote_path = f"/tmp/glances_{video_file}.csv" csv_local_path = os.path.join( common.TRACES_DIR, f"glances_{video_file}.csv") try: glances_proc = common.start_glances_monitoring( args, csv_remote_path) # pylint: disable=consider-using-with rec_proc_local = subprocess.Popen( host_recording_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) logging.info("ffmpeg recording process started. Waiting for 'Stream " "mapping:' confirmation...") while True: line = rec_proc_local.stderr.readline() # Use local variable if line: line = line.strip() logging.info("FFMPEG STARTUP: %s", line) if "Stream mapping:" in line: logging.info("Started recording.") break def _wait_js_condition(driver, element, condition: str) -> bool: """Waits a condition on the element once a second for at most 30 seconds, returns True if the condition met.""" start = time.time() while not driver.execute_script(f'return arguments[0].{condition};', element): if time.time() - start >= 30: return False time.sleep(1) return True video = driver.find_element(By.ID, 'video') with common.measures.time_consumption( video_file, 'video_perf', 'playback', 'loading'), \ RepeatingLog(f'Waiting for video {video_file} to be loaded.'): if not _wait_js_condition(driver, video, 'readyState >= 2'): logging.warning( '%s may never be loaded, still go ahead to play it.', video_file) common.measures.average(video_file, 'video_perf', 'playback', 'failed_to_load').record(1) video.click() logging.info("Started playing video.") logging.info("Playing media for 30 seconds (script will then quit)...") time.sleep(30) rec_proc_local.communicate() logging.info("recording finished.") results = common.video_analyzer.from_original_video( output_file, f"/usr/local/cipd/videostack_videos_30s/{video_file}") if not results: raise RuntimeError("Missing video analyzer results. See log for " "further details.") def record(key: str) -> None: # If the video_analyzer does not generate any result, treat it as an # error and use the default value to filter them out instead of # failing the tests. common.measures.average(video_file, 'video_perf', key).record( results.get(key, common.FAIL_CODE)) for metric in common.METRICS: record(metric) original_video = f"/usr/local/cipd/videostack_videos_30s/{video_file}" common.calculate_psnr_ssim(video_file, output_file, original_video) logging.warning('Video analysis result of %s: %s', video_file, results) finally: if glances_proc: try: common.stop_glances_monitoring( args, glances_proc, csv_remote_path, csv_local_path) common.parse_glances_csv_and_record( video_file, csv_local_path, args.sender_os) except Exception as e: logging.error( "Failed to stop or parse glances monitoring: %s", e) return rec_proc_local def main(): """ Runs the performance testing suite for all videos. This function sets up a single remote test environment (Chromedriver and SSH tunnel) and then iterates through a list of videos. For each video, it runs a performance test, logs any errors, and cleans up the video-specific resources (like the recording process). Finally, it tears down the shared test environment (Chromedriver and SSH tunnel). Returns: int: The exit code for the script, typically 0 for success. """ logging.getLogger().setLevel(logging.INFO) parser = argparse.ArgumentParser( description="Performance test for media played on a laptop.", ) parser.add_argument('--username', help='Sender device username.') parser.add_argument('--sender', help='Sender device IP.') parser.add_argument( '--chrome-version', default=None, help='Chrome for Testing version to use. Defaults to the latest ' 'known good version.') parser.add_argument('--sender-os', choices=['mac', 'win', 'linux', 'cros'], help='OS of the sender device.') args, _ = parser.parse_known_args() cv = args.chrome_version if os.path.exists(common.RECORDINGS_DIR): shutil.rmtree(common.RECORDINGS_DIR) os.makedirs(common.RECORDINGS_DIR) driver = None tunnel_proc = None actual_version = None try: driver, tunnel_proc, actual_version = setup_test_environment(args, cv) for video in common.VIDEOS: # TODO(b/512198717): Enable HEVC tests on ChromeOS. # Currently these tests are rendering a blank white screen, so we # skip them to bring up the other cros tests. if args.sender_os == 'cros' and 'HEVC' in video['name']: logging.info("Skipping HEVC on ChromeOS: %s", video['name']) continue logging.info("Starting test for video: %s", video['name']) rec_proc = None try: rec_proc = run_performance_test(video['name'], driver, args) except Exception: # pylint: disable=broad-exception-caught logging.exception("Error during video %s test", video['name']) raise finally: common.teardown_recording_process(rec_proc) finally: common.finalize_results(actual_version) common.teardown_test_environment(driver, tunnel_proc, args) if __name__ == '__main__': with common.StartProcess(common.server.start, [common.SERVER_PORT], True): sys.exit(main())