diff --git a/nix/tests/vm/lib/komorebi_vm/platforms/kde/extras/commands.py b/nix/tests/vm/lib/komorebi_vm/platforms/kde/extras/commands.py new file mode 100644 index 00000000..3a675d8c --- /dev/null +++ b/nix/tests/vm/lib/komorebi_vm/platforms/kde/extras/commands.py @@ -0,0 +1,291 @@ +"""What only KDE can tell about the commands, round the shared `commands` scenarios. + +Every window the daemon has is a window KWin has, and the frame the daemon last asked KWin for is +the one KWin carries out. The shared scenarios check the layouts with five windows and that every +command is understood and leaves the daemon healthy, as the operating system sees it; what is here +reads KWin (`kdotool`) and the daemon's log. + +The suite runs `setup`, `layouts_by_count`, the shared `layouts` and `sweep`, +`replace_configuration`, `handover`, `title_rule`, the shared `prints`, `queries` and `at_the_end`. +The daemon runs on a session with two outputs, and `clients` are five windows on the first. +""" + +import shlex + +from komorebi_vm import daemon +from komorebi_vm.core.checks import Checks +from komorebi_vm.core.platform import Client +from komorebi_vm.kde_platform import CELL, KdePlatform + +OUTPUTS = 2 +LAYOUTS = ( + "bsp", + "columns", + "rows", + "vertical-stack", + "horizontal-stack", + "ultrawide-vertical-stack", + "grid", + "right-main-vertical-stack", + "scrolling", +) +QUERIES = ( + "focused-monitor-index", + "focused-workspace-index", + "focused-container-index", + "focused-window-index", + "focused-workspace-name", + "focused-workspace-layout", + "focused-container-kind", + "version", +) + + +def _live_windows(platform: KdePlatform) -> list[str]: + """The ids of the windows KWin has: every client of the suite is a foot.""" + return platform.kde.run("kdotool search --class foot")[1].split() + + +def healthy(platform: KdePlatform) -> int: + """0 when the daemon is running, answers `komorebic state` and has every window KWin has. + + Otherwise the reason: 1 not running, 2 no state, 3 not the windows KWin has. + """ + kde = platform.kde + if not kde.komorebi_running(): + return 1 + if kde.run("komorebic state")[0] != 0: + return 2 + if kde.managed_all() != len(_live_windows(platform)): + return 3 + return 0 + + +def _outside_or_overlapping(platform: KdePlatform) -> list[str]: + """The windows that are outside the outputs, or on another window (2 pixels are allowed).""" + kde = platform.kde + width, height = platform.size + rects = {wid: kde.rect(wid) for wid in _live_windows(platform)} + bad = [] + for wid, r in rects.items(): + if r.x < -1 or r.y < -1 or r.x + r.w > width + 1 or r.y + r.h > height + 1: + bad.append(f"{kde.title(wid)} outside at {r}") + ids = list(rects) + for i, a in enumerate(ids): + for b in ids[i + 1 :]: + ra, rb = rects[a], rects[b] + if ( + ra.x < rb.x + rb.w - 2 + and rb.x < ra.x + ra.w - 2 + and ra.y < rb.y + rb.h - 2 + and rb.y < ra.y + ra.h - 2 + ): + bad.append(f"{kde.title(a)} on {kde.title(b)}") + return bad + + +def _open_window(platform: KdePlatform, n: int) -> None: + platform.spawn(f"win{n}") + + +def _close_window(platform: KdePlatform, n: int) -> None: + kde = platform.kde + title = f"win{n}" + kde.close(kde.wid(title)) + kde.wait_for(lambda: kde.wid(title) == "", timeout=10) + + +def _where_asked(platform: KdePlatform, wid: str) -> str | None: + """None when KWin has the window where the daemon last asked, else what is wrong.""" + kde = platform.kde + title = kde.title(wid) + want, have = daemon.asked(kde.komorebi_log(), title), kde.rect(wid) + if want is None: + return f"{title}: the daemon never asked for a frame" + if (have.x, have.y) != (want.x, want.y) or not ( + 0 <= want.w - have.w < CELL and 0 <= want.h - have.h < CELL + ): + return f"{title}: asked {want}, has {have}" + return None + + +def setup(platform: KdePlatform, checks: Checks, clients: list[Client]) -> None: + """The daemon answers with both outputs and has the five windows, as KWin has them.""" + kde = platform.kde + checks.check( + "the daemon answers komorebic, with both outputs", + kde.wait_for( + lambda: ( + kde.run("komorebic monitor-information")[0] == 0 + and len(kde.state()["monitors"]["elements"]) == OUTPUTS + ), + timeout=30, + ), + ) + checks.check( + "five windows, and the daemon has all five", + kde.wait_for(lambda: kde.managed() == len(clients) and healthy(platform) == 0, timeout=20), + ) + + +def layouts_by_count(platform: KdePlatform, checks: Checks, clients: list[Client]) -> None: + """Every layout with five, three, two and one windows: the windows are in their outputs and + apart, the daemon is healthy, and KWin has each window where the daemon last asked for it. + + The shared `layouts` checks five windows. The windows that were closed are opened again, and + the layout is `bsp` at the end. + """ + kde = platform.kde + for layout in LAYOUTS: + kde.komorebic("change-layout", layout) + verdict = [] + for keep in (5, 3, 2, 1): + while kde.managed() > keep: + _close_window(platform, kde.managed()) + kde.wait_for(lambda: healthy(platform) == 0, timeout=10) + # the daemon retiles on the close; give KWin the time to carry it out + kde.wait_for( + lambda: all(_where_asked(platform, w) is None for w in _live_windows(platform)), + timeout=10, + ) + # the scrolling layout is a strip whose columns can be past the edge of the output + if layout != "scrolling": + verdict += [f"{keep}w: {bad}" for bad in _outside_or_overlapping(platform)] + if reason := healthy(platform): + verdict.append(f"{keep}w: not healthy ({reason})") + checks.check( + f"{layout}: the windows are in their outputs and do not overlap, and the daemon is" + " healthy" + "".join(f"\n {v}" for v in verdict), + verdict == [], + ) + for n in range(kde.managed() + 1, len(clients) + 1): + _open_window(platform, n) + kde.wait_for(lambda: kde.managed() == len(clients), timeout=20) + kde.komorebic("change-layout", "bsp") + + +def replace_configuration(platform: KdePlatform, checks: Checks, clients: list[Client]) -> None: + """After the shared sweep ran `replace-configuration`, the window manager is made again from + the file, whose first workspace is named `again`.""" + first = platform.kde.state()["monitors"]["elements"][0]["workspaces"]["elements"][0] + checks.check( + "replace-configuration made the window manager again from the file", + first["name"] == "again", + ) + + +def handover(platform: KdePlatform, checks: Checks, clients: list[Client]) -> None: + """`minimize`, `unmanage` and `manage` on the focused window: the daemon lets go of it, and + takes it up again. + + They take a window out of the daemon's hands, so the shared sweep does not run them. + """ + kde = platform.kde + focused = kde.active() + title = kde.title(focused) + before = kde.managed() + kde.wait_for(lambda: kde.focused_title() == title, timeout=10) + kde.komorebic("minimize") + checks.check( + f"minimize: the daemon lets go of {title} when KWin minimizes it", + kde.wait_for(lambda: kde.managed() == before - 1, timeout=10) and kde.wid(title) != "", + ) + kde.activate(focused) + checks.check( + f"{title}, restored, is tiled again", + kde.wait_for(lambda: kde.managed() == before and healthy(platform) == 0, timeout=10), + ) + kde.wait_for(lambda: kde.focused_title() == title, timeout=10) + kde.komorebic("unmanage") + checks.check( + f"unmanage: the daemon lets go of {title}, which stays open", + kde.wait_for(lambda: kde.managed() == before - 1, timeout=10) and kde.wid(title) != "", + ) + width, height = platform.size[0] // OUTPUTS, platform.size[1] + + def centered() -> bool: + r = kde.rect(focused) + return ( + abs(r.x + r.w // 2 - width // 2) <= CELL and abs(r.y + r.h // 2 - height // 2) <= CELL + ) + + checks.check(f"{title} is in the middle of the work area", kde.wait_for(centered, timeout=10)) + kde.komorebic("manage") + checks.check( + f"manage: {title} is tiled again", + kde.wait_for(lambda: kde.managed() == before and healthy(platform) == 0, timeout=10), + ) + + +def title_rule(platform: KdePlatform, checks: Checks, clients: list[Client]) -> None: + """A rule that matches a title decides when the window has its title: a window that opens + with a title the ignore rule matches is left alone, and tiled once it has another.""" + kde = platform.kde + kde.komorebic("ignore-rule", "title", "pending-rule") + before_all = kde.managed_all() + + def ignored(title: str) -> bool: + """True once the daemon has logged that it leaves the window titled `title` alone.""" + return any( + "should not be managed" in line and f"title: {title}," in line + for line in kde.komorebi_log().splitlines() + ) + + # foot is titled pending-rule when it opens, and gives itself another title after 4 seconds + rename = 'sleep 4; printf "\\033]0;renamed-rule\\007"; sleep 3000' + kde.spawn(f"foot -T pending-rule sh -c {shlex.quote(rename)}") + checks.check( + "a window the ignore rule matches by its title is left alone", + kde.wait_for(lambda: ignored("pending-rule"), timeout=20) + and kde.managed_all() == before_all, + ) + checks.check( + "once it has a title the rule does not match, it is tiled", + kde.wait_for( + lambda: kde.wid("renamed-rule") != "" and kde.managed_all() == before_all + 1, + timeout=20, + ) + and healthy(platform) == 0, + ) + # the daemon's log names a window by the title it had when it was focused, which for this one + # is the first: close it, so that the last check finds every window it looks for in the log + kde.close(kde.wid("renamed-rule")) + checks.check( + "closed, it leaves komorebi", + kde.wait_for(lambda: kde.wid("renamed-rule") == "" and healthy(platform) == 0, timeout=10), + ) + + +def queries(platform: KdePlatform, checks: Checks, clients: list[Client]) -> None: + """`komorebic query` answers, and the resize commands that save and load are understood.""" + kde = platform.kde + for query in QUERIES: + status, out = kde.run(f"komorebic query {query}") + checks.check(f"query {query} answers", status == 0 and out.strip() != "") + checks.check( + "quick-save-resize and quick-load-resize", + kde.run("komorebic quick-save-resize && komorebic quick-load-resize")[0] == 0, + ) + checks.check( + "save-resize and load-resize", + kde.run( + "komorebic save-resize /tmp/e2e-resize.json" + " && komorebic load-resize /tmp/e2e-resize.json" + )[0] + == 0, + ) + + +def at_the_end(platform: KdePlatform, checks: Checks, clients: list[Client]) -> None: + """The daemon has every window KWin has, each is where the daemon last asked for it, and a + clean stop leaves the daemon not running.""" + kde = platform.kde + checks.check("the daemon has every window KWin has, and answers", healthy(platform) == 0) + for wid in _live_windows(platform): + print(f" {kde.title(wid)}: {kde.rect(wid)}") + mismatch = [m for m in (_where_asked(platform, w) for w in _live_windows(platform)) if m] + for m in mismatch: + print(f" {m}") + checks.check("every window is where the daemon last asked for it", mismatch == []) + kde.komorebic("stop") + checks.check("a clean stop", kde.wait_for(lambda: not kde.komorebi_running())) diff --git a/nix/tests/vm/suites/commands/test.py b/nix/tests/vm/suites/commands/test.py index 6449a44e..dcf1dc08 100644 --- a/nix/tests/vm/suites/commands/test.py +++ b/nix/tests/vm/suites/commands/test.py @@ -1,35 +1,23 @@ # Every layout, and a sweep of komorebic's commands, against a real headless KWin with two outputs. # -# After each command the same things have to be true: komorebi is alive, it can still say what it -# knows (`komorebic state`), and every window it has is a window KWin has. After each layout, the -# windows are inside the outputs and do not lie on each other. A command is also expected to be -# understood (exit status 0). The Rust tests check what each command does against the fake; this -# suite checks that none of them breaks the daemon in a real session, and that what the daemon -# asks KWin for is what KWin carries out. Every command `komorebic` has is in the sweep, but -# `minimize` and `unmanage`, which take a window out of komorebi's hands and have a section of -# their own. +# After each command the same things have to be true: the daemon is alive, it can still say what it +# knows (`komorebic state`), and every window it has is a window the system has. The shared scenarios +# (`komorebi_vm.core.scenarios.commands`, which run on Windows too) check that, for every layout with +# five windows and for every command, and that a command is understood (exit status 0). What only +# a real KWin can tell is in `komorebi_vm.platforms.kde.extras.commands`: every layout with fewer +# windows, that what the daemon asks KWin for is what KWin carries out, and the commands that take +# a window out of the daemon's hands. The Rust tests check what each command does against the fake. import shlex -from komorebi_vm import Checks, Kde, daemon -from komorebi_vm.core.platform import Client +from komorebi_vm import Checks, Kde from komorebi_vm.core.scenarios import commands +from komorebi_vm.core.suite import platform_only, run_suite, shared from komorebi_vm.kde_platform import KdePlatform +from komorebi_vm.platforms.kde.extras import commands as kde_commands W, H = 1920, 1080 OUTPUTS = 2 -# foot sizes itself to whole cells of its font, so its frame can be short of the one asked for -CELL = 24 -LAYOUTS = [ - "bsp", - "columns", - "rows", - "vertical-stack", - "horizontal-stack", - "ultrawide-vertical-stack", - "grid", - "right-main-vertical-stack", - "scrolling", -] +CONFIG = "/tmp/e2e-config.json" start_all() machine.wait_for_unit("multi-user.target") @@ -37,247 +25,31 @@ machine.wait_for_unit("user@1000.service") checks = Checks() kde = Kde.headless(machine, "main", size=(W, H), outputs=OUTPUTS, scale=SCALE) - - -def live_windows() -> list[str]: - """The ids of the windows KWin has: every client of this suite is a foot.""" - return kde.run("kdotool search --class foot")[1].split() - - -def healthy() -> int: - """0 when komorebi is running, answers `komorebic state` and has every window KWin has. - - Otherwise the reason: 1 not running, 2 no state, 3 not the windows KWin has. - """ - if not kde.komorebi_running(): - return 1 - status, _ = kde.run("komorebic state") - if status != 0: - return 2 - if kde.managed_all() != len(live_windows()): - return 3 - return 0 - - -def outside_or_overlapping() -> list[str]: - """The windows that are outside the outputs, or on another window (2 pixels are allowed).""" - rects = {wid: kde.rect(wid) for wid in live_windows()} - bad = [] - for wid, r in rects.items(): - if r.x < -1 or r.y < -1 or r.x + r.w > W * OUTPUTS + 1 or r.y + r.h > H + 1: - bad.append(f"{kde.title(wid)} outside at {r}") - ids = list(rects) - for i, a in enumerate(ids): - for b in ids[i + 1 :]: - ra, rb = rects[a], rects[b] - if ( - ra.x < rb.x + rb.w - 2 - and rb.x < ra.x + ra.w - 2 - and ra.y < rb.y + rb.h - 2 - and rb.y < ra.y + ra.h - 2 - ): - bad.append(f"{kde.title(a)} on {kde.title(b)}") - return bad - - -def open_window(n: int) -> None: - title = f"win{n}" - kde.spawn(f"foot -T {title} sleep 3000") - kde.wait_for(lambda: kde.wid(title) != "", timeout=30) - - -def close_window(n: int) -> None: - title = f"win{n}" - kde.close(kde.wid(title)) - kde.wait_for(lambda: kde.wid(title) == "", timeout=10) - - -def where_asked(wid: str) -> str | None: - """None when KWin has the window where the daemon last asked, else what is wrong.""" - title = kde.title(wid) - want, have = daemon.asked(kde.komorebi_log(), title), kde.rect(wid) - if want is None: - return f"{title}: the daemon never asked for a frame" - if (have.x, have.y) != (want.x, want.y) or not ( - 0 <= want.w - have.w < CELL and 0 <= want.h - have.h < CELL - ): - return f"{title}: asked {want}, has {have}" - return None - - -print("== 1. start, and five windows on the first output") -kde.start_komorebi() -checks.check( - "the daemon answers komorebic, with both outputs", - kde.wait_for( - lambda: kde.run("komorebic monitor-information")[0] == 0 - and len(kde.state()["monitors"]["elements"]) == OUTPUTS, - timeout=30, - ), -) -for n in range(1, 6): - open_window(n) -checks.check( - "five windows, and komorebi has all five", - kde.wait_for(lambda: kde.managed() == 5 and healthy() == 0, timeout=20), -) - -print("== 2. every layout, with five, three, two and one windows") -for layout in LAYOUTS: - kde.komorebic("change-layout", layout) - verdict = [] - for keep in (5, 3, 2, 1): - while kde.managed() > keep: - close_window(kde.managed()) - kde.wait_for(lambda: healthy() == 0, timeout=10) - # the daemon retiles on the close; give KWin the time to carry it out - kde.wait_for(lambda: all(where_asked(w) is None for w in live_windows()), timeout=10) - # the scrolling layout is a strip whose columns can be past the edge of the output - if layout != "scrolling": - verdict += [f"{keep}w: {bad}" for bad in outside_or_overlapping()] - reason = healthy() - if reason: - verdict.append(f"{keep}w: not healthy ({reason})") - checks.check( - f"{layout}: the windows are in their outputs and do not overlap, and komorebi is healthy" - + "".join(f"\n {v}" for v in verdict), - verdict == [], - ) - for n in range(kde.managed() + 1, 6): - open_window(n) - kde.wait_for(lambda: kde.managed() == 5, timeout=20) -kde.komorebic("change-layout", "bsp") - -print("== 3. a sweep of commands (each must be understood, and leave komorebi healthy)") platform = KdePlatform(kde, (W * OUTPUTS, H)) -clients = [Client(f"win{n}") for n in range(1, 6)] -# the same layouts and the same commands run against kfw: komorebi_vm.core.scenarios.commands has the -# list, and a command a port lacks is skipped there with its name -print(" the layouts, through the neutral platform") -commands.layouts(platform, checks, clients) -kde.komorebic("change-layout", "bsp") -CONFIG = "/tmp/e2e-config.json" + +platform.start_komorebi() +clients = [platform.spawn(f"win{n}") for n in range(1, 6)] kde.sh( "printf %s " + shlex.quote('{ "monitors": [ { "workspaces": [ { "name": "again", "layout": "BSP" } ] } ] }') + f" > {CONFIG}" ) -commands.sweep(platform, checks, clients, CONFIG) -first_workspace = kde.state()["monitors"]["elements"][0]["workspaces"]["elements"][0] -checks.check( - "replace-configuration made the window manager again from the file", - first_workspace["name"] == "again", +args = (platform, checks, clients) +run_suite( + checks, + platform_only("kde", kde_commands.setup, *args), + platform_only("kde", kde_commands.layouts_by_count, *args), + shared(commands.layouts, *args), + platform_only("kde", kde.komorebic, "change-layout", "bsp"), + shared(commands.sweep, *args, CONFIG), + platform_only("kde", kde_commands.replace_configuration, *args), + platform_only("kde", kde_commands.handover, *args), + platform_only("kde", kde_commands.title_rule, *args), + shared(commands.prints, platform, checks), + platform_only("kde", kde_commands.queries, *args), + platform_only("kde", kde_commands.at_the_end, *args), ) -print("== 3b. minimize, unmanage and manage the focused window") -focused = kde.active() -title = kde.title(focused) -before = kde.managed() -kde.wait_for(lambda: kde.focused_title() == title, timeout=10) -kde.komorebic("minimize") -checks.check( - f"minimize: komorebi lets go of {title} when KWin minimizes it", - kde.wait_for(lambda: kde.managed() == before - 1, timeout=10) - and kde.wid(title) != "", -) -kde.activate(focused) -checks.check( - f"{title}, restored, is tiled again", - kde.wait_for(lambda: kde.managed() == before and healthy() == 0, timeout=10), -) -kde.wait_for(lambda: kde.focused_title() == title, timeout=10) -kde.komorebic("unmanage") -checks.check( - f"unmanage: komorebi lets go of {title}, which stays open", - kde.wait_for(lambda: kde.managed() == before - 1, timeout=10) and kde.wid(title) != "", -) - - -def centered() -> bool: - r = kde.rect(focused) - return abs(r.x + r.w // 2 - W // 2) <= CELL and abs(r.y + r.h // 2 - H // 2) <= CELL - - -checks.check(f"{title} is in the middle of the work area", kde.wait_for(centered, timeout=10)) -kde.komorebic("manage") -checks.check( - f"manage: {title} is tiled again", - kde.wait_for(lambda: kde.managed() == before and healthy() == 0, timeout=10), -) - -print("== 3c. a rule that matches a title decides when the window has its title") -kde.komorebic("ignore-rule", "title", "pending-rule") -before_all = kde.managed_all() - - -def ignored(title: str) -> bool: - """True once the daemon has logged that it leaves the window titled `title` alone.""" - return any( - "should not be managed" in line and f"title: {title}," in line - for line in kde.komorebi_log().splitlines() - ) - - -# foot is titled pending-rule when it opens, and gives itself another title after 4 seconds -rename = 'sleep 4; printf "\\033]0;renamed-rule\\007"; sleep 3000' -kde.spawn(f"foot -T pending-rule sh -c {shlex.quote(rename)}") -checks.check( - "a window the ignore rule matches by its title is left alone", - kde.wait_for(lambda: ignored("pending-rule"), timeout=20) and kde.managed_all() == before_all, -) -checks.check( - "once it has a title the rule does not match, it is tiled", - kde.wait_for( - lambda: kde.wid("renamed-rule") != "" and kde.managed_all() == before_all + 1, - timeout=20, - ) - and healthy() == 0, -) -# the daemon's log names a window by the title it had when it was focused, which for this one is -# the first: close it, so that the last check finds every window it looks for in the log -kde.close(kde.wid("renamed-rule")) -checks.check( - "closed, it leaves komorebi", - kde.wait_for(lambda: kde.wid("renamed-rule") == "" and healthy() == 0, timeout=10), -) - -print("== 4. what komorebic only prints") -commands.prints(platform, checks) -for query in [ - "focused-monitor-index", - "focused-workspace-index", - "focused-container-index", - "focused-window-index", - "focused-workspace-name", - "focused-workspace-layout", - "focused-container-kind", - "version", -]: - status, out = kde.run(f"komorebic query {query}") - checks.check(f"query {query} answers", status == 0 and out.strip() != "") -checks.check( - "quick-save-resize and quick-load-resize", - kde.run("komorebic quick-save-resize && komorebic quick-load-resize")[0] == 0, -) -checks.check( - "save-resize and load-resize", - kde.run( - "komorebic save-resize /tmp/e2e-resize.json && komorebic load-resize /tmp/e2e-resize.json" - )[0] - == 0, -) - -print("== 5. still all there") -checks.check("komorebi has every window KWin has, and answers", healthy() == 0) -for wid in live_windows(): - print(f" {kde.title(wid)}: {kde.rect(wid)}") -mismatch = [m for m in (where_asked(w) for w in live_windows()) if m is not None] -for m in mismatch: - print(f" {m}") -checks.check("every window is where komorebi last asked for it", mismatch == []) -kde.komorebic("stop") -checks.check("a clean stop", kde.wait_for(lambda: not kde.komorebi_running())) - kde.stop() checks.finish()