diff --git a/tests/utils/blender_headless.py b/tests/utils/blender_headless.py new file mode 100644 index 00000000000..06e140149a1 --- /dev/null +++ b/tests/utils/blender_headless.py @@ -0,0 +1,326 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: 2011-2023 Blender Authors +# +# SPDX-License-Identifier: GPL-2.0-or-later + +""" +Wrapper for Blender that launches a graphical instances of Blender +in it's own display-server. + +This can be useful when a graphical context is required (when ``--background`` can't be used) +and it's preferable not to have windows opening on the users system. + +The main use case for this is tests that run simulated events, see: ``bl_run_operators_event_simulate.py``. + +- All arguments are forwarded to Blender. +- Headless operation checks for environment variables. +- Blender's exit code is used on exit. + +Environment Variables: + +- ``BLENDER_BIN``: the Blender binary to run. + (defaults to ``blender`` which must be in the ``PATH``). + + +WAYLAND Environment Variables: + +- ``WESTON_BIN``: The weston binary to run, + (defaults to ``weston`` which must be in the ``PATH``). +- ``WAYLAND_ROOT_DIR``: The base directory (prefix) of a portable WAYLAND installation, + (may be left unset, in that case the system's installed WAYLAND is used). +- ``WESTON_ROOT_DIR``: The base directory (prefix) of a portable WESTON installation, + (may be left unset, in that case the system's installed WESTON is used). + +Currently only WAYLAND is supported, other systems could be added. +""" +import subprocess +import sys +import signal +import os +import tempfile + +from typing import ( + Any, + Dict, + Iterator, + Optional, + Sequence, + Tuple, +) + + +# ----------------------------------------------------------------------------- +# Constants + +# For debugging, print out all information. +VERBOSE = False + +BLENDER_BIN = os.environ.get("BLENDER_BIN", "blender") + + +# ----------------------------------------------------------------------------- +# Generic Utilities + +def scantree(path: str) -> Iterator[os.DirEntry[str]]: + """Recursively yield DirEntry objects for given directory.""" + for entry in os.scandir(path): + if entry.is_dir(follow_symlinks=False): + yield from scantree(entry.path) + else: + yield entry + + +# ----------------------------------------------------------------------------- +# Implementation Back-Ends + +class backend_base: + @staticmethod + def run(args: Sequence[str]) -> int: + sys.stderr.write("No headless back-ends for {!r} with args {!r}\n".format(sys.platform, args)) + return 1 + + +class backend_wayland(backend_base): + @staticmethod + def _wait_for_wayland_server(*, socket: str, timeout: float) -> bool: + """ + Uses the expected socket file in `XDG_RUNTIME_DIR` to detect when the WAYLAND server starts. + """ + import time + time_idle = min(timeout / 100.0, 0.05) + + xdg_runtime_dir = os.environ.get("XDG_RUNTIME_DIR", "") + if not xdg_runtime_dir: + xdg_runtime_dir = "/var/run/user/{:d}".format(os.getuid()) + + filepath = os.path.join(xdg_runtime_dir, socket) + + t_beg = time.time() + t_end = t_beg + timeout + while True: + if os.path.exists(filepath): + return True + if time.time() >= t_end: + break + time.sleep(time_idle) + return False + + @staticmethod + def _weston_env_and_ini_from_portable( + *, + wayland_root_dir: Optional[str], + weston_root_dir: Optional[str], + ) -> Tuple[Optional[Dict[str, str]], str]: + """ + Construct a portable environment to run WESTON in. + """ + # NOTE(@ideasman42): WESTON does not make it convenient to run a portable instance, + # a reasonable amount of logic here is simply to get WESTON running with references to portable paths. + # Once pcakges are available on RedHad8, we might consider to remove this entire function. + weston_env = {} + weston_ini = [] + ld_library_paths = [] + + if weston_root_dir is None: + # There is very little to do, simply write a configuration + # that removes the panel to give some extra screen real estate. + weston_ini.extend([ + "[shell]", + "background-color=0x00000000", + "panel-position=none", + # Don't look for a background image. + "background-image=", + ]) + else: + weston_ini.extend([ + "[core]", + "", + "[shell]", + "background-color=0x00000000", + "client={:s}/libexec/weston-desktop-shell".format(weston_root_dir), + "panel-position=none", + # Don't look for a background image. + "background-image=", + "", + "[keyboard]", + "numlock-on=true", + "", + "[output]", + "seat=default", + "", + "[input-method]", + "path={:s}/libexec/weston-keyboard".format(weston_root_dir), + ]) + + if wayland_root_dir is not None: + ld_library_paths.append(os.path.join(wayland_root_dir, "lib64")) + + if weston_root_dir is not None: + weston_lib_dir = os.path.join(weston_root_dir, "lib") + ld_library_paths.extend([ + weston_lib_dir, + os.path.join(weston_lib_dir, "weston"), + ]) + + # Setup the `WESTON_MODULE_MAP`. + weston_map_filenames = { + "wayland-backend.so": "", + "gl-renderer.so": "", + "headless-backend.so": "", + "desktop-shell.so": "", + } + + for entry in scantree(weston_lib_dir): + if entry.name in weston_map_filenames: + weston_map_filenames[entry.name] = os.path.normpath(entry.path) + + module_map = [] + for key, value in sorted(weston_map_filenames.items()): + if not value: + raise Exception("Failure to find {!r} in {!r}".format(key, weston_lib_dir)) + module_map.append("{:s}={:s}".format(key, value)) + + weston_env["WESTON_MODULE_MAP"] = ";".join(module_map) + del module_map + + if ld_library_paths: + ld_library_paths_str = os.environ.get("LD_LIBRARY_PATH", "") + if ld_library_paths_str: + ld_library_paths.insert(0, ld_library_paths_str.rstrip(":")) + weston_env["LD_LIBRARY_PATH"] = ":".join(ld_library_paths) + del ld_library_paths_str + + return ( + {**os.environ, **weston_env} if weston_env else None, + "\n".join(weston_ini), + ) + + @staticmethod + def _weston_env_and_ini_from_system() -> Tuple[Optional[Dict[str, str]], str]: + weston_env = None + weston_ini = [ + "[shell]", + "background-color=0x00000000", + "panel-position=none", + # Don't look for a background image. + "background-image=", + ] + return ( + weston_env, + "\n".join(weston_ini), + ) + + @staticmethod + def _weston_env_and_ini() -> Tuple[Optional[Dict[str, str]], str]: + wayland_root_dir = os.environ.get("WAYLAND_ROOT_DIR") + weston_root_dir = os.environ.get("WESTON_ROOT_DIR") + + if wayland_root_dir or weston_root_dir: + weston_env, weston_ini = backend_wayland._weston_env_and_ini_from_portable( + wayland_root_dir=wayland_root_dir, + weston_root_dir=weston_root_dir, + ) + else: + weston_env, weston_ini = backend_wayland._weston_env_and_ini_from_system() + return weston_env, weston_ini + + @staticmethod + def run(blender_args: Sequence[str]) -> int: + # Use the PID to support running multiple tests at once. + socket = "wl-blender-{:d}".format(os.getpid()) + + weston_bin = os.environ.get("WESTON_BIN", "weston") + + # Ensure the WAYLAND server is NOT running (for this socket). + if backend_wayland._wait_for_wayland_server(socket=socket, timeout=0.0): + sys.stderr.write("Wayland server for socket \"{:s}\" already running, exiting!\n".format(socket)) + return 1 + + weston_env, weston_ini = backend_wayland._weston_env_and_ini() + + cmd = [ + weston_bin, + "--socket={:s}".format(socket), + "--backend=headless", + "--width=800", + "--height=600", + # `--config={..}` is added to point to a temp file. + ] + cmd_kw: Dict[str, Any] = {} + if weston_env is not None: + cmd_kw["env"] = weston_env + if not VERBOSE: + cmd_kw["stderr"] = subprocess.PIPE + cmd_kw["stdout"] = subprocess.PIPE + + if VERBOSE: + print("Env:", weston_env) + print("Run:", cmd) + + with tempfile.NamedTemporaryFile( + prefix="weston_", + suffix=".ini", + mode='w', + encoding="utf-8", + ) as weston_ini_tempfile: + weston_ini_tempfile.write(weston_ini) + weston_ini_tempfile.flush() + with subprocess.Popen( + [*cmd, "--config={:s}".format(weston_ini_tempfile.name)], + **cmd_kw, + ) as proc_server: + del cmd, cmd_kw + if not backend_wayland._wait_for_wayland_server(socket=socket, timeout=1.0): + # The verbose mode will have written to standard out/error already. + # Only show the output is the server wasn't able to start. + if not VERBOSE: + assert proc_server.stdout is not None + assert proc_server.stderr is not None + sys.stderr.write("Unable to start wayland server, exiting!\n") + sys.stderr.write(proc_server.stdout.read().decode("utf-8", errors="surrogateescape")) + sys.stderr.write(proc_server.stderr.read().decode("utf-8", errors="surrogateescape")) + sys.stderr.write("\n") + proc_server.send_signal(signal.SIGINT) + # Wait for the interrupt to be handled. + proc_server.communicate() + return 1 + + blender_env = {**os.environ, "WAYLAND_DISPLAY": socket} + + cmd = [ + BLENDER_BIN, + *blender_args, + ] + if VERBOSE: + print("Env:", blender_env) + print("Run:", cmd) + with subprocess.Popen(cmd, env=blender_env) as proc_blender: + proc_blender.communicate() + blender_exit_code = proc_blender.returncode + del cmd + + # Blender has finished, close the server. + proc_server.send_signal(signal.SIGINT) + # Wait for the interrupt to be handled. + proc_server.communicate() + + # Forward Blender's exit code. + return blender_exit_code + + +# ----------------------------------------------------------------------------- +# Main Function + +def main() -> int: + match sys.platform: + case "darwin": + backend = backend_base + case "win32": + backend = backend_base + case _: + backend = backend_wayland + return backend.run(sys.argv[1:]) + + +if __name__ == "__main__": + sys.exit(main())