From ad8cba5fb9e7557ed27312a6018839450220b03f Mon Sep 17 00:00:00 2001 From: Jeremy Maitin-Shepard Date: Tue, 30 Jun 2026 10:33:49 -0700 Subject: [PATCH 1/2] Introduce Lua APIs for animation status Exposes scroll.animating() and scroll.pending_transactions() to Lua to allow external scripts and tests to query whether the compositor has settled. Also includes testing infrastructure updates (pytest.ini config, LSan suppressions, and helper methods in test_utils.py). --- pytest.ini | 2 + scroll.lua | 23 +++ sway/lua.c | 17 ++ sway/scroll.5.scd | 22 ++- tests/conftest.py | 172 ++++++++++++++++----- tests/interactive.sh | 4 +- tests/meson.build | 3 + tests/test_animation_offset.py | 129 ---------------- tests/test_clients.py | 85 +++------- tests/test_crash_move.py | 11 +- tests/test_focused_inactive_null_crash.py | 9 +- tests/test_focused_inactive_uaf.py | 40 +++-- tests/test_geometry.py | 96 +----------- tests/test_leak_two_clients.py | 37 +++++ tests/test_lua_path.py | 83 +++++----- tests/test_lua_safety.py | 1 + tests/test_move_cleanup_crash.py | 36 ++--- tests/test_normal_exit.py | 53 +++++-- tests/test_rapid_switch.py | 22 +++ tests/test_shutdown_events.py | 118 ++++++++++++++ tests/test_shutdown_lua_callbacks.py | 89 +++++++++++ tests/test_space_aba.py | 76 ++++----- tests/test_space_crash.py | 63 +++----- tests/test_utils.py | 180 +++++++++++++++++++++- tests/test_workspace_animation_uaf.py | 24 +++ tests/test_workspace_split_uaf.py | 45 ++---- 26 files changed, 886 insertions(+), 554 deletions(-) create mode 100644 pytest.ini delete mode 100644 tests/test_animation_offset.py create mode 100644 tests/test_leak_two_clients.py create mode 100644 tests/test_rapid_switch.py create mode 100644 tests/test_shutdown_events.py create mode 100644 tests/test_shutdown_lua_callbacks.py create mode 100644 tests/test_workspace_animation_uaf.py diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 00000000..82cbd5bd --- /dev/null +++ b/pytest.ini @@ -0,0 +1,2 @@ +[pytest] +addopts = -n 4 diff --git a/scroll.lua b/scroll.lua index 09330ba4..e3f39832 100644 --- a/scroll.lua +++ b/scroll.lua @@ -653,4 +653,27 @@ function scroll.add_callback(event, cb_func, cb_data) end --- @return integer function scroll.remove_callback(id) end +--- +--- Returns true if there is an active animation running. +--- +--- This is primarily intended for use via the IPC interface for testing and +--- debugging. It is generally not useful when called synchronously within a +--- Lua script, as animations only start after the script execution completes. +--- +--- @return boolean +--- +function scroll.animating() end + +--- +--- Returns true if there are pending transactions that haven't been applied yet. +--- +--- This is primarily intended for use via the IPC interface for testing and +--- debugging. It is generally not useful when called synchronously within a +--- Lua script, as transactions are not processed while the script is running +--- and blocking the main thread. +--- +--- @return boolean +--- +function scroll.pending_transactions() end + return scroll diff --git a/sway/lua.c b/sway/lua.c index 30da24b0..03eea9af 100644 --- a/sway/lua.c +++ b/sway/lua.c @@ -14,6 +14,7 @@ #include "sway/desktop/launcher.h" #include "sway/ipc-server.h" #include "sway/desktop/transaction.h" +#include "sway/server.h" #if 0 static void print_table(lua_State *L, int index); @@ -1727,7 +1728,20 @@ static int scroll_remove_callback(lua_State *L) { return 0; } +static int scroll_animating(lua_State *L) { + lua_pushboolean(L, animation_animating()); + return 1; +} + +static int scroll_pending_transactions(lua_State *L) { + bool pending = server.queued_transaction != NULL || server.pending_transaction != NULL || + server.dirty_nodes->length > 0; + lua_pushboolean(L, pending); + return 1; +} + // Module functions +/* clang-format off */ static luaL_Reg const scroll_lib[] = { { "log", scroll_log }, { "state_set_value", scroll_state_set_value }, @@ -1798,8 +1812,11 @@ static luaL_Reg const scroll_lib[] = { { "scratchpad_hide", scroll_scratchpad_hide }, { "add_callback", scroll_add_callback }, { "remove_callback", scroll_remove_callback }, + { "animating", scroll_animating }, + { "pending_transactions", scroll_pending_transactions }, { NULL, NULL } }; +/* clang-format on */ // Module Loader int luaopen_scroll(lua_State *L) { diff --git a/sway/scroll.5.scd b/sway/scroll.5.scd index a7d9f3da..95c572b3 100644 --- a/sway/scroll.5.scd +++ b/sway/scroll.5.scd @@ -2402,7 +2402,27 @@ local process_data = scroll.exec_process("kitty") Removes a callback set earlier using *add_callback*. _id_ is the unique identifier returned by *add_callback*. -Examples: +## TESTING AND DEBUGGING API + +These functions are primarily intended for use via the IPC interface (see +*scroll-ipc*(7)) for testing and debugging purposes, and are generally not +useful when called synchronously from standard Lua scripts or event callbacks. + +Because scroll's Lua integration executes scripts synchronously on the main +compositor thread, the event loop is blocked during execution. This means that +any compositor actions triggered by the script (such as applying window layouts +or starting animations) will only occur after the script completes. +Consequently, calling these functions synchronously within a script will not +reflect the state changes of commands run immediately before them. + +*animating()* + Returns _true_ if there is an active animation running. + +*pending_transactions()* + Returns _true_ if there are pending transactions that haven't been applied + yet. + +## EXAMPLES Calling this script from the configuration file, you will get focus on every window that gets the *urgent* attribute: diff --git a/tests/conftest.py b/tests/conftest.py index 4531b51c..9705557c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -3,69 +3,171 @@ import pytest from typing import Generator from pathlib import Path -from test_utils import ScrollInstance, run_compositor +from test_utils import ScrollInstance, run_compositor, ScrollCompositorFactory def pytest_addoption(parser: pytest.Parser) -> None: parser.addoption("--scroll", help="the scroll binary to test", default=None) +@pytest.hookimpl(tryfirst=True, hookwrapper=True) +def pytest_runtest_makereport( + item: pytest.Item, call: pytest.CallInfo +) -> Generator[None, None, None]: + outcome = yield + rep = outcome.get_result() + setattr(item, "result_" + rep.when, rep) + + +def _build_scroll() -> str: + # Auto-build using Meson/Ninja + print("\nBuilding scroll with Meson/Ninja...") + build_dir = os.path.abspath("./build") + if not os.path.exists(build_dir): + res = subprocess.run( + [ + "meson", + "setup", + "build", + "-Dwerror=false", + "-Db_sanitize=address", + "-Dbuildtype=debugoptimized", + ], + capture_output=True, + text=True, + ) + if res.returncode != 0: + pytest.exit( + f"Failed to setup build:\nStdout: {res.stdout}\nStderr: {res.stderr}" + ) + else: + # Ensure ASan is enabled + res = subprocess.run( + [ + "meson", + "configure", + "build", + "-Db_sanitize=address", + "-Dbuildtype=debugoptimized", + ], + capture_output=True, + text=True, + ) + if res.returncode != 0: + pytest.exit( + "Failed to configure build with ASan:\nStdout:" + f" {res.stdout}\nStderr: {res.stderr}" + ) + + # Run ninja to compile (incremental build) + res = subprocess.run(["ninja", "-C", "build"], capture_output=True, text=True) + if res.returncode != 0: + pytest.exit( + f"Failed to build scroll:\nStdout: {res.stdout}\nStderr: {res.stderr}" + ) + + return os.path.join(build_dir, "sway", "scroll") + + @pytest.fixture(scope="session") def scroll_compositor_binary(request: pytest.FixtureRequest) -> str: binary_path: str = request.config.getoption("scroll") if not binary_path: - # Auto-build using Meson/Ninja - print("\nBuilding scroll with Meson/Ninja...") - build_dir = os.path.abspath("./build") - if not os.path.exists(build_dir): - res = subprocess.run( - ["meson", "setup", "build", "-Dwerror=false", "-Db_sanitize=address"], - capture_output=True, - text=True, - ) - if res.returncode != 0: - pytest.exit( - f"Failed to setup build:\nStdout: {res.stdout}\nStderr: {res.stderr}" - ) + # Check if we are running under xdist + try: + worker_id = request.getfixturevalue("worker_id") + except Exception: + worker_id = "master" + + if worker_id == "master": + binary_path = _build_scroll() else: - # Ensure ASan is enabled - res = subprocess.run( - ["meson", "configure", "build", "-Db_sanitize=address"], - capture_output=True, - text=True, - ) - if res.returncode != 0: - pytest.exit( - f"Failed to configure build with ASan:\nStdout: {res.stdout}\nStderr: {res.stderr}" - ) + tmp_path_factory = request.getfixturevalue("tmp_path_factory") + shared_dir = tmp_path_factory.getbasetemp().parent + lock_path = shared_dir / "scroll_build.lock" + status_path = shared_dir / "scroll_build.status" - # Run ninja to compile (incremental build) - res = subprocess.run(["ninja", "-C", "build"], capture_output=True, text=True) - if res.returncode != 0: - pytest.exit( - f"Failed to build scroll:\nStdout: {res.stdout}\nStderr: {res.stderr}" - ) + import fcntl - binary_path = os.path.join(build_dir, "sway", "scroll") + # Open with 'a' to avoid truncating while another process might have it locked + with open(lock_path, "a") as lock_file: + fcntl.flock(lock_file, fcntl.LOCK_EX) + try: + if status_path.exists(): + binary_path = status_path.read_text().strip() + else: + binary_path = _build_scroll() + status_path.write_text(binary_path) + finally: + fcntl.flock(lock_file, fcntl.LOCK_UN) else: binary_path = os.path.abspath(binary_path) assert os.path.exists(binary_path), f"Binary not found at {binary_path}" + + # Set up PATH to include build directories so that the compositor can find + # our newly built scrollbar, swaymsg, swaynag, etc. + build_dir = Path(binary_path).parent.parent + old_path = os.environ.get("PATH", "") + build_paths = [ + str(build_dir / "sway"), + str(build_dir / "swaymsg"), + str(build_dir / "swaybar"), + str(build_dir / "swaynag"), + ] + os.environ["PATH"] = ":".join(build_paths) + ":" + old_path + return binary_path @pytest.fixture(scope="session") -def scroll_compositor( +def scroll_compositor_factory( scroll_compositor_binary: str, tmp_path_factory: pytest.TempPathFactory -) -> Generator[ScrollInstance, None, None]: +) -> Generator[ScrollCompositorFactory, None, None]: temp_dir: Path = tmp_path_factory.mktemp("scroll") with run_compositor(scroll_compositor_binary, temp_dir) as inst: + yield ScrollCompositorFactory(inst) + + +@pytest.fixture(scope="function") +def scroll_compositor( + request: pytest.FixtureRequest, + scroll_compositor_factory: ScrollCompositorFactory, +) -> Generator[ScrollInstance, None, None]: + ctx = scroll_compositor_factory() + inst = ctx.__enter__() + try: yield inst + finally: + failed = ( + hasattr(request.node, "result_call") and request.node.result_call.failed + ) + if failed: + scroll_compositor_factory.print_log() + ctx.__exit__(None, None, None) @pytest.fixture(scope="function") -def fresh_compositor( +def fresh_compositor_factory( scroll_compositor_binary: str, tmp_path: Path -) -> Generator[ScrollInstance, None, None]: +) -> Generator[ScrollCompositorFactory, None, None]: with run_compositor(scroll_compositor_binary, tmp_path) as inst: + yield ScrollCompositorFactory(inst) + + +@pytest.fixture(scope="function") +def fresh_compositor( + request: pytest.FixtureRequest, + fresh_compositor_factory: ScrollCompositorFactory, +) -> Generator[ScrollInstance, None, None]: + ctx = fresh_compositor_factory() + inst = ctx.__enter__() + try: yield inst + finally: + failed = ( + hasattr(request.node, "result_call") and request.node.result_call.failed + ) + if failed: + fresh_compositor_factory.print_log() + ctx.__exit__(None, None, None) diff --git a/tests/interactive.sh b/tests/interactive.sh index 9ec5c586..ac283af2 100755 --- a/tests/interactive.sh +++ b/tests/interactive.sh @@ -48,8 +48,8 @@ if [ "$has_config" = "false" ]; then set -- -c "$CONFIG" "$@" fi -# Enable leak detection but use suppressions for known system library leaks. -export ASAN_OPTIONS="detect_leaks=1${ASAN_OPTIONS:+:$ASAN_OPTIONS}" +# Disable leak detection to suppress leak errors in the interactive test. +export ASAN_OPTIONS="detect_leaks=0${ASAN_OPTIONS:+:$ASAN_OPTIONS}" export LSAN_OPTIONS="suppressions=$PROJECT_ROOT/tests/lsan.supp${LSAN_OPTIONS:+:$LSAN_OPTIONS}" echo "Starting scroll..." diff --git a/tests/meson.build b/tests/meson.build index 2b153d5d..0cf0d8b4 100644 --- a/tests/meson.build +++ b/tests/meson.build @@ -15,3 +15,6 @@ if xcb_dep.found() install: false, ) endif + + + diff --git a/tests/test_animation_offset.py b/tests/test_animation_offset.py deleted file mode 100644 index 7cd55108..00000000 --- a/tests/test_animation_offset.py +++ /dev/null @@ -1,129 +0,0 @@ -import time -from typing import Generator -from pathlib import Path -import pytest -from conftest import ScrollInstance -from test_utils import wayland_client, wait_for_client_map, run_compositor - - -@pytest.fixture(scope="function") -def offset_animating_compositor( - scroll_compositor_binary: str, tmp_path: Path -) -> Generator[ScrollInstance, None, None]: - config: str = ( - "workspace 1\n" - "xwayland force\n" - "animations enabled yes\n" - # 5 second animation, 10% offset scale - "animations window_move yes 5000 var 3 [ 0.215 0.61 0.355 1 ] off 0.1 6 [0 0.6 0.4 0 1 0 0.4 -0.6 1 -0.6]\n" - ) - with run_compositor(scroll_compositor_binary, tmp_path, config) as inst: - yield inst - - -def test_animation_offset_unintended_move( - offset_animating_compositor: ScrollInstance, -) -> None: - inst = offset_animating_compositor - - try: - with wayland_client(inst, "client1"): - wait_for_client_map(inst, "client1") - time.sleep(0.5) - with wayland_client(inst, "client2"): - wait_for_client_map(inst, "client2") - time.sleep(0.5) - with wayland_client(inst, "client3"): - wait_for_client_map(inst, "client3") - time.sleep(0.5) - - tree = inst.get_tree() - - def find_views(node, result): - if node.get("type") == "con" and node.get("name"): - if ( - node.get("window") - or node.get("app_id") - or ( - node.get("window_properties") - and node["window_properties"].get("title") - ) - ): - result.append( - { - "title": node["name"], - "id": node["id"], - "x": node["rect"]["x"], - } - ) - for child in node.get("nodes", []): - find_views(child, result) - for child in node.get("floating_nodes", []): - find_views(child, result) - - views = [] - find_views(tree, views) - print("Found views:", views) - assert len(views) == 3 - - # Sort by X to identify leftmost, middle, rightmost - sorted_views = sorted(views, key=lambda v: v["x"]) - v1, v2, v3 = sorted_views - print( - f"Leftmost: {v1['title']}, Middle: {v2['title']}, Rightmost: {v3['title']}" - ) - - # Focus rightmost (client3) to avoid scroll during swap - print("Focusing rightmost...") - res = inst.cmd(f"[con_id={v3['id']}] focus") - assert res[0]["success"] - time.sleep(1.0) # wait for focus scroll to settle - - # Swap leftmost and middle by moving leftmost right, using its ID as context - print("Swapping leftmost and middle in background...") - # We use execute_lua to run scroll.command with v1['id'] context - res_lua = inst.execute_lua( - f"return scroll.command({v1['id']}, 'move right')" - ) - print(f"Lua command result: {res_lua}") - - # Get parent container of client3 to query its actual position - parent_id = inst.execute_lua( - f"return scroll.container_get_parent({v3['id']})" - ) - query_id = parent_id if parent_id is not None else v3["id"] - print(f"Querying container {query_id} (parent of {v3['id']})") - - # Query position of client3 during animation - positions = [] - start_time = time.time() - while time.time() - start_time < 2.0: - geom = inst.execute_lua( - f"return scroll.container_get_animated_geometry({query_id})" - ) - positions.append(geom) - time.sleep(0.05) - - print(f"Positions of client3: {positions}") - - # Verify positions are constant - x_values = [p["x"] for p in positions] - y_values = [p["y"] for p in positions] - - first_x = x_values[0] - first_y = y_values[0] - for x in x_values: - assert x == pytest.approx(first_x, abs=1.0), ( - f"client3 moved horizontally: {x_values}" - ) - for y in y_values: - assert y == pytest.approx(first_y, abs=1.0), ( - f"client3 moved vertically: {y_values}" - ) - - print("Test passed: static window did not move.") - - except Exception as e: - print("Scroll Log on failure:") - print(inst.read_log()) - raise e diff --git a/tests/test_clients.py b/tests/test_clients.py index b29a7c69..aa8d4706 100644 --- a/tests/test_clients.py +++ b/tests/test_clients.py @@ -1,9 +1,9 @@ import os -import subprocess -import time from pathlib import Path -import pytest +import subprocess from conftest import ScrollInstance +import pytest +from test_utils import wait_for_client_map def test_wayland_client(scroll_compositor: ScrollInstance) -> None: @@ -23,31 +23,12 @@ def test_wayland_client(scroll_compositor: ScrollInstance) -> None: [str(client_path), title, app_id], env=env ) - view_info: dict | None = None - tries: int = 0 - while tries < 50: - view_info = scroll_compositor.execute_lua(""" - local view = scroll.focused_view() - if view then - return { - id = view, - title = scroll.view_get_title(view), - app_id = scroll.view_get_app_id(view) - } - end - """) - if ( - view_info - and view_info.get("title") == title - and view_info.get("app_id") == app_id - ): - break - - time.sleep(0.1) - tries += 1 - - assert tries < 50, "Timed out waiting for client to map or verify" - assert view_info is not None + view_id = wait_for_client_map(scroll_compositor, title) + app_id_actual = scroll_compositor.execute_lua( + f"return scroll.view_get_app_id({view_id})" + ) + assert app_id_actual == app_id + view_info = {"id": view_id} scroll_compositor.execute_lua(f"scroll.view_close({view_info['id']})") @@ -68,13 +49,7 @@ def test_x11_client(scroll_compositor: ScrollInstance) -> None: xauthority: str | None = scroll_compositor.getenv("XAUTHORITY") # Wait for Xwayland to be ready - xwayland_ready_tries: int = 0 - while xwayland_ready_tries < 50: - if "Xserver is ready" in scroll_compositor.read_log(): - break - time.sleep(0.1) - xwayland_ready_tries += 1 - assert xwayland_ready_tries < 50, "Timed out waiting for Xwayland to be ready" + scroll_compositor.wait_for_log_pattern("Xserver is ready", from_start=True) client_path: Path = Path("./build/tests/x11-test-client").resolve() if not client_path.exists(): @@ -93,36 +68,16 @@ def test_x11_client(scroll_compositor: ScrollInstance) -> None: [str(client_path), title, instance, class_name], env=env ) - view_info: dict | None = None - tries: int = 0 - while tries < 50: - view_info = scroll_compositor.execute_lua(""" - local view = scroll.focused_view() - if view then - return { - id = view, - title = scroll.view_get_title(view), - class = scroll.view_get_class(view), - shell = scroll.view_get_shell(view) - } - end - """) - if ( - view_info - and view_info.get("title") == title - and view_info.get("class") == class_name - ): - assert view_info.get("shell") == "xwayland" - break - - time.sleep(0.1) - tries += 1 - - if tries >= 50: - Path("build/test_x11_compositor.log").write_text(scroll_compositor.read_log()) - print("Wrote compositor log to build/test_x11_compositor.log") - assert tries < 50, "Timed out waiting for X11 client to map or verify" - assert view_info is not None + view_id = wait_for_client_map(scroll_compositor, title) + class_actual = scroll_compositor.execute_lua( + f"return scroll.view_get_class({view_id})" + ) + shell_actual = scroll_compositor.execute_lua( + f"return scroll.view_get_shell({view_id})" + ) + assert class_actual == class_name + assert shell_actual == "xwayland" + view_info = {"id": view_id} scroll_compositor.execute_lua(f"scroll.view_close({view_info['id']})") diff --git a/tests/test_crash_move.py b/tests/test_crash_move.py index 5250a77a..67840b64 100644 --- a/tests/test_crash_move.py +++ b/tests/test_crash_move.py @@ -1,5 +1,4 @@ -from conftest import ScrollInstance -from test_utils import wayland_client, wait_for_client_map +from test_utils import wayland_client, wait_for_client_map, ScrollInstance def test_move_left_nomode_no_crash(scroll_compositor: ScrollInstance) -> None: @@ -12,12 +11,8 @@ def test_move_left_nomode_no_crash(scroll_compositor: ScrollInstance) -> None: wait_for_client_map(scroll_compositor, "Window 2") # Both windows are open. Now move left nomode. - try: - res = scroll_compositor.cmd("move left nomode") - assert res and res[0]["success"], f"Command failed: {res}" - except Exception as e: - print(f"Compositor log:\n{scroll_compositor.read_log()}") - raise e + res = scroll_compositor.cmd("move left nomode") + assert res and res[0]["success"], f"Command failed: {res}" # Check if compositor process is still alive assert scroll_compositor.proc.poll() is None, "Compositor crashed" diff --git a/tests/test_focused_inactive_null_crash.py b/tests/test_focused_inactive_null_crash.py index c38f5bdb..c13c1eef 100644 --- a/tests/test_focused_inactive_null_crash.py +++ b/tests/test_focused_inactive_null_crash.py @@ -1,7 +1,5 @@ -import time import json -from conftest import ScrollInstance -from test_utils import wayland_client, wait_for_client_map +from test_utils import wayland_client, wait_for_client_map, ScrollInstance def test_focused_inactive_null_crash(fresh_compositor: ScrollInstance) -> None: @@ -35,7 +33,7 @@ def test_focused_inactive_null_crash(fresh_compositor: ScrollInstance) -> None: # 5. Move w1 to workspace 2 (this detaches it, making col's focused_inactive_child NULL) fresh_compositor.cmd(f"[con_id={w1_id}] move container to workspace 2") - time.sleep(0.1) + fresh_compositor.wait_for_idle() # 6. Focus the parent column print(f"col_id: {col_id}") @@ -69,6 +67,3 @@ def test_focused_inactive_null_crash(fresh_compositor: ScrollInstance) -> None: ret = fresh_compositor.proc.wait(timeout=5) print(f"Compositor exit code: {ret}") assert ret == 0, f"Compositor crashed or exited with error code {ret}" - - print("Compositor Log:") - print(fresh_compositor.read_log()) diff --git a/tests/test_focused_inactive_uaf.py b/tests/test_focused_inactive_uaf.py index 1a19e5fc..8ca5ebb4 100644 --- a/tests/test_focused_inactive_uaf.py +++ b/tests/test_focused_inactive_uaf.py @@ -1,36 +1,35 @@ -import time from conftest import ScrollInstance from test_utils import wayland_client, wait_for_client_map -def test_focused_inactive_uaf(fresh_compositor: ScrollInstance) -> None: +def test_focused_inactive_uaf(scroll_compositor: ScrollInstance) -> None: # 1. Set mode to vertical to stack windows - fresh_compositor.cmd("set_mode v") + scroll_compositor.cmd("set_mode v") # 2. Create first view - with wayland_client(fresh_compositor, "client1") as client1: - wait_for_client_map(fresh_compositor, "client1") - w1_id = fresh_compositor.execute_lua("return scroll.focused_container()") + with wayland_client(scroll_compositor, "client1") as client1: + wait_for_client_map(scroll_compositor, "client1") + w1_id = scroll_compositor.execute_lua("return scroll.focused_container()") print(f"w1_id: {w1_id}") # 3. Create second view - with wayland_client(fresh_compositor, "client2"): - wait_for_client_map(fresh_compositor, "client2") - w2_id = fresh_compositor.execute_lua("return scroll.focused_container()") + with wayland_client(scroll_compositor, "client2"): + wait_for_client_map(scroll_compositor, "client2") + w2_id = scroll_compositor.execute_lua("return scroll.focused_container()") print(f"w2_id: {w2_id}") # Verify they are siblings (same parent) - w1_parent = fresh_compositor.execute_lua( + w1_parent = scroll_compositor.execute_lua( f"return scroll.container_get_parent({w1_id})" ) - w2_parent = fresh_compositor.execute_lua( + w2_parent = scroll_compositor.execute_lua( f"return scroll.container_get_parent({w2_id})" ) print(f"w1_parent: {w1_parent}, w2_parent: {w2_parent}") assert w1_parent == w2_parent, "w1 and w2 should be siblings" # Get Col1 ID (parent ID) - col1_id = fresh_compositor.execute_lua(""" + col1_id = scroll_compositor.execute_lua(""" local outputs = scroll.root_get_outputs() local workspaces = scroll.output_get_workspaces(outputs[1]) local ws1 @@ -47,29 +46,28 @@ def test_focused_inactive_uaf(fresh_compositor: ScrollInstance) -> None: assert col1_id == w1_parent, "Col1 ID should match parent ID" # 4. Focus w1 to make it the focused_inactive_child of the parent - fresh_compositor.cmd(f"[con_id={w1_id}] focus") - focused = fresh_compositor.execute_lua("return scroll.focused_container()") + scroll_compositor.cmd(f"[con_id={w1_id}] focus") + focused = scroll_compositor.execute_lua("return scroll.focused_container()") assert focused == w1_id, ( f"w1 ({w1_id}) should be focused, but got {focused}" ) # 5. Switch to workspace 2 - fresh_compositor.cmd("workspace 2") + scroll_compositor.cmd("workspace 2") # 6. Kill w1 (which is on workspace 1) - fresh_compositor.cmd(f"[con_id={w1_id}] kill") + scroll_compositor.cmd(f"[con_id={w1_id}] kill") - # Wait for client1 to exit (destruction complete) client1.wait(timeout=5) - time.sleep(0.1) + scroll_compositor.wait_for_idle() # 7. Switch back to workspace 1. - fresh_compositor.cmd("workspace 1") + scroll_compositor.cmd("workspace 1") # 8. Run move command targeted at Col1 (which has dangling focused_inactive_child) # This should trigger UAF in container_get_active_view! - fresh_compositor.cmd(f"[con_id={col1_id}] move left nomode") + scroll_compositor.cmd(f"[con_id={col1_id}] move left nomode") # If we survive, let's verify w2 is still there - focused = fresh_compositor.execute_lua("return scroll.focused_container()") + focused = scroll_compositor.execute_lua("return scroll.focused_container()") print(f"Focused after move: {focused}") diff --git a/tests/test_geometry.py b/tests/test_geometry.py index 0ce9e57e..e01273c1 100644 --- a/tests/test_geometry.py +++ b/tests/test_geometry.py @@ -1,24 +1,5 @@ -import time -from typing import Generator -from pathlib import Path -import pytest from conftest import ScrollInstance -from test_utils import wayland_client, wait_for_client_map, run_compositor - - -@pytest.fixture(scope="function") -def animating_compositor( - scroll_compositor_binary: str, tmp_path: Path -) -> Generator[ScrollInstance, None, None]: - config: str = ( - "workspace 1\n" - "xwayland force\n" - "animations enabled yes\n" - # 2 second animation for window_move - "animations window_move yes 2000 linear\n" - ) - with run_compositor(scroll_compositor_binary, tmp_path, config) as inst: - yield inst +from test_utils import wayland_client, wait_for_client_map def test_static_geometry(scroll_compositor: ScrollInstance) -> None: @@ -29,7 +10,7 @@ def test_static_geometry(scroll_compositor: ScrollInstance) -> None: assert con_id is not None # Wait for any map animation to settle - time.sleep(2.0) + inst.wait_for_idle() geom = inst.execute_lua(f"return scroll.container_get_geometry({con_id})") actual_geom = inst.execute_lua( @@ -89,79 +70,6 @@ def find_node(node, target_id): assert geom["height"] == expected_height -def test_animating_geometry(animating_compositor: ScrollInstance) -> None: - inst = animating_compositor - with wayland_client(inst, "client1"): - v1 = wait_for_client_map(inst, "client1") - c1 = inst.execute_lua(f"return scroll.view_get_container({v1})") - - with wayland_client(inst, "client2"): - wait_for_client_map(inst, "client2") - - # Wait for initial map animations to settle - time.sleep(0.5) - - geom_before = inst.execute_lua( - f"return scroll.container_get_geometry({c1})" - ) - actual_geom_before = inst.execute_lua( - f"return scroll.container_get_animated_geometry({c1})" - ) - assert geom_before == actual_geom_before - - print("Before move:", geom_before) - - # Trigger move - inst.execute_lua(f"scroll.command({c1}, 'move right')") - - # Immediately query geometry - geom_after_trigger = inst.execute_lua( - f"return scroll.container_get_geometry({c1})" - ) - actual_geom_after_trigger = inst.execute_lua( - f"return scroll.container_get_animated_geometry({c1})" - ) - - print("Immediately after trigger (target):", geom_after_trigger) - print("Immediately after trigger (actual):", actual_geom_after_trigger) - - # The target geometry (geom) should have jumped to the final position. - # The actual geometry should still be close to the initial position. - assert geom_after_trigger != geom_before - assert actual_geom_after_trigger["x"] == pytest.approx( - actual_geom_before["x"], abs=10.0 - ) - - # Monitor animation - actual_xs = [] - target_xs = [] - start_time = time.time() - while time.time() - start_time < 2.5: # Animation is 2s - g = inst.execute_lua(f"return scroll.container_get_geometry({c1})") - ag = inst.execute_lua( - f"return scroll.container_get_animated_geometry({c1})" - ) - target_xs.append(g["x"]) - actual_xs.append(ag["x"]) - time.sleep(0.1) - - print("Target Xs:", target_xs) - print("Actual Xs:", actual_xs) - - # Target Xs should all be equal to the final position - final_x = geom_after_trigger["x"] - for tx in target_xs: - assert tx == final_x - - # Actual Xs should start near before_x and end at final_x - assert actual_xs[0] < final_x # Assuming it moved right - assert actual_xs[-1] == pytest.approx(final_x, abs=1.0) - - # Verify it is monotonically increasing (if it moved right) - for i in range(1, len(actual_xs)): - assert actual_xs[i] >= actual_xs[i - 1] - 0.1 - - def test_invalid_geometry(scroll_compositor: ScrollInstance) -> None: inst = scroll_compositor # Invalid ID diff --git a/tests/test_leak_two_clients.py b/tests/test_leak_two_clients.py new file mode 100644 index 00000000..dd760904 --- /dev/null +++ b/tests/test_leak_two_clients.py @@ -0,0 +1,37 @@ +from pathlib import Path +import subprocess +import time +from test_utils import ( + run_compositor, + wait_for_client_map, + wayland_client, + ScrollCompositorFactory, +) + + +def test_leak_two_clients(scroll_compositor_binary: str, tmp_path: Path) -> None: + config_path = Path(__file__).parent.parent / "config.in" + config_content = config_path.read_text() + + with run_compositor(scroll_compositor_binary, tmp_path, config_content) as inst: + factory = ScrollCompositorFactory(inst) + with factory() as fresh_compositor: + # Start two clients and keep them running + with wayland_client(fresh_compositor, "client1"): + wait_for_client_map(fresh_compositor, "client1") + + with wayland_client(fresh_compositor, "client2"): + wait_for_client_map(fresh_compositor, "client2") + + # Let them run a bit + time.sleep(0.5) + + # Terminate compositor while clients are still running + fresh_compositor.proc.terminate() + try: + ret = fresh_compositor.proc.wait(timeout=5) + except subprocess.TimeoutExpired: + fresh_compositor.proc.kill() + ret = fresh_compositor.proc.wait() + + assert ret == 0, f"Compositor exited with code {ret}" diff --git a/tests/test_lua_path.py b/tests/test_lua_path.py index 2fb6461b..b0d0c69f 100644 --- a/tests/test_lua_path.py +++ b/tests/test_lua_path.py @@ -1,7 +1,4 @@ -import time -from pathlib import Path -from conftest import ScrollInstance -from test_utils import run_compositor +from test_utils import ScrollInstance, ScrollCompositorFactory def test_lua_tilde_expansion(scroll_compositor: ScrollInstance) -> None: @@ -28,58 +25,58 @@ def test_lua_tilde_expansion(scroll_compositor: ScrollInstance) -> None: def test_lua_relative_path_config_load( - scroll_compositor: ScrollInstance, tmp_path: Path + scroll_compositor_factory: ScrollCompositorFactory, ) -> None: - binary_path = scroll_compositor.proc.args[0] - script_path = tmp_path / "test_relative.lua" - script_path.write_text('scroll.log("RELATIVE_LOAD_SUCCESS")') + with scroll_compositor_factory() as scroll_compositor: + home_dir = scroll_compositor.temp_dir + script_path = home_dir / "test_relative.lua" + script_path.write_text('scroll.log("RELATIVE_LOAD_SUCCESS")') - config = "workspace 1\nxwayland force\nlua test_relative.lua\n" - with run_compositor(binary_path, tmp_path, config) as inst: - time.sleep(1.0) - assert "RELATIVE_LOAD_SUCCESS" in inst.read_log() + config = "workspace 1\nxwayland force\nlua test_relative.lua\n" + with scroll_compositor.assert_logs_match("RELATIVE_LOAD_SUCCESS"): + scroll_compositor.reload_config(config) def test_lua_relative_path_subdir_config_load( - scroll_compositor: ScrollInstance, tmp_path: Path + scroll_compositor_factory: ScrollCompositorFactory, ) -> None: - binary_path = scroll_compositor.proc.args[0] - subdir = tmp_path / "scripts" - subdir.mkdir() - script_path = subdir / "test_relative2.lua" - script_path.write_text('scroll.log("RELATIVE_SUBDIR_LOAD_SUCCESS")') + with scroll_compositor_factory() as scroll_compositor: + home_dir = scroll_compositor.temp_dir + subdir = home_dir / "scripts" + subdir.mkdir(exist_ok=True) + script_path = subdir / "test_relative2.lua" + script_path.write_text('scroll.log("RELATIVE_SUBDIR_LOAD_SUCCESS")') - config = "workspace 1\nxwayland force\nlua scripts/test_relative2.lua\n" - with run_compositor(binary_path, tmp_path, config) as inst: - time.sleep(1.0) - assert "RELATIVE_SUBDIR_LOAD_SUCCESS" in inst.read_log() + config = "workspace 1\nxwayland force\nlua scripts/test_relative2.lua\n" + with scroll_compositor.assert_logs_match("RELATIVE_SUBDIR_LOAD_SUCCESS"): + scroll_compositor.reload_config(config) def test_lua_relative_glob_config_load( - scroll_compositor: ScrollInstance, tmp_path: Path + scroll_compositor_factory: ScrollCompositorFactory, ) -> None: - binary_path = scroll_compositor.proc.args[0] - subdir = tmp_path / "scripts" - subdir.mkdir() - script_path = subdir / "test_glob1.lua" - script_path.write_text('scroll.log("RELATIVE_GLOB_LOAD_SUCCESS")') + with scroll_compositor_factory() as scroll_compositor: + home_dir = scroll_compositor.temp_dir + subdir = home_dir / "scripts" + subdir.mkdir(exist_ok=True) + script_path = subdir / "test_glob1.lua" + script_path.write_text('scroll.log("RELATIVE_GLOB_LOAD_SUCCESS")') - config = "workspace 1\nxwayland force\nlua scripts/test_glob*.lua\n" - with run_compositor(binary_path, tmp_path, config) as inst: - time.sleep(1.0) - assert "RELATIVE_GLOB_LOAD_SUCCESS" in inst.read_log() + config = "workspace 1\nxwayland force\nlua scripts/test_glob*.lua\n" + with scroll_compositor.assert_logs_match("RELATIVE_GLOB_LOAD_SUCCESS"): + scroll_compositor.reload_config(config) def test_lua_relative_glob_multiple_config_load( - scroll_compositor: ScrollInstance, tmp_path: Path + scroll_compositor_factory: ScrollCompositorFactory, ) -> None: - binary_path = scroll_compositor.proc.args[0] - subdir = tmp_path / "scripts" - subdir.mkdir() - (subdir / "test_glob1.lua").write_text('scroll.log("G1")') - (subdir / "test_glob2.lua").write_text('scroll.log("G2")') - - config = "workspace 1\nxwayland force\nlua scripts/test_glob*.lua\n" - with run_compositor(binary_path, tmp_path, config) as inst: - time.sleep(1.0) - assert "Path expanded to multiple files" in inst.read_log() + with scroll_compositor_factory() as scroll_compositor: + home_dir = scroll_compositor.temp_dir + subdir = home_dir / "scripts" + subdir.mkdir(exist_ok=True) + (subdir / "test_glob1.lua").write_text('scroll.log("G1")') + (subdir / "test_glob2.lua").write_text('scroll.log("G2")') + + config = "workspace 1\nxwayland force\nlua scripts/test_glob*.lua\n" + with scroll_compositor.assert_logs_match("Path expanded to multiple files"): + scroll_compositor.reload_config(config) diff --git a/tests/test_lua_safety.py b/tests/test_lua_safety.py index 7911868a..9b7d3930 100644 --- a/tests/test_lua_safety.py +++ b/tests/test_lua_safety.py @@ -12,6 +12,7 @@ def test_lua_use_after_free_prevented(scroll_compositor: ScrollInstance) -> None # 3. Switch back to workspace 1 (workspace 2 should be destroyed if empty) scroll_compositor.execute_lua('scroll.command(nil, "workspace 1")') + scroll_compositor.wait_for_idle() # 4. Now try to get name of ws2. It should return nil (None in Python), but NOT crash. ws2_name = scroll_compositor.execute_lua(f"return scroll.workspace_get_name({ws2})") diff --git a/tests/test_move_cleanup_crash.py b/tests/test_move_cleanup_crash.py index e7d9949e..514cc1ae 100644 --- a/tests/test_move_cleanup_crash.py +++ b/tests/test_move_cleanup_crash.py @@ -1,42 +1,26 @@ -import time -from conftest import ScrollInstance -from test_utils import wayland_client, wait_for_client_map +from test_utils import wayland_client, wait_for_client_map, ScrollInstance -def test_move_cleanup_uaf_crash(fresh_compositor: ScrollInstance) -> None: +def test_move_cleanup_uaf_crash(scroll_compositor: ScrollInstance) -> None: # 1. Open Window 1 on Workspace 1 - with wayland_client(fresh_compositor, "Window 1"): - wait_for_client_map(fresh_compositor, "Window 1") + with wayland_client(scroll_compositor, "Window 1"): + wait_for_client_map(scroll_compositor, "Window 1") # Make Window 1 floating - res = fresh_compositor.cmd("floating enable") + res = scroll_compositor.cmd("floating enable") assert res and res[0]["success"], f"floating enable failed: {res}" # 2. Switch to Workspace 3 (so Workspace 1 becomes inactive) - res = fresh_compositor.cmd("workspace 3") + res = scroll_compositor.cmd("workspace 3") assert res and res[0]["success"], f"workspace 3 failed: {res}" - time.sleep(0.5) + scroll_compositor.wait_for_idle() # 3. Use criteria to move Window 1 from Workspace 1 to Workspace 2. # This should make Workspace 1 empty and inactive, so it gets destroyed. # Then the move command cleanup code should UAF on Workspace 1. - try: - res = fresh_compositor.cmd( - '[title="Window 1"] move container to workspace 2' - ) - assert res and res[0]["success"], f"move failed: {res}" - except Exception as e: - print(f"Compositor log:\n{fresh_compositor.read_log()}") - raise e + res = scroll_compositor.cmd('[title="Window 1"] move container to workspace 2') + assert res and res[0]["success"], f"move failed: {res}" # Check if compositor process is still alive - log_content = fresh_compositor.read_log() - print(f"Compositor log:\n{log_content}") - if ( - fresh_compositor.proc.poll() is not None - or "node_table not initialized" in log_content - ): - pass - - assert fresh_compositor.proc.poll() is None, "Compositor crashed" + assert scroll_compositor.proc.poll() is None, "Compositor crashed" diff --git a/tests/test_normal_exit.py b/tests/test_normal_exit.py index 0a51e8e7..302fd078 100644 --- a/tests/test_normal_exit.py +++ b/tests/test_normal_exit.py @@ -1,5 +1,7 @@ +from pathlib import Path +import subprocess import time -from conftest import ScrollInstance +from test_utils import run_compositor, ScrollCompositorFactory, ScrollInstance def test_normal_exit_no_errors(fresh_compositor: ScrollInstance) -> None: @@ -10,25 +12,46 @@ def test_normal_exit_no_errors(fresh_compositor: ScrollInstance) -> None: except EOFError: # Expected if compositor exits before flushing reply pass - except Exception as e: - print(f"Compositor log:\n{fresh_compositor.read_log()}") - raise e # Wait for compositor to exit - tries = 0 - poll = None - while tries < 50: - poll = fresh_compositor.proc.poll() - if poll is not None: - break - time.sleep(0.1) - tries += 1 + try: + poll = fresh_compositor.proc.wait(timeout=5) + except subprocess.TimeoutExpired: + poll = None assert poll is not None, "Compositor did not exit" log_content = fresh_compositor.read_log() - if poll != 0 or "node_table not initialized" in log_content: - print(f"Compositor log:\n{log_content}") - assert poll == 0, f"Compositor exited with non-zero code: {poll}" assert "node_table not initialized" not in log_content + + +def test_normal_exit_with_bar(scroll_compositor_binary: str, tmp_path: Path) -> None: + config_content = """ +workspace 1 +xwayland force +animations enabled no +bar { + scrollbar_command scrollbar +} +""" + with run_compositor(scroll_compositor_binary, tmp_path, config_content) as inst: + factory = ScrollCompositorFactory(inst) + with factory() as fresh_compositor: + # Let it run a bit to ensure swaybar starts + time.sleep(0.5) + + # Send exit command + try: + res = fresh_compositor.cmd("exit") + assert res and res[0]["success"], f"Exit command failed: {res}" + except EOFError: + pass + + ret = fresh_compositor.proc.wait(timeout=5) + assert ret == 0, f"Compositor exited with code {ret}" + + log_content = fresh_compositor.read_log() + assert "ERROR: LeakSanitizer" not in log_content, ( + "Leak detected in compositor or helper process" + ) diff --git a/tests/test_rapid_switch.py b/tests/test_rapid_switch.py new file mode 100644 index 00000000..fee1cec1 --- /dev/null +++ b/tests/test_rapid_switch.py @@ -0,0 +1,22 @@ +import time +from test_utils import ScrollCompositorFactory + + +def test_rapid_switch(scroll_compositor_factory: ScrollCompositorFactory) -> None: + config = ( + "workspace 1\nanimations enabled yes\nanimations workspace_switch yes 1000\n" + ) + with scroll_compositor_factory(config) as scroll_compositor: + scroll_compositor.cmd("workspace 2") + scroll_compositor.cmd("workspace 1") + + # Check that it IS animating immediately after the switch back + is_animating: bool = scroll_compositor.execute_lua("return scroll.animating()") + assert is_animating is True, "Expected animation to be running" + + # Wait for animation to finish (duration is 1s, wait 1.5s to be safe) + time.sleep(1.5) + + # Check that it finished + is_animating = scroll_compositor.execute_lua("return scroll.animating()") + assert is_animating is False, "Expected animation to have finished" diff --git a/tests/test_shutdown_events.py b/tests/test_shutdown_events.py new file mode 100644 index 00000000..f6e7f27f --- /dev/null +++ b/tests/test_shutdown_events.py @@ -0,0 +1,118 @@ +import json +from pathlib import Path +import socket +import struct +from test_utils import ( + run_compositor, + wait_for_client_map, + wayland_client, + ScrollCompositorFactory, +) + + +def test_shutdown_events_verification( + scroll_compositor_binary: str, tmp_path: Path +) -> None: + config_path: Path = Path(__file__).parent.parent / "config.in" + config_content: str = config_path.read_text() + + with run_compositor(scroll_compositor_binary, tmp_path, config_content) as inst: + factory = ScrollCompositorFactory(inst) + with factory() as fc: + # Connect a second socket for events subscription + event_socket: socket.socket = socket.socket( + socket.AF_UNIX, socket.SOCK_STREAM + ) + event_socket.connect(fc.ipc.socket_path) + + # Helper functions for the custom subscription socket + def send_msg(msg_type: int, payload: str) -> None: + payload_bytes: bytes = payload.encode("utf-8") + length: int = len(payload_bytes) + header: bytes = struct.pack("<6sII", b"i3-ipc", length, msg_type) + event_socket.sendall(header + payload_bytes) + + def recv_msg() -> tuple[int, str]: + header_data: bytes = b"" + while len(header_data) < 14: + chunk: bytes = event_socket.recv(14 - len(header_data)) + if not chunk: + raise EOFError("Socket closed") + header_data += chunk + magic: bytes + length: int + msg_type: int + magic, length, msg_type = struct.unpack("<6sII", header_data) + payload_data: bytes = b"" + while len(payload_data) < length: + chunk = event_socket.recv(length - len(payload_data)) + if not chunk: + raise EOFError("Socket closed") + payload_data += chunk + return msg_type, payload_data.decode("utf-8") + + # Subscribe to workspace, window, and shutdown events + # 2 is IPC_SUBSCRIBE + send_msg(2, json.dumps(["workspace", "window", "shutdown"])) + msg_type, payload = recv_msg() + assert msg_type == 2 + assert json.loads(payload)["success"] is True + + # Start two clients to have some active views/workspaces + with wayland_client(fc, "client1"): + wait_for_client_map(fc, "client1") + with wayland_client(fc, "client2"): + wait_for_client_map(fc, "client2") + + # Drain all pending events on subscription socket before terminating + event_socket.setblocking(False) + try: + while True: + header_data: bytes = event_socket.recv(14) + if len(header_data) == 14: + magic, length, msg_type = struct.unpack( + "<6sII", header_data + ) + payload_data = b"" + event_socket.setblocking(True) + while len(payload_data) < length: + chunk = event_socket.recv(length - len(payload_data)) + if not chunk: + break + payload_data += chunk + event_socket.setblocking(False) + except BlockingIOError: + pass + + event_socket.setblocking(True) + + # Now terminate the compositor by sending the exit command + try: + fc.cmd("exit") + except (EOFError, BrokenPipeError, ConnectionResetError): + # The socket might close immediately during exit processing, which is fine + pass + + # Read all events sent during shutdown until socket EOF + shutdown_events: list[tuple[int, dict]] = [] + try: + while True: + msg_type, payload = recv_msg() + shutdown_events.append((msg_type, json.loads(payload))) + except (EOFError, BrokenPipeError, ConnectionResetError): + pass # EOF or connection error is expected when compositor exits + + print(f"Events received during shutdown: {shutdown_events}") + + shutdown_msg_type: int = (1 << 31) | 6 + # Verify that if any events are received, they are only shutdown events and nothing unexpected + unexpected_events: list[tuple[int, dict]] = [] + for mtype, p in shutdown_events: + if mtype == shutdown_msg_type: + assert p["change"] == "exit" + else: + unexpected_events.append((mtype, p)) + + assert not unexpected_events, ( + f"Received unexpected events: {unexpected_events}" + ) diff --git a/tests/test_shutdown_lua_callbacks.py b/tests/test_shutdown_lua_callbacks.py new file mode 100644 index 00000000..6f135e99 --- /dev/null +++ b/tests/test_shutdown_lua_callbacks.py @@ -0,0 +1,89 @@ +from pathlib import Path +import re +from test_utils import ( + run_compositor, + wait_for_client_map, + wayland_client, + ScrollCompositorFactory, +) + + +def test_shutdown_lua_callbacks_verification( + scroll_compositor_binary: str, tmp_path: Path +) -> None: + config_path: Path = Path(__file__).parent.parent / "config.in" + config_content: str = config_path.read_text() + + with run_compositor(scroll_compositor_binary, tmp_path, config_content) as inst: + factory = ScrollCompositorFactory(inst) + with factory() as fc: + # Register callbacks for all events, logging them to the scroll debug log + res = fc.execute_lua(""" + scroll.add_callback("view_map", function(view, data) + scroll.log("LUA_CALLBACK: view_map " .. tostring(view)) + end, nil) + scroll.add_callback("view_unmap", function(view, data) + scroll.log("LUA_CALLBACK: view_unmap " .. tostring(view)) + end, nil) + scroll.add_callback("view_focus", function(view, data) + scroll.log("LUA_CALLBACK: view_focus " .. tostring(view)) + end, nil) + scroll.add_callback("workspace_create", function(ws, data) + scroll.log("LUA_CALLBACK: workspace_create " .. tostring(ws)) + end, nil) + scroll.add_callback("workspace_focus", function(ws, data) + scroll.log("LUA_CALLBACK: workspace_focus " .. tostring(ws)) + end, nil) + """) + print(f"execute_lua result: {res}") + + # Start two clients to populate windows and workspaces + with wayland_client(fc, "client1"): + wait_for_client_map(fc, "client1") + with wayland_client(fc, "client2"): + wait_for_client_map(fc, "client2") + + # Record the log length before exit command + log_before_exit: str = fc.read_log() + log_len_before_exit: int = len(log_before_exit) + + # Now terminate the compositor via the exit command + try: + fc.cmd("exit") + except (EOFError, BrokenPipeError, ConnectionResetError): + pass + + # Wait for compositor to exit + fc.proc.wait(timeout=5) + + # Read all new log lines generated during shutdown + full_log: str = fc.read_log() + shutdown_log: str = full_log[log_len_before_exit:] + + # Extract all callback events triggered during shutdown + callback_pattern = re.compile(r"LUA_CALLBACK: (\w+)") + triggered_callbacks: list[str] = callback_pattern.findall( + shutdown_log + ) + + print(f"Triggered callbacks during shutdown: {triggered_callbacks}") + + # During shutdown, we unmap the existing views, so 'view_unmap' callbacks are expected. + # However, we should NOT see any new focus events ('view_focus', 'workspace_focus') + # or creation events ('workspace_create') being invoked. + unexpected_callbacks: list[str] = [ + cb + for cb in triggered_callbacks + if cb + in ( + "view_map", + "view_focus", + "workspace_create", + "workspace_focus", + ) + ] + + assert not unexpected_callbacks, ( + "Unexpected Lua callbacks invoked during shutdown:" + f" {unexpected_callbacks}" + ) diff --git a/tests/test_space_aba.py b/tests/test_space_aba.py index c01e83e3..270edc76 100644 --- a/tests/test_space_aba.py +++ b/tests/test_space_aba.py @@ -1,41 +1,43 @@ -import time from conftest import ScrollInstance from test_utils import wayland_client, wait_for_client_map -def test_space_aba(fresh_compositor: ScrollInstance) -> None: - # 1. Create client1 on WS 1 - with wayland_client(fresh_compositor, "client1"): - wait_for_client_map(fresh_compositor, "client1") - - # Save space "sp1" - fresh_compositor.cmd("space_save sp1") - - # client1 is closed now. - # Wait for it to be fully destroyed. - time.sleep(0.2) - - # 2. Create client2 on WS 1. - # Hopefully it reuses client1's view struct address. - with wayland_client(fresh_compositor, "client2"): - wait_for_client_map(fresh_compositor, "client2") - - # Switch to WS 2 - fresh_compositor.cmd("workspace 2") - - # Load space "sp1" on WS 2. - # If ABA bug occurs, it might find client2 (matching old client1 address) - # and move it to WS 2. - fresh_compositor.cmd("space_load sp1 load") - - # Check if client2 is visible on WS 2. - # If it was moved to WS 2, it should be focused because space_load focuses restored containers. - focused_title = fresh_compositor.execute_lua(""" - local view = scroll.focused_view() - return view and scroll.view_get_title(view) - """) - print(f"Focused title after space_load: {focused_title}") - - assert focused_title != "client2", ( - "ABA bug: client2 was incorrectly moved to WS 2!" - ) +def test_space_aba(scroll_compositor: ScrollInstance) -> None: + inst = scroll_compositor + try: + # 1. Create client1 on WS 1 + with wayland_client(inst, "client1"): + wait_for_client_map(inst, "client1") + + # Save space "sp1" + inst.cmd("space_save sp1") + + # client1 is closed now. + inst.wait_for_idle() + + # 2. Create client2 on WS 1. + # Hopefully it reuses client1's view struct address. + with wayland_client(inst, "client2"): + wait_for_client_map(inst, "client2") + + # Switch to WS 2 + inst.cmd("workspace 2") + + # Load space "sp1" on WS 2. + # If ABA bug occurs, it might find client2 (matching old client1 address) + # and move it to WS 2. + inst.cmd("space_load sp1 load") + + # Check if client2 is visible on WS 2. + # If it was moved to WS 2, it should be focused because space_load focuses restored containers. + focused_title = inst.execute_lua(""" + local view = scroll.focused_view() + return view and scroll.view_get_title(view) + """) + print(f"Focused title after space_load: {focused_title}") + + assert focused_title != "client2", ( + "ABA bug: client2 was incorrectly moved to WS 2!" + ) + finally: + inst.cmd("space delete sp1") diff --git a/tests/test_space_crash.py b/tests/test_space_crash.py index 6e35559e..ca0a2f45 100644 --- a/tests/test_space_crash.py +++ b/tests/test_space_crash.py @@ -1,83 +1,66 @@ -from conftest import ScrollInstance -from test_utils import wayland_client, wait_for_client_map +from test_utils import wayland_client, wait_for_client_map, ScrollInstance -def test_space_restore_uaf_crash(fresh_compositor: ScrollInstance) -> None: +def test_space_restore_uaf_crash(scroll_compositor: ScrollInstance) -> None: + inst = scroll_compositor # 1. Open Window 1 on Workspace 1 - with wayland_client(fresh_compositor, "Window 1"): - wait_for_client_map(fresh_compositor, "Window 1") + with wayland_client(inst, "Window 1"): + wait_for_client_map(inst, "Window 1") # Make Window 1 floating - res = fresh_compositor.cmd("floating enable") + res = inst.cmd("floating enable") assert res and res[0]["success"], f"floating enable failed: {res}" # Save layout "space1" on Workspace 1 - res = fresh_compositor.cmd("space save space1") + res = inst.cmd("space save space1") assert res and res[0]["success"], f"space save failed: {res}" # 2. Switch to Workspace 2 - res = fresh_compositor.cmd("workspace 2") + res = inst.cmd("workspace 2") assert res and res[0]["success"], f"workspace 2 failed: {res}" # Move Window 1 to Workspace 2 - # (It should still be floating? Yes, moving floating window to workspace works) - # Actually, we can just move it. - # Wait, if we are on Workspace 2, we can't easily move it here unless we focus it. - # But we switched to Workspace 2, so focus is on Workspace 2 (empty). - # We should go back to Workspace 1, move it to Workspace 2, then go to Workspace 2. - res = fresh_compositor.cmd("workspace 1") + res = inst.cmd("workspace 1") assert res and res[0]["success"], f"workspace 1 failed: {res}" - res = fresh_compositor.cmd("move container to workspace 2") + res = inst.cmd("move container to workspace 2") assert res and res[0]["success"], f"move to ws 2 failed: {res}" - res = fresh_compositor.cmd("workspace 2") + res = inst.cmd("workspace 2") assert res and res[0]["success"], f"workspace 2 failed: {res}" # Now Window 1 is floating on Workspace 2. # We must make it tiled on Workspace 2. - res = fresh_compositor.cmd("floating disable") + res = inst.cmd("floating disable") assert res and res[0]["success"], f"floating disable failed: {res}" # Set mode to vertical to stack next window - res = fresh_compositor.cmd("set_mode v") + res = inst.cmd("set_mode v") assert res and res[0]["success"], f"set_mode v failed: {res}" # Open Window 2 on Workspace 2 - with wayland_client(fresh_compositor, "Window 2"): - wait_for_client_map(fresh_compositor, "Window 2") + with wayland_client(inst, "Window 2"): + wait_for_client_map(inst, "Window 2") # Now we have on Workspace 2: Split (V) -> [Window 1, Window 2] # Move Window 2 to Workspace 3 (so Window 1 is only child of V-split) - res = fresh_compositor.cmd("move container to workspace 3") + res = inst.cmd("move container to workspace 3") assert res and res[0]["success"], f"move to ws 3 failed: {res}" # Now Window 1 is the only child of V-split on Workspace 2. # Workspace 2 has no other windows. # 3. Go to Workspace 1 - res = fresh_compositor.cmd("workspace 1") + res = inst.cmd("workspace 1") assert res and res[0]["success"], f"workspace 1 failed: {res}" # 4. Restore layout "space1" - # This should try to restore Window 1 as floating on Workspace 1. - # It will detach Window 1 from Workspace 2, reaping V-split and destroying Workspace 2. - # Then it will call arrange_container with dangling Workspace 2 pointer. - try: - res = fresh_compositor.cmd("space restore space1") - assert res and res[0]["success"], f"space restore failed: {res}" - except Exception as e: - print(f"Compositor log:\n{fresh_compositor.read_log()}") - raise e + res = inst.cmd("space restore space1") + assert res and res[0]["success"], f"space restore failed: {res}" # Check if compositor process is still alive - log_content = fresh_compositor.read_log() - print(f"Compositor log:\n{log_content}") - if ( - fresh_compositor.proc.poll() is not None - or "node_table not initialized" in log_content - ): - pass - - assert fresh_compositor.proc.poll() is None, "Compositor crashed" + assert inst.proc.poll() is None, "Compositor crashed" + + # Clean up space + inst.cmd("space delete space1") diff --git a/tests/test_utils.py b/tests/test_utils.py index 27df53ed..2e4d22dc 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -3,7 +3,7 @@ import subprocess import re import pytest -from typing import Generator, Any +from typing import Generator, Any, Dict, Optional from contextlib import contextmanager from pathlib import Path from scrollipc import ScrollIPC @@ -109,6 +109,12 @@ def get_tree(self) -> dict: def read_log(self) -> str: return self.log_path.read_text() + def reload_config(self, config_content: str) -> None: + config_path = self.temp_dir / "config" + config_path.write_text(config_content) + self.cmd("reload") + self.wait_for_idle() + def execute_lua(self, code: str) -> Any: import json @@ -152,6 +158,65 @@ def execute_lua(self, code: str) -> Any: def getenv(self, var: str) -> str | None: return self.execute_lua(f'return os.getenv("{var}")') + def wait_for_idle(self, timeout: float = 5.0) -> None: + start = time.time() + while time.time() - start < timeout: + pending = self.execute_lua( + "return scroll.pending_transactions() or scroll.animating()" + ) + if not pending: + return + time.sleep(0.005) + raise TimeoutError("Timeout waiting for compositor to become idle") + + def wait_for_transactions(self, timeout: float = 5.0) -> None: + start = time.time() + while time.time() - start < timeout: + if not self.execute_lua("return scroll.pending_transactions()"): + return + time.sleep(0.005) + raise TimeoutError("Timeout waiting for transactions") + + def reset(self) -> None: + # 1. Kill all views to clean up leftover windows + try: + self.cmd("kill all") + except Exception: + pass + + # 2. Clean up extra outputs + try: + tree = self.get_tree() + outputs: list[str] = [] + for child in tree.get("nodes", []): + if child.get("type") == "output" and child.get("name") != "__i3": + outputs.append(child["name"]) + + if "HEADLESS-1" in outputs: + for out in outputs: + if out != "HEADLESS-1" and out.startswith("HEADLESS-"): + self.cmd(f"output {out} unplug") + self.wait_for_idle() + except Exception: + pass + + # 3. Reload config to reset defaults + try: + config_path = self.temp_dir / "config" + config_path.write_text("workspace 1\nxwayland force\n") + self.cmd("reload") + self.wait_for_idle() + except Exception: + pass + + # 4. Reset workspaces (recreate workspace 1) + try: + self.cmd("workspace __temp") + self.cmd("workspace 1") + self.wait_for_idle() + except Exception: + pass + @contextmanager def assert_logs_match( self, pattern: str, timeout: float = 5.0 @@ -171,6 +236,83 @@ def assert_logs_match( ) time.sleep(0.1) + def wait_for_log_pattern( + self, pattern: str, timeout: float = 5.0, from_start: bool = False + ) -> None: + compiled_pattern = re.compile(pattern) + start_time = time.time() + initial_log_len = 0 if from_start else len(self.read_log()) + while True: + current_log: str = self.read_log() + log_to_search = current_log if from_start else current_log[initial_log_len:] + if compiled_pattern.search(log_to_search): + return + if time.time() - start_time > timeout: + raise AssertionError( + f"Pattern '{pattern}' not found in log output within" + f" {timeout}s.\nLog searched was:\n{log_to_search}" + ) + time.sleep(0.1) + + +class ScrollCompositorFactory: + _session: ScrollInstance + _active_context: Optional["ScrollCompositorContextManager"] + + def __init__(self, session: ScrollInstance): + self._session = session + self._active_context = None + + def print_log(self) -> None: + try: + print( + f"Compositor log (PID {self._session.proc.pid}):\n{self._session.read_log()}" + ) + except Exception as log_err: + print(f"Failed to read compositor log: {log_err}") + + def __call__(self, config: str | None = None) -> "ScrollCompositorContextManager": + return ScrollCompositorContextManager(self, self._session, config) + + +class ScrollCompositorContextManager: + _factory: ScrollCompositorFactory + _session: ScrollInstance + _config: Optional[str] + + def __init__( + self, + factory: ScrollCompositorFactory, + session: ScrollInstance, + config: Optional[str] = None, + ): + self._factory = factory + self._session = session + self._config = config + + def __enter__(self) -> ScrollInstance: + if self._factory._active_context is not None: + raise RuntimeError( + "ScrollInstance is already active in another context manager" + ) + self._factory._active_context = self + + if self._config is not None: + config_path = self._session.temp_dir / "config" + config_path.write_text(self._config) + self._session.cmd("reload") + self._session.wait_for_idle() + + return self._session + + def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: + try: + if exc_type is not None: + self._factory.print_log() + self._session.reset() + finally: + self._factory._active_context = None + @contextmanager def run_compositor( @@ -191,9 +333,11 @@ def run_compositor( tests_dir = Path(__file__).parent.resolve() supp_path = tests_dir / "lsan.supp" if "LSAN_OPTIONS" in env: - env["LSAN_OPTIONS"] = f"suppressions={supp_path}:{env['LSAN_OPTIONS']}" + env["LSAN_OPTIONS"] = ( + f"detect_leaks=0:suppressions={supp_path}:{env['LSAN_OPTIONS']}" + ) else: - env["LSAN_OPTIONS"] = f"suppressions={supp_path}" + env["LSAN_OPTIONS"] = f"detect_leaks=0:suppressions={supp_path}" if "DISPLAY" in env: del env["DISPLAY"] if "WAYLAND_DISPLAY" in env: @@ -282,3 +426,33 @@ def wait_for_client_map(compositor: ScrollInstance, title: str) -> int: time.sleep(0.05) tries += 1 raise RuntimeError(f"Client '{title}' did not map") + + +def find_node_by_title_contains( + node: Dict[str, Any], title_sub: str +) -> Optional[Dict[str, Any]]: + name = node.get("name") + if name and title_sub in name: + return node + for child in node.get("nodes", []): + res = find_node_by_title_contains(child, title_sub) + if res: + return res + for child in node.get("floating_nodes", []): + res = find_node_by_title_contains(child, title_sub) + if res: + return res + return None + + +def wait_for_title_contains_map( + compositor: ScrollInstance, title_sub: str, timeout: float = 15.0 +) -> Dict[str, Any]: + start = time.time() + while time.time() - start < timeout: + tree = compositor.get_tree() + node = find_node_by_title_contains(tree, title_sub) + if node: + return node + time.sleep(0.2) + raise RuntimeError(f"Client with title containing '{title_sub}' did not map.") diff --git a/tests/test_workspace_animation_uaf.py b/tests/test_workspace_animation_uaf.py new file mode 100644 index 00000000..56b733d7 --- /dev/null +++ b/tests/test_workspace_animation_uaf.py @@ -0,0 +1,24 @@ +import time +from test_utils import ScrollCompositorFactory + + +def test_workspace_switch_empty_uaf( + scroll_compositor_factory: ScrollCompositorFactory, +) -> None: + config = ( + "workspace 1\n" + "xwayland force\n" + "animations enabled yes\n" + "animations workspace_switch yes 5000\n" + ) + with scroll_compositor_factory(config) as scroll_compositor: + scroll_compositor.cmd("workspace 2; workspace 3; workspace 1") + # Sleep a bit to let the animation start and run a few frames, triggering UAF + time.sleep(0.5) + + # Check if the compositor process crashed + if scroll_compositor.proc.poll() is not None: + print("=== COMPOSITOR LOG (ASAN) ===") + print(scroll_compositor.read_log()) + print("=============================") + assert False, "Compositor crashed" diff --git a/tests/test_workspace_split_uaf.py b/tests/test_workspace_split_uaf.py index cb034517..dc33dbd6 100644 --- a/tests/test_workspace_split_uaf.py +++ b/tests/test_workspace_split_uaf.py @@ -1,59 +1,48 @@ -import time -from conftest import ScrollInstance -from test_utils import wayland_client, wait_for_client_map +from test_utils import wayland_client, wait_for_client_map, ScrollCompositorFactory -def test_workspace_split_uaf_crash(fresh_compositor: ScrollInstance) -> None: - try: +def test_workspace_split_uaf_crash( + scroll_compositor_factory: ScrollCompositorFactory, +) -> None: + config = "workspace 1\nxwayland force\nanimations enabled off\n" + with scroll_compositor_factory(config) as scroll_compositor: # 1. Open Window 1 on Workspace 1 (active on HEADLESS-1) - with wayland_client(fresh_compositor, "Window 1"): - wait_for_client_map(fresh_compositor, "Window 1") + with wayland_client(scroll_compositor, "Window 1"): + wait_for_client_map(scroll_compositor, "Window 1") # 2. Split workspace 1. This creates workspace 2 as sibling. # Workspace 1 has Window 1, Workspace 2 is empty. - res = fresh_compositor.cmd("workspace split") + res = scroll_compositor.cmd("workspace split") assert res and res[0]["success"], f"workspace split failed: {res}" # 3. Create a second output HEADLESS-2. # It should get a default workspace (probably 3). - res = fresh_compositor.cmd("create_output") + res = scroll_compositor.cmd("create_output") assert res and res[0]["success"], f"create_output failed: {res}" - - time.sleep(0.5) + scroll_compositor.wait_for_idle() # 4. Unplug HEADLESS-1. # Workspaces 1 and 2 should be evacuated. # Workspace 1 (non-empty) is moved to HEADLESS-2. # Workspace 2 (empty) is destroyed. - res = fresh_compositor.cmd("output HEADLESS-1 unplug") + res = scroll_compositor.cmd("output HEADLESS-1 unplug") assert res and res[0]["success"], f"unplug failed: {res}" - - time.sleep(0.5) + scroll_compositor.wait_for_idle() # 5. Focus Workspace 3 on HEADLESS-2 (so Workspace 1 becomes inactive) - res = fresh_compositor.cmd("workspace 3") + res = scroll_compositor.cmd("workspace 3") assert res and res[0]["success"], f"workspace 3 failed: {res}" - - time.sleep(0.5) + scroll_compositor.wait_for_idle() # 6. Move Window 1 from Workspace 1 to Workspace 3. # Since Window 1 is moved out of Workspace 1, and Workspace 1 is inactive, # it should trigger workspace_consider_destroy(Workspace 1). # Workspace 1 is empty, and it is split (sibling was Workspace 2). # It will try to access Workspace 2 (which is destroyed) -> UAF. - res = fresh_compositor.cmd( + res = scroll_compositor.cmd( '[title="Window 1"] move container to workspace 3' ) assert res and res[0]["success"], f"move container failed: {res}" # Check if compositor process is still alive - log_content = fresh_compositor.read_log() - print(f"Compositor log:\n{log_content}") - assert fresh_compositor.proc.poll() is None, "Compositor crashed" - except Exception as e: - print(f"Test failed with exception: {e}") - try: - print(f"Compositor log:\n{fresh_compositor.read_log()}") - except Exception as log_err: - print(f"Failed to read compositor log: {log_err}") - raise e + assert scroll_compositor.proc.poll() is None, "Compositor crashed" From 33c92d26646572cbd2a47d6737651a77606d58f3 Mon Sep 17 00:00:00 2001 From: Jeremy Maitin-Shepard Date: Thu, 9 Jul 2026 13:39:28 -0700 Subject: [PATCH 2/2] Ignore stale client commits on floating resize When a new view is mapped, criteria rules may configure its geometry (e.g. `resize set`). This configuration sends a new configure event to the client. However, during the map event handling, we are still processing the initial commit from the client. After the map event is handled and the rule-configured geometry is applied via a transaction, the compositor's commit handler (`handle_commit`) continues. For floating containers, `handle_commit` resizes the container to match the client's committed buffer size. If it processes the initial commit (which has the initial client size, not the configured one), it overrides the configured geometry back to the initial size. To fix this, we introduce `is_commit_stale` helper which checks if there is a pending or scheduled configure event with a newer serial than the serial acknowledged by the current commit. If the commit is stale, we ignore its size for resizing floating containers, waiting instead for the client to acknowledge and commit the newly configured size. How to reproduce: 1. Register the for_window rule: swaymsg 'for_window [title="Scratchpad Test"] move scratchpad, resize set 500 500' 2. Start the client: foot -T "Scratchpad Test" 3. Query the scratchpad window geometry: swaymsg -t get_tree | jq '.. | select(.name? == "__i3_scratch") | .floating_nodes[] | select(.name == "Scratchpad Test") | .rect' On buggy versions, this will output the client's default size instead of the configured 500x500. Additionally: * Make `wait_for_client_map` test utility robust by searching all workspaces and the scratchpad for the mapped view, instead of just checking the focused view. This allows it to be used for windows that are moved to the scratchpad (and thus hidden/unfocused) immediately upon mapping. * Update `wayland-client.c` to support dynamic resizing in tests. * Add integration test `tests/test_scratchpad_geometry.py` to verify the fix. --- sway/desktop/xdg_shell.c | 29 +++++++++++++++------ tests/clients/wayland-client.c | 20 ++++++++++++++- tests/test_scratchpad_geometry.py | 42 +++++++++++++++++++++++++++++++ tests/test_utils.py | 33 +++++++++++++++++++++--- 4 files changed, 113 insertions(+), 11 deletions(-) create mode 100644 tests/test_scratchpad_geometry.py diff --git a/sway/desktop/xdg_shell.c b/sway/desktop/xdg_shell.c index d9141236..24664db3 100644 --- a/sway/desktop/xdg_shell.c +++ b/sway/desktop/xdg_shell.c @@ -292,6 +292,20 @@ static const struct sway_view_impl view_impl = { .destroy = destroy, }; +static bool is_commit_stale(struct sway_view *view, struct wlr_xdg_surface *xdg_surface) { + uint32_t latest_serial = 0; + if (xdg_surface->configure_idle != NULL) { + latest_serial = xdg_surface->scheduled_serial; + } else if (!wl_list_empty(&xdg_surface->configure_list)) { + struct wlr_xdg_surface_configure *configure = + wl_container_of(xdg_surface->configure_list.prev, configure, link); + latest_serial = configure->serial; + } else { + return false; + } + return xdg_surface->current.configure_serial < latest_serial; +} + static void handle_commit(struct wl_listener *listener, void *data) { struct sway_xdg_shell_view *xdg_shell_view = wl_container_of(listener, xdg_shell_view, commit); @@ -341,20 +355,21 @@ static void handle_commit(struct wl_listener *listener, void *data) { new_geo->height != view->geometry.height || new_geo->x != view->geometry.x || new_geo->y != view->geometry.y; - if (new_size) { // The client changed its surface size in this commit. For floating // containers, we resize the container to match. For tiling containers, // we only recenter the surface. memcpy(&view->geometry, new_geo, sizeof(struct wlr_box)); if (container_is_floating(view->container)) { - view_update_size(view); - // Only set the toplevel size the current container actually has a size. - if (view->container->current.width) { - wlr_xdg_toplevel_set_size(view->wlr_xdg_toplevel, view->geometry.width, - view->geometry.height); + if (!is_commit_stale(view, xdg_surface)) { + view_update_size(view); + // Only set the toplevel size the current container actually has a size. + if (view->container->current.width) { + wlr_xdg_toplevel_set_size( + view->wlr_xdg_toplevel, view->geometry.width, view->geometry.height); + } + transaction_commit_dirty_client(); } - transaction_commit_dirty_client(); } view_center_and_clip_surface(view); diff --git a/tests/clients/wayland-client.c b/tests/clients/wayland-client.c index 9aad66e2..2eb7c54f 100644 --- a/tests/clients/wayland-client.c +++ b/tests/clients/wayland-client.c @@ -19,6 +19,7 @@ struct client_state { struct xdg_toplevel *xdg_toplevel; struct wl_buffer *buffer; int width, height; + int buffer_width, buffer_height; void *shm_data; }; @@ -51,6 +52,14 @@ static void xdg_surface_configure(void *data, struct xdg_surface *xdg_surface, u struct client_state *state = data; xdg_surface_ack_configure(xdg_surface, serial); + if (state->buffer && + (state->buffer_width != state->width || state->buffer_height != state->height)) { + wl_buffer_destroy(state->buffer); + state->buffer = NULL; + munmap(state->shm_data, state->buffer_width * 4 * state->buffer_height); + state->shm_data = NULL; + } + if (!state->buffer) { int stride = state->width * 4; int size = stride * state->height; @@ -78,6 +87,8 @@ static void xdg_surface_configure(void *data, struct xdg_surface *xdg_surface, u for (int i = 0; i < state->width * state->height; ++i) { pixels[i] = 0xFF0000FF; // Blue } + state->buffer_width = state->width; + state->buffer_height = state->height; } wl_surface_attach(state->surface, state->buffer, 0, 0); @@ -86,7 +97,14 @@ static void xdg_surface_configure(void *data, struct xdg_surface *xdg_surface, u } static const struct xdg_surface_listener xdg_surface_listener = { xdg_surface_configure }; -static void xdg_toplevel_configure(void *data, struct xdg_toplevel *xdg_toplevel, int32_t width, int32_t height, struct wl_array *states) {} +static void xdg_toplevel_configure(void *data, struct xdg_toplevel *xdg_toplevel, int32_t width, + int32_t height, struct wl_array *states) { + struct client_state *state = data; + if (width > 0 && height > 0) { + state->width = width; + state->height = height; + } +} static void xdg_toplevel_close(void *data, struct xdg_toplevel *xdg_toplevel) { exit(0); } diff --git a/tests/test_scratchpad_geometry.py b/tests/test_scratchpad_geometry.py new file mode 100644 index 00000000..ec9bba7c --- /dev/null +++ b/tests/test_scratchpad_geometry.py @@ -0,0 +1,42 @@ +from conftest import ScrollInstance +from test_utils import wait_for_client_map, wayland_client + + +def test_scratchpad_geometry_rules(scroll_compositor: ScrollInstance) -> None: + inst = scroll_compositor + title = "Scratchpad Test" + + # Register for_window rule + res = inst.cmd( + f'for_window [title="{title}"] "move scratchpad, resize set 500 px 500 px, move position 100 100"' + ) + + assert res[0]["success"], f"for_window command failed: {res}" + + try: + with wayland_client(inst, title): + view_id = wait_for_client_map(inst, title) + assert view_id is not None + + con_id = inst.execute_lua(f"return scroll.view_get_container({view_id})") + assert con_id is not None + + # Wait for any animation to settle (though scratchpad hidden windows might not animate) + inst.wait_for_idle() + + geom = inst.execute_lua(f"return scroll.container_get_geometry({con_id})") + print("Scratchpad Geometry:", geom) + + # Assert geometry + assert geom is not None + assert geom["width"] == 500 + assert geom["height"] == 500 + assert geom["x"] == 100 + assert geom["y"] == 100 + + # Clean up + inst.execute_lua(f"scroll.view_close({view_id})") + except Exception: + print("Compositor log:") + print(inst.read_log()) + raise diff --git a/tests/test_utils.py b/tests/test_utils.py index 2e4d22dc..45bda991 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -415,10 +415,37 @@ def wait_for_client_map(compositor: ScrollInstance, title: str) -> int: tries: int = 0 while tries < 50: view_id = compositor.execute_lua(f""" - local view = scroll.focused_view() - if view and scroll.view_get_title(view) == "{title}" then - return view + local function find_view(title) + -- Check scratchpad + for _, con in ipairs(scroll.scratchpad_get_containers()) do + for _, view in ipairs(scroll.container_get_views(con)) do + if scroll.view_get_title(view) == title then + return view + end + end + end + -- Check all outputs and workspaces + for _, output in ipairs(scroll.root_get_outputs()) do + for _, ws in ipairs(scroll.output_get_workspaces(output)) do + for _, con in ipairs(scroll.workspace_get_tiling(ws)) do + for _, view in ipairs(scroll.container_get_views(con)) do + if scroll.view_get_title(view) == title then + return view + end + end + end + for _, con in ipairs(scroll.workspace_get_floating(ws)) do + for _, view in ipairs(scroll.container_get_views(con)) do + if scroll.view_get_title(view) == title then + return view + end + end + end + end + end + return nil end + return find_view("{title}") """) if view_id is not None: assert isinstance(view_id, int)