feat(e2e): a runner for the macOS guest

python -m platforms.tart_macos.run GOLDEN BIN_DIR OUT SCENARIO... gives each scenario a fresh
clone of the golden image, uploads the programs under test, starts the daemon and runs the shared
scenario, on the checks, step loop and result file the Windows runner uses (komorebi_vm.runner).
The scenarios a guest cannot run (a second display, a pointer, a lock screen, a window class) are
listed with the reason and skipped, not failed.

The unit tests of the port need no guest: the ignore rules, which windows count as hidden, the
paths a scenario may ask for, and that every scenario the runner names is a module of the core.

tiling (10 checks) and focus (17) pass against a real guest.

e2e/platforms/tart_macos/run.py

@@ -0,0 +1,186 @@
+"""Runs neutral scenarios against a fresh macOS guest.
+
+usage: python -m platforms.tart_macos.run GOLDEN BIN_DIR OUT SCENARIO...
+
+GOLDEN is the name of the golden Tart image, BIN_DIR holds komorebi, komorebic and
+komorebi-testwin (built for macOS, see `build-bins.sh`), OUT gets the run's artifacts. Every
+scenario gets its own guest, a clone of the golden image. Those that take clients start from the
+windows that `tiling` opens.
+
+Every run writes OUT/<scenario>/result.json, naming the scenario, the port, the commit under test
+and each check, so that a session can be matched with its run.
+
+With `KOMOREBI_E2E_SCALE=1.5` every wait of a scenario is 1.5 times as long, for a busy host.
+With `KOMOREBI_E2E_LOG_LEVEL=debug` the daemon logs at that level.
+"""
+
+import importlib
+import json
+import os
+import sys
+import time
+import traceback
+from pathlib import Path
+
+from komorebi_vm.core.suite import shared
+from komorebi_vm.runner import RecordedChecks, commit, run_steps
+
+from .guest import Guest
+from .macos_platform import MacosPlatform
+
+BINARIES = ["komorebi", "komorebic", "komorebi-testwin"]
+
+# scenarios that take the windows `tiling` opens
+WITH_CLIENTS = {
+    "focus", "moving", "stacks", "workspaces", "resize", "monocle", "closing",
+    "commands", "lifecycle", "layouts", "floating", "minimizing", "padding",
+    "maximizing", "workspace_moves", "resize_commands", "promoting",
+    "flipping", "pausing", "tiling_off", "lifecycle_hidden",
+    "scrolling", "handover", "fullscreen",
+}
+
+# what a scenario file has besides `run`, which its suite runs after it
+MORE = {
+    "resize": ("edge_cases",),
+    "lifecycle_hidden": ("crash",),
+    "monocle": ("focus_behaviour",),
+}
+
+# scenarios a macOS guest cannot run, and why: they are skipped, not failed
+UNSUPPORTED = {
+    "hotplug": "a macOS guest has one display",
+    "unknown_monitor": "a macOS guest has one display",
+    "display_change": "a macOS guest has one display",
+    "monitors": "a macOS guest has one display",
+    "focus_follows_mouse": "a macOS guest has no pointer, and the daemon has no focus follows mouse",
+    "locking": "a macOS guest has no lock screen to test",
+    "wallpaper": "the wallpaper of a macOS guest is not read yet",
+    "rules_class": "a macOS window has no class",
+}
+
+
+class GuestDied(RuntimeError):
+    """The guest exited while a scenario was running: not what the daemon did."""
+
+
+def run_one(golden: str, bin_dir: Path, out: Path, name: str) -> RecordedChecks:
+    out.mkdir(parents=True, exist_ok=True)
+    scenario = importlib.import_module(f"komorebi_vm.core.scenarios.{name}")
+    tiling = importlib.import_module("komorebi_vm.core.scenarios.tiling")
+
+    checks = RecordedChecks()
+    print(f"== {name}")
+    began = time.monotonic()
+    timings: dict[str, float] = {}
+
+    def lap(phase: str) -> None:
+        now = time.monotonic()
+        timings[phase] = round(now - began - sum(timings.values()), 1)
+
+    with Guest(golden, out / "guest") as guest:
+        lap("guest")
+        guest.upload(bin_dir, BINARIES)
+        lap("upload")
+        platform = MacosPlatform(
+            guest.agent, out, scale=float(os.environ.get("KOMOREBI_E2E_SCALE", "1"))
+        )
+        try:
+            platform.start_komorebi(log_level=os.environ.get("KOMOREBI_E2E_LOG_LEVEL"))
+        except Exception:
+            _diagnose(platform, out)
+            raise
+        lap("daemon")
+        try:
+            if name in WITH_CLIENTS:
+                clients = tiling.run(platform, checks)
+                run_steps(
+                    checks,
+                    shared(scenario.run, platform, checks, clients),
+                    *(
+                        shared(getattr(scenario, more), platform, checks, clients)
+                        for more in MORE.get(name, ())
+                    ),
+                )
+            else:
+                run_steps(
+                    checks,
+                    shared(scenario.run, platform, checks),
+                    *(
+                        shared(getattr(scenario, more), platform, checks)
+                        for more in MORE.get(name, ())
+                    ),
+                )
+        except Exception as e:
+            died = guest.died()
+            if died:
+                # the guest is gone, so there is nothing to fetch from it
+                raise GuestDied(died) from e
+            _diagnose(platform, out)
+            raise
+        finally:
+            lap("scenario")
+            print(f"     seconds: {timings}")
+            try:
+                platform.screenshot(f"final-{name}")
+            except Exception as e:  # noqa: BLE001 - a screenshot is for the reader, not a result
+                print(f"     no screenshot: {e}")
+            (out / "result.json").write_text(json.dumps({
+                "scenario": name,
+                "port": "macos",
+                "commit": commit(),
+                "passed": checks.passed,
+                "failed": checks.failed,
+                "skipped": len(checks.skipped),
+                "known_failures": len(checks.known),
+                "start_retries": platform.start_notes,
+                "seconds": timings,
+                "steps": checks.steps,
+                "checks": checks.results,
+            }, indent=2))
+    return checks
+
+
+def _diagnose(platform: MacosPlatform, out: Path) -> None:
+    """After a failure: the daemon's output and log, the processes, and a screenshot, into `out`."""
+    try:
+        (out / "ps.txt").write_text(platform.agent_exec("ps", "-axo", "pid,etime,command")["stdout"])
+        platform.screenshot("failure")
+    except Exception as e:  # noqa: BLE001 - diagnosis must not hide the failure it is for
+        print(f"     no diagnosis: {e}")
+    for name, remote in (("komorebi.out", "/Users/admin/e2e/komorebi.out"),):
+        try:
+            (out / name).write_bytes(platform.agent_get_file(remote))
+        except Exception as e:  # noqa: BLE001
+            print(f"     no {name}: {e}")
+
+
+def main(argv: list[str]) -> int:
+    golden, bin_dir, out, *names = argv
+    failed = 0
+    for name in names:
+        if name in UNSUPPORTED:
+            print(f"== {name}: skipped, {UNSUPPORTED[name]}")
+            continue
+        try:
+            try:
+                checks = run_one(golden, Path(bin_dir), Path(out) / name, name)
+            except GuestDied as e:
+                # the guest itself went: not a result about the daemon, so the scenario runs again
+                print(f"{name}: THE GUEST DIED ({e}); running it again")
+                checks = run_one(golden, Path(bin_dir), Path(out) / name, name)
+        except Exception:  # noqa: BLE001 - a scenario that cannot run is a failure of its own
+            traceback.print_exc()
+            print(f"{name}: did not run, counted as 1 failed")
+            failed += 1
+            continue
+        print(f"{name}: passed {checks.passed} failed {checks.failed}")
+        if checks.skipped:
+            print(f"{name}: skipped {len(checks.skipped)}")
+        if checks.known:
+            print(f"{name}: known failures {len(checks.known)}")
+        failed += checks.failed
+    return 1 if failed else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))

e2e/platforms/tart_macos/tests/test_platform.py

@@ -0,0 +1,134 @@
+"""The macOS port, without a guest: what it decides before and after it asks the agent."""
+
+import importlib
+from pathlib import Path
+from typing import Any
+
+import pytest
+from komorebi_vm.core.platform import Client
+
+from platforms.tart_macos import run
+from platforms.tart_macos.macos_platform import (
+    IGNORED_APPS,
+    MacosPlatform,
+    with_ignored_apps,
+)
+
+
+class FakeAgent:
+    """An agent with a desktop of one 1280 by 800 display and the windows it is given."""
+
+    def __init__(self, windows: list[dict[str, Any]] | None = None) -> None:
+        self.windows = windows or []
+        self.calls: list[tuple[str, tuple, dict]] = []
+
+    def os_windows(self) -> list[dict[str, Any]]:
+        return self.windows
+
+    def displays(self) -> list[dict[str, Any]]:
+        frame = {"left": 0, "top": 0, "right": 1280, "bottom": 800}
+        return [{"monitor": frame, "work_area": {**frame, "top": 30}, "primary": True}]
+
+    def exec(self, *args: str) -> dict[str, Any]:
+        self.calls.append(("exec", args, {}))
+        return {"status": 0, "stdout": "", "stderr": ""}
+
+    def call(self, verb: str, **fields: Any) -> Any:
+        self.calls.append((verb, (), fields))
+        return True
+
+
+def window(title: str, left: int, top: int, w: int, h: int, minimized: bool = False) -> dict:
+    return {
+        "title": title, "minimized": minimized, "focused": False,
+        "frame": {"left": left, "top": top, "right": w, "bottom": h},
+    }
+
+
+def platform(agent: FakeAgent, tmp_path: Path) -> MacosPlatform:
+    return MacosPlatform(agent, tmp_path)  # type: ignore[arg-type]
+
+
+def test_the_guests_own_programs_are_left_to_themselves() -> None:
+    merged = with_ignored_apps({"monitors": []})
+    ids = [rule["id"] for rule in merged["ignore_rules"]]
+    assert ids == list(IGNORED_APPS)
+    assert all(rule["kind"] == "Exe" for rule in merged["ignore_rules"])
+
+
+def test_a_scenarios_own_ignore_rules_are_kept_and_not_changed() -> None:
+    mine = {"kind": "Exe", "id": "mine", "matching_strategy": "Equals"}
+    config = {"ignore_rules": [mine]}
+    merged = with_ignored_apps(config)
+    assert merged["ignore_rules"][0] == mine
+    assert config == {"ignore_rules": [mine]}
+
+
+def test_a_window_off_every_display_is_hidden(tmp_path: Path) -> None:
+    agent = FakeAgent([window("away", 5000, 5000, 400, 300)])
+    assert platform(agent, tmp_path).hidden(Client("away"))
+
+
+def test_a_window_on_the_display_is_not_hidden(tmp_path: Path) -> None:
+    agent = FakeAgent([window("here", 100, 100, 400, 300)])
+    assert not platform(agent, tmp_path).hidden(Client("here"))
+
+
+def test_a_window_the_user_minimized_is_not_hidden(tmp_path: Path) -> None:
+    agent = FakeAgent([window("min", 5000, 5000, 400, 300, minimized=True)])
+    assert not platform(agent, tmp_path).hidden(Client("min"))
+
+
+def test_a_window_that_does_not_exist_is_not_hidden(tmp_path: Path) -> None:
+    assert not platform(FakeAgent(), tmp_path).hidden(Client("none"))
+
+
+def test_the_frame_is_the_one_the_agent_gives(tmp_path: Path) -> None:
+    agent = FakeAgent([window("w", 10, 40, 625, 333)])
+    seen = platform(agent, tmp_path).os_window(Client("w"))
+    assert seen is not None and (seen.frame.x, seen.frame.y, seen.frame.w, seen.frame.h) == (
+        10, 40, 625, 333,
+    )
+
+
+def test_the_work_area_leaves_out_the_menu_bar(tmp_path: Path) -> None:
+    area = platform(FakeAgent(), tmp_path).work_area()
+    assert area.y == 30 and area.w == 1280
+
+
+@pytest.mark.parametrize("name", ["../up", "a/../b", "", "/abs"])
+def test_a_temp_path_cannot_leave_its_folder(name: str, tmp_path: Path) -> None:
+    with pytest.raises(ValueError):
+        platform(FakeAgent(), tmp_path).temp_path(name)
+
+
+def test_a_temp_path_is_in_the_temp_folder(tmp_path: Path) -> None:
+    path = platform(FakeAgent(), tmp_path).temp_path("sub/file.json")
+    assert path.endswith("/tmp/sub/file.json")
+
+
+def test_what_a_guest_cannot_do_is_not_offered(tmp_path: Path) -> None:
+    p = platform(FakeAgent(), tmp_path)
+    assert not p.has("pointer") and not p.has("lock")
+    assert p.resizable_display() is None
+    with pytest.raises(NotImplementedError):
+        p.add_display()
+
+
+def test_every_scenario_the_runner_names_exists() -> None:
+    named = run.WITH_CLIENTS | set(run.UNSUPPORTED) | set(run.MORE)
+    for name in sorted(named):
+        # `unknown_monitor` is a step of `commands`, as the Windows runner has it
+        module = "commands" if name == "unknown_monitor" else name
+        importlib.import_module(f"komorebi_vm.core.scenarios.{module}")
+
+
+def test_the_extra_steps_of_a_scenario_exist() -> None:
+    for name, extra in run.MORE.items():
+        module = importlib.import_module(f"komorebi_vm.core.scenarios.{name}")
+        for step in extra:
+            assert callable(getattr(module, step)), f"{name}.{step}"
+
+
+def test_a_scenario_is_not_both_supported_and_not() -> None:
+    assert not set(run.UNSUPPORTED) & set(run.WITH_CLIENTS)