diff --git a/e2e/platforms/tart_macos/run.py b/e2e/platforms/tart_macos/run.py new file mode 100644 index 00000000..77c7b2f2 --- /dev/null +++ b/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//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:])) diff --git a/e2e/platforms/tart_macos/tests/__init__.py b/e2e/platforms/tart_macos/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/e2e/platforms/tart_macos/tests/test_platform.py b/e2e/platforms/tart_macos/tests/test_platform.py new file mode 100644 index 00000000..e60d1e40 --- /dev/null +++ b/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)