starred/komorebi
1
0 Fork 0
mirror of https://github.com/LGUG2Z/komorebi.git synced 2026-11-15 05:12:28 +01:00
Code Issues 87 Packages Projects Releases 11 Wiki Activity

feat(nix): commands as shared steps and KDE extras

Browse Source
This commit is contained in:
LGUG2Z committed 2026-10-30 12:03:01 -07:00
1 parent b09011dce7
commit bc5f4babc4
2 files changed
+319 -256
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

nix/tests/vm/lib/komorebi_vm/platforms/kde/extras/commands.py
+291
View File
@@ -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()))
nix/tests/vm/suites/commands/test.py
+28 -256
View File
@@ -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()