fix(nix): the shared core: neutral names, E2E_ONLY, waits and diagnostics

This commit is contained in:
LGUG2Z committed 2026-10-30 14:07:21 -07:00
1 parent 225bf22a07
commit 4920650155
12 files changed
+124 -36

No files matched your search

+2 -2
View File
@@ -387,14 +387,14 @@ the fake and fails on KDE has found a bug in the fake, the library or the daemon
`nix build .#komorebi-full`, and what a user installs, still has the real commit.
- **Each step says how long it took.** `run_suite` prints `[scope] module.function: 12.3 s` after
every step, so a log shows which scenario the time went to.
- **Run only some of the steps.** Set `KFDE_ONLY` to a comma-separated list, and only the steps
- **Run only some of the steps.** Set `E2E_ONLY` (`KFDE_ONLY` is the older name, still read) to a comma-separated list, and only the steps
whose name (as printed, `module.function`) contains one of the words run; the others are printed
as left out. A step that the others depend on (the one that opens the windows) is made with
`always=True` and runs whatever the list says. Evaluate with `--impure` for the variable to reach
the test:
```sh
KFDE_ONLY=stacks,monocle nix build --impure .#checks.x86_64-linux.vm-core-e2e -L
E2E_ONLY=stacks,monocle nix build --impure .#checks.x86_64-linux.vm-core-e2e -L
```
A step that is left out returns `None`, so a suite that uses what a step returned must run that
+8 -4
View File
@@ -23,9 +23,13 @@ let
in
if path == "" then null else builtins.storePath path;
# `KFDE_ONLY=stacks,monocle` (with `--impure`) runs only the steps of a suite whose name has one
# of the words; see `komorebi_vm.core.suite`
only = builtins.getEnv "KFDE_ONLY";
# `E2E_ONLY=stacks,monocle` (with `--impure`; `KFDE_ONLY` is the older name) runs only the steps of
# a suite whose name has one of the words; see `komorebi_vm.core.suite`
only =
let
named = builtins.getEnv "E2E_ONLY";
in
if named != "" then named else builtins.getEnv "KFDE_ONLY";
# the library the suites import, built for the python of the test driver
komorebiVm =
@@ -107,7 +111,7 @@ let
''
+ lib.optionalString (only != "") ''
import os
os.environ["KFDE_ONLY"] = ${builtins.toJSON only}
os.environ["E2E_ONLY"] = ${builtins.toJSON only}
''
+ builtins.readFile (dir + "/${suite.script or "test.py"}");
};
@@ -1,5 +1,8 @@
"""What does not depend on the platform: the vocabulary and the helpers a scenario is made of.
"""The core the ports share: the vocabulary and the helpers a scenario is made of.
Nothing in this package may import from `komorebi_vm.kde` or know the name of a compositor,
so that it can move to the Windows port unchanged.
kfde (KDE) and kfw (Windows) run the same scenarios, so this package is the same in both
repositories (kfw vendors it with `scripts/sync-e2e-core`). Nothing in it may import from a port's
own code (`komorebi_vm.kde`, `komorebi_vm.platforms`, a runner) or name a compositor, a window
class or a tool of one port; what a port does differently is in its `Platform`, and what only a
port can check is in its extras.
"""
+4 -3
View File
@@ -19,8 +19,8 @@ class Checks:
self.skipped: list[str] = []
self.known: list[str] = []
self.scope = ""
"""Where the checks being made come from (`shared`, `kde`, `windows`), or empty. It prefixes
the description of each check, and a subclass that records checks may read it."""
"""Where the checks being made come from (`shared`, or the name of a port), or empty. It
prefixes the description of each check, and a subclass that records checks may read it."""
@contextmanager
def scoped(self, scope: str) -> Generator[None]:
@@ -33,7 +33,8 @@ class Checks:
self.scope = before
def labelled(self, description: str) -> str:
"""The description with the scope in front of it, `[kde] ...`, when there is one."""
"""The description with the scope in front of it, `[shared] ...` or `[<port>] ...`, when
there is one."""
return f"[{self.scope}] {description}" if self.scope else description
def ok(self, description: str) -> None:
+2 -2
View File
@@ -39,8 +39,8 @@ def ring(image: Image.Image, frame: Rect, color: Color, margin: int = 5) -> list
"""Returns what is wrong with the ring around a window: a line for each side that has none.
The ring is looked for `margin` pixels outside the middle of each of the four edges of
`frame`, which is the window with its title bar, as KWin reports it. A side where that pixel is
not near `color` is named, with the pixel it has. An empty list is a ring all round.
`frame`, which is the window with its title bar, as the platform reports it. A side where that
pixel is not near `color` is named, with the pixel it has. An empty list is a ring all round.
"""
middle_x = frame.x + frame.w // 2
middle_y = frame.y + frame.h // 2
@@ -3,7 +3,7 @@ again."""
from komorebi_vm.core.checks import Checks
from komorebi_vm.core.platform import Client, Platform
from komorebi_vm.core.scenarios.support import focus_agrees
from komorebi_vm.core.scenarios.support import describe, focus_agrees
from komorebi_vm.core.scenarios.tiling import _where_the_layout_says
@@ -56,9 +56,10 @@ def run(platform: Platform, checks: Checks, clients: list[Client]) -> None:
window is not None and window.frame.inside(area, 2) and not _on_a_tile(platform, target)
)
checks.check(
if not checks.check(
f"{title} is inside the work area and not on a tile of the layout",
platform.wait_for(left_alone, timeout=10),
)
):
print(describe(platform, clients))
platform.komorebic("manage")
checks.check(f"manage: {title} is tiled again", platform.wait_for(has_it, timeout=10))
@@ -2,7 +2,7 @@
from komorebi_vm.core.checks import Checks
from komorebi_vm.core.platform import Client, Platform
from komorebi_vm.core.scenarios.support import focus_agrees
from komorebi_vm.core.scenarios.support import describe, focus_agrees
def run(platform: Platform, checks: Checks, clients: list[Client]) -> None:
@@ -54,11 +54,29 @@ def crash(platform: Platform, checks: Checks, clients: list[Client]) -> None:
"""Turns a monocle on, which puts the other windows away, kills the daemon, and starts it
again: nothing is left out of sight for the new daemon to find, and it manages every window.
`clients` are two tiled windows or more of the first workspace, each in a container of its own.
The monocle is off at the end.
`clients` are two windows or more, each in a container of its own. They are gathered on the
first workspace, which is shown, first: a window another scenario left on a workspace that is
not in view would be out of sight after the restart on a platform that keeps it there. The
monocle is off at the end.
"""
target, other = clients[-1], clients[0]
def gather() -> None:
platform.komorebic("focus-workspace", "0")
for client in clients:
home = platform.state().workspace_of(client.title)
if home not in (0, None):
platform.komorebic("focus-workspace", str(home))
platform.activate(client)
platform.komorebic("move-to-workspace", "0")
platform.komorebic("focus-workspace", "0")
platform.wait_for(
lambda: all(platform.state().workspace_of(c.title) == 0 for c in clients), timeout=20
)
platform.wait_for(lambda: not any(platform.hidden(c) for c in clients), timeout=20)
gather()
def in_front() -> None:
# a window that already has the focus gives the daemon no event: another one first
platform.activate(other)
@@ -87,10 +105,11 @@ def crash(platform: Platform, checks: Checks, clients: list[Client]) -> None:
"the next daemon manages every window",
platform.wait_for(lambda: platform.state().managed() >= len(clients), timeout=30),
)
checks.check(
if not checks.check(
"... and none is left out of sight",
platform.wait_for(lambda: not any(platform.hidden(c) for c in clients), timeout=20),
)
):
print(describe(platform, clients))
area = platform.work_area()
frames = [w.frame for c in clients if (w := platform.os_window(c)) is not None]
checks.check(
@@ -107,7 +126,8 @@ def crash(platform: Platform, checks: Checks, clients: list[Client]) -> None:
platform.wait_for(only_target_in_view, timeout=20),
)
platform.komorebic("toggle-monocle")
checks.check(
if not checks.check(
"... and turned off, nothing is out of sight",
platform.wait_for(lambda: not any(platform.hidden(c) for c in clients), timeout=20),
)
):
print(describe(platform, clients))
@@ -3,6 +3,7 @@
from komorebi_vm.core.checks import Checks
from komorebi_vm.core.geometry import Rect
from komorebi_vm.core.platform import Client, Platform
from komorebi_vm.core.scenarios.layouts import tiled
# how far outside a window's frame its resize border is pressed: the gap between two tiles
OFFSET = 4
@@ -233,6 +234,8 @@ def _problems(
end = (max(area.x, px + dx), max(area.y, py + dy))
platform.pointer.drag((px, py), end)
platform.wait_for(lambda: _tiles(platform) != tiles_before, timeout=3)
# a neighbour's frame follows its tile a moment later; it is read when it has
platform.wait_for(lambda: tiled(platform, clients), timeout=10)
after = _settled(platform, clients)
tiles_after = _tiles(platform)
@@ -1,5 +1,5 @@
"""Rules that match a window by its class (its app id on Wayland) decide where it goes, or that it
is left alone."""
"""Rules that match a window by its class (the app id on Wayland, the window class on Windows)
decide where it goes, or that it is left alone."""
from komorebi_vm.core.checks import Checks
from komorebi_vm.core.platform import Platform
@@ -9,6 +9,21 @@ from komorebi_vm.core.platform import Client, Platform
QUIET = 3
def describe(platform: Platform, clients: list[Client]) -> str:
"""What the daemon and the system say of each of `clients`, for a failed check to print."""
state = platform.state()
layout = [str(r) for r in state.workspace().latest_layout]
lines = [f" work area {platform.work_area()}; the layout {layout}"]
for client in clients:
window = platform.os_window(client)
frame = window.frame if window is not None else None
lines.append(
f" {client.title}: workspace {state.workspace_of(client.title)},"
f" hidden {platform.hidden(client)}, frame {frame}"
)
return "\n".join(lines)
def os_focused(platform: Platform, clients: list[Client]) -> str:
"""The title of the window the operating system says has the focus, among `clients`.
+15 -7
View File
@@ -2,12 +2,13 @@
A suite file says what runs, in order, and each step labels the checks it makes: `shared` for a
scenario that runs on every platform, and a platform's own name for what only that platform can
check. The output reads `[shared] ...` and `[kde] ...`, and the same labels say, for each check,
check. The output reads `[shared] ...` and `[<port>] ...`, and the same labels say, for each check,
which port it belongs to.
Each step prints the time it took, so that a log says which scenario the time went to.
To run only some of the steps, set `KFDE_ONLY` to a comma-separated list: a step runs when its
To run only some of the steps, set `E2E_ONLY` (or `KFDE_ONLY`, its older name) to a
comma-separated list: a step runs when its
name (`module.function`, as printed) contains one of the words, and a step made with `always=True`
(what the others depend on, such as the one that opens the windows) runs whatever the list says.
The steps that do not run are printed, and return `None`.
@@ -20,7 +21,9 @@ from dataclasses import dataclass
from komorebi_vm.core.checks import Checks
ONLY = "KFDE_ONLY"
# the variable that names the steps to run, and the older name of it
ONLY = "E2E_ONLY"
OLD_ONLY = "KFDE_ONLY"
@dataclass(frozen=True)
@@ -35,7 +38,7 @@ class Step:
def _name(fn: Callable[..., object]) -> str:
"""`module.function` of a scenario or an extra, for the line that says what runs."""
module = getattr(fn, "__module__", "").rsplit(".", 1)[-1]
module = (getattr(fn, "__module__", None) or "").rsplit(".", 1)[-1]
return f"{module}.{getattr(fn, '__name__', '')}".strip(".")
@@ -51,20 +54,25 @@ def platform_only(
return Step(scope, lambda: extra(*args), _name(extra), always)
def _variable() -> str:
"""The name of the variable that is set: `E2E_ONLY` before `KFDE_ONLY`."""
return ONLY if os.environ.get(ONLY) else OLD_ONLY
def _wanted() -> list[str]:
return [word.strip() for word in os.environ.get(ONLY, "").split(",") if word.strip()]
return [word.strip() for word in os.environ.get(_variable(), "").split(",") if word.strip()]
def run_suite(checks: Checks, *steps: Step) -> list[object]:
"""Runs the steps in order, each with its scope on `checks`, and returns what each returned,
so that a later part of a suite can use the windows a scenario opened.
A step that `KFDE_ONLY` leaves out is not run. A step that is run prints how long it took."""
A step that the list leaves out is not run. A step that is run prints how long it took."""
wanted = _wanted()
results: list[object] = []
for step in steps:
if wanted and not step.always and not any(word in step.name for word in wanted):
print(f"== [{step.scope}] {step.name}: left out by {ONLY}")
print(f"== [{step.scope}] {step.name}: left out by {_variable()}")
results.append(None)
continue
if step.name:
+36 -3
View File
@@ -103,7 +103,7 @@ def test_a_step_prints_how_long_it_took(capsys: pytest.CaptureFixture[str]) -> N
def test_steps_not_named_in_the_list_are_left_out(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
monkeypatch.setenv("KFDE_ONLY", "second, nothing")
monkeypatch.setenv("E2E_ONLY", "second, nothing")
ran: list[str] = []
def first() -> str:
@@ -126,11 +126,44 @@ def test_steps_not_named_in_the_list_are_left_out(
)
assert ran == ["setup", "second"]
assert results == ["c", None, "b"]
assert "test_suite.first: left out by KFDE_ONLY" in capsys.readouterr().out
assert "test_suite.first: left out by E2E_ONLY" in capsys.readouterr().out
def test_with_no_list_every_step_runs(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("KFDE_ONLY", raising=False)
monkeypatch.delenv("E2E_ONLY", raising=False)
ran: list[int] = []
run_suite(Checks(), shared(lambda: ran.append(1)), shared(lambda: ran.append(2)))
assert ran == [1, 2]
def test_the_older_name_of_the_variable_still_works(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
monkeypatch.delenv("E2E_ONLY", raising=False)
monkeypatch.setenv("KFDE_ONLY", "second")
ran: list[str] = []
def first() -> None:
ran.append("first")
def second() -> None:
ran.append("second")
run_suite(Checks(), shared(first), shared(second))
assert ran == ["second"]
assert "test_suite.first: left out by KFDE_ONLY" in capsys.readouterr().out
def test_the_new_name_wins_over_the_older_one(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("E2E_ONLY", "second")
monkeypatch.setenv("KFDE_ONLY", "first")
ran: list[str] = []
def first() -> None:
ran.append("first")
def second() -> None:
ran.append("second")
run_suite(Checks(), shared(first), shared(second))
assert ran == ["second"]