Add isolated window screenshot targets with native selection
This commit is contained in:
Samuel Onoja committed
1 parent
c8ed8e6c30
commit
0ab5f2e05b
12 files changed
+681
-95
No files matched your search
+91
@@ -0,0 +1,91 @@
|
||||
/* Two overlapping clients with distinct colors for isolated capture tests. */
|
||||
#define _GNU_SOURCE
|
||||
#include <stdint.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <sys/mman.h>
|
||||
#include <unistd.h>
|
||||
#include <wayland-client.h>
|
||||
#include "xdg-shell-client-protocol.h"
|
||||
|
||||
static struct wl_compositor *compositor;
|
||||
static struct wl_shm *shm;
|
||||
static struct xdg_wm_base *wm;
|
||||
static struct wl_surface *surface;
|
||||
static struct wl_buffer *buffer;
|
||||
|
||||
static void ping(void *data, struct xdg_wm_base *base, uint32_t serial) {
|
||||
(void)data;
|
||||
xdg_wm_base_pong(base, serial);
|
||||
}
|
||||
static const struct xdg_wm_base_listener wm_listener = {.ping = ping};
|
||||
|
||||
static void global(void *data, struct wl_registry *registry, uint32_t name,
|
||||
const char *interface, uint32_t version) {
|
||||
(void)data; (void)version;
|
||||
if (!strcmp(interface, "wl_compositor"))
|
||||
compositor = wl_registry_bind(registry, name, &wl_compositor_interface, 4);
|
||||
else if (!strcmp(interface, "wl_shm"))
|
||||
shm = wl_registry_bind(registry, name, &wl_shm_interface, 1);
|
||||
else if (!strcmp(interface, "xdg_wm_base")) {
|
||||
wm = wl_registry_bind(registry, name, &xdg_wm_base_interface, 1);
|
||||
xdg_wm_base_add_listener(wm, &wm_listener, NULL);
|
||||
}
|
||||
}
|
||||
|
||||
static void removed(void *data, struct wl_registry *registry, uint32_t name) {
|
||||
(void)data; (void)registry; (void)name;
|
||||
}
|
||||
static const struct wl_registry_listener registry_listener = {global, removed};
|
||||
|
||||
static void configure(void *data, struct xdg_surface *xdg, uint32_t serial) {
|
||||
(void)data;
|
||||
xdg_surface_ack_configure(xdg, serial);
|
||||
xdg_surface_set_window_geometry(xdg, 0, 0, 640, 480);
|
||||
wl_surface_attach(surface, buffer, 0, 0);
|
||||
wl_surface_damage(surface, 0, 0, 640, 480);
|
||||
wl_surface_commit(surface);
|
||||
}
|
||||
static const struct xdg_surface_listener surface_listener = {.configure = configure};
|
||||
|
||||
static void size(void *data, struct xdg_toplevel *top, int32_t w, int32_t h,
|
||||
struct wl_array *states) {
|
||||
(void)data; (void)top; (void)w; (void)h; (void)states;
|
||||
}
|
||||
|
||||
static void close_window(void *data, struct xdg_toplevel *top) {
|
||||
(void)data; (void)top; exit(0);
|
||||
}
|
||||
static const struct xdg_toplevel_listener top_listener = {.configure = size, .close = close_window};
|
||||
|
||||
int main(int argc, char **argv) {
|
||||
struct wl_display *display = wl_display_connect(NULL);
|
||||
if (!display) return 1;
|
||||
struct wl_registry *registry = wl_display_get_registry(display);
|
||||
wl_registry_add_listener(registry, ®istry_listener, NULL);
|
||||
if (wl_display_roundtrip(display) < 0 || !compositor || !shm || !wm) return 2;
|
||||
int fd = memfd_create("floating-size-test", MFD_CLOEXEC);
|
||||
size_t bytes = 640 * 480 * 4;
|
||||
if (fd < 0 || ftruncate(fd, bytes)) return 3;
|
||||
uint32_t *pixels = mmap(NULL, bytes, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
|
||||
if (pixels == MAP_FAILED) return 4;
|
||||
int covered = argc > 1 && !strcmp(argv[1], "cover");
|
||||
for (size_t i = 0; i < 640 * 480; i++) {
|
||||
pixels[i] = covered ? 0xff00ff00 : (i % 640 < 80 ? 0x80800000 : (i / 640 < 240 ? 0xffff0000 : 0xff0000ff));
|
||||
}
|
||||
struct wl_shm_pool *pool = wl_shm_create_pool(shm, fd, bytes);
|
||||
buffer = wl_shm_pool_create_buffer(pool, 0, 640, 480, 640 * 4, WL_SHM_FORMAT_ARGB8888);
|
||||
wl_shm_pool_destroy(pool);
|
||||
close(fd);
|
||||
surface = wl_compositor_create_surface(compositor);
|
||||
struct xdg_surface *xdg = xdg_wm_base_get_xdg_surface(wm, surface);
|
||||
xdg_surface_add_listener(xdg, &surface_listener, NULL);
|
||||
struct xdg_toplevel *top = xdg_surface_get_toplevel(xdg);
|
||||
xdg_toplevel_add_listener(top, &top_listener, NULL);
|
||||
xdg_toplevel_set_app_id(top, "ferese.test.window-capture");
|
||||
xdg_toplevel_set_title(top, covered ? "Occluding window" : "Capture target");
|
||||
wl_surface_commit(surface);
|
||||
while (wl_display_dispatch(display) != -1) {}
|
||||
return 0;
|
||||
}
|
||||
@@ -0,0 +1,91 @@
|
||||
"""Verify isolated window capture without exposing host desktop content.
|
||||
|
||||
FERESE_TEST_WINDOW_CAPTURE=1 python3 scripts/tests/test_window_capture_isolated.py
|
||||
Requires built debug binaries, a Wayland host, cc, wayland-scanner and Pillow.
|
||||
"""
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import signal
|
||||
import subprocess
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
|
||||
REPO = Path(__file__).resolve().parents[2]
|
||||
|
||||
|
||||
@unittest.skipUnless(os.environ.get("FERESE_TEST_WINDOW_CAPTURE") == "1", "requires a Wayland host")
|
||||
class WindowCapture(unittest.TestCase):
|
||||
def test_occluding_window_does_not_appear_and_rows_keep_orientation(self):
|
||||
from PIL import Image
|
||||
processes = []
|
||||
with tempfile.TemporaryDirectory(prefix="ferese-window-capture-") as directory:
|
||||
root = Path(directory)
|
||||
protocols = subprocess.check_output(["pkg-config", "--variable=pkgdatadir", "wayland-protocols"], text=True).strip()
|
||||
xml = str(Path(protocols) / "stable/xdg-shell/xdg-shell.xml")
|
||||
for kind, output in [("client-header", "xdg-shell-client-protocol.h"), ("private-code", "xdg-shell-protocol.c")]:
|
||||
subprocess.run(["wayland-scanner", kind, xml, str(root / output)], check=True)
|
||||
flags = subprocess.check_output(["pkg-config", "--cflags", "--libs", "wayland-client"], text=True).split()
|
||||
subprocess.run(["cc", "-Wall", "-Wextra", "-I", str(root), str(REPO / "scripts/tests/fixtures/window-capture.c"),
|
||||
str(root / "xdg-shell-protocol.c"), "-o", str(root / "client"), *flags], check=True)
|
||||
runtime = root / "runtime"
|
||||
runtime.mkdir(mode=0o700)
|
||||
config = root / "config/ferese/config.kdl"
|
||||
config.parent.mkdir(parents=True)
|
||||
config.write_text('window-rule app-id="ferese.test.window-capture" floating=#true\n')
|
||||
display = Path(os.environ["WAYLAND_DISPLAY"])
|
||||
if not display.is_absolute():
|
||||
display = Path(os.environ["XDG_RUNTIME_DIR"]) / display
|
||||
env = dict(os.environ, XDG_RUNTIME_DIR=str(runtime), XDG_CONFIG_HOME=str(root / "config"), WAYLAND_DISPLAY=str(display), FERESE_ENABLE_SCREENCOPY="1")
|
||||
env.pop("WAYLAND_SOCKET", None)
|
||||
env.pop("FERESE_SHELL_CONTROL_SOCKET", None)
|
||||
log = (root / "compositor.log").open("w")
|
||||
try:
|
||||
compositor = subprocess.Popen([str(REPO / "target/debug/ferese"), "--backend=nested"], env=env, stdout=log, stderr=log, start_new_session=True)
|
||||
processes.append(compositor)
|
||||
deadline = time.monotonic() + 10
|
||||
while not (runtime / "ferese/control.sock").exists():
|
||||
if compositor.poll() is not None or time.monotonic() > deadline:
|
||||
self.fail((root / "compositor.log").read_text())
|
||||
time.sleep(0.02)
|
||||
sockets = list(runtime.glob("wayland-*"))
|
||||
env["WAYLAND_DISPLAY"] = str(next(path for path in sockets if not path.name.endswith(".lock")))
|
||||
|
||||
def call(*args):
|
||||
return subprocess.check_output([str(REPO / "target/debug/feresectl"), *args], env=env, timeout=10)
|
||||
|
||||
def windows(count):
|
||||
deadline = time.monotonic() + 10
|
||||
while time.monotonic() < deadline:
|
||||
result = json.loads(call("get-windows"))
|
||||
if len(result) == count and all(window["mapped"] and window["width"] == 640 for window in result):
|
||||
return result
|
||||
time.sleep(0.02)
|
||||
self.fail((root / "compositor.log").read_text())
|
||||
|
||||
processes.append(subprocess.Popen([str(root / "client")], env=env, stdout=log, stderr=log, start_new_session=True))
|
||||
target = windows(1)[0]
|
||||
processes.append(subprocess.Popen([str(root / "client"), "cover"], env=env, stdout=log, stderr=log, start_new_session=True))
|
||||
current = windows(2)
|
||||
cover = next(window for window in current if window["title"] == "Occluding window")
|
||||
self.assertEqual((target["x"], target["y"], target["width"], target["height"]),
|
||||
(cover["x"], cover["y"], cover["width"], cover["height"]))
|
||||
scale = json.loads(call("get-outputs"))[0]["scale"]
|
||||
with Image.open(io.BytesIO(call("screenshot-window", str(target["id"])))) as image:
|
||||
self.assertEqual(image.size, (round(640 * scale), round(480 * scale)))
|
||||
self.assertEqual(image.convert("RGB").getpixel((image.width // 2, image.height // 4)), (255, 0, 0))
|
||||
self.assertEqual(image.convert("RGB").getpixel((image.width // 2, image.height * 3 // 4)), (0, 0, 255))
|
||||
self.assertEqual(image.convert("RGBA").getpixel((round(40 * scale), round(80 * scale))), (255, 0, 0, 128))
|
||||
with Image.open(io.BytesIO(call("screenshot-window", str(cover["id"])))) as image:
|
||||
self.assertEqual(image.convert("RGB").getpixel((image.width // 2, image.height // 2)), (0, 255, 0))
|
||||
finally:
|
||||
for process in reversed(processes):
|
||||
os.killpg(process.pid, signal.SIGTERM)
|
||||
process.wait(timeout=5)
|
||||
log.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user