| 1 | from __future__ import annotations |
| 2 | |
| 3 | import subprocess |
| 4 | import struct |
| 5 | import sys |
| 6 | import types |
| 7 | from pathlib import Path |
| 8 | |
| 9 | import pytest |
| 10 | |
| 11 | PROJECT_ROOT = Path(__file__).resolve().parents[1] |
| 12 | if str(PROJECT_ROOT) not in sys.path: |
| 13 | sys.path.insert(0, str(PROJECT_ROOT)) |
| 14 | |
| 15 | from helpers import virtual_desktop |
| 16 | from plugins._desktop.helpers import desktop_state |
| 17 | |
| 18 | |
| 19 | def _completed(command, returncode=0, stdout="", stderr=""): |
| 20 | return subprocess.CompletedProcess(command, returncode, stdout, stderr) |
| 21 | |
| 22 | |
| 23 | def test_desktop_state_collects_x11_state_from_mocked_tools(tmp_path, monkeypatch): |
| 24 | session_dir = tmp_path / "sessions" |
| 25 | profile_dir = tmp_path / "profiles" / desktop_state.SESSION_ID |
| 26 | session_dir.mkdir(parents=True) |
| 27 | profile_dir.mkdir(parents=True) |
| 28 | (session_dir / f"{desktop_state.SESSION_ID}.json").write_text( |
| 29 | '{"display": 120, "profile_dir": "%s"}' % profile_dir, |
| 30 | encoding="utf-8", |
| 31 | ) |
| 32 | |
| 33 | monkeypatch.setattr(desktop_state, "SESSION_DIR", session_dir) |
| 34 | monkeypatch.setattr(desktop_state, "PROFILE_DIR", tmp_path / "profiles") |
| 35 | monkeypatch.setattr(desktop_state, "SCREENSHOT_DIR", tmp_path / "screenshots") |
| 36 | monkeypatch.setattr( |
| 37 | desktop_state.shutil, |
| 38 | "which", |
| 39 | lambda name: f"/usr/bin/{name}" |
| 40 | if name in {"xdotool", "xrandr", "xwininfo", "xprop", "xwd", "xclip"} |
| 41 | else "", |
| 42 | ) |
| 43 | |
| 44 | def fake_run(command, **kwargs): |
| 45 | name = Path(command[0]).name |
| 46 | if name == "xrandr": |
| 47 | return _completed(command, stdout="Screen 0: current 1440 x 900, maximum 1920 x 1080\n") |
| 48 | if name == "xdotool" and command[1:3] == ["getmouselocation", "--shell"]: |
| 49 | return _completed(command, stdout="X=12\nY=34\nSCREEN=0\nWINDOW=111\n") |
| 50 | if name == "xdotool" and command[1] == "getactivewindow": |
| 51 | return _completed(command, stdout="111\n") |
| 52 | if name == "xdotool" and command[1] == "search": |
| 53 | return _completed(command, stdout="111\n222\n") |
| 54 | if name == "xdotool" and command[1] == "getwindowname": |
| 55 | return _completed(command, stdout={"111": "LibreOffice Calc", "222": "Terminal"}[command[2]] + "\n") |
| 56 | if name == "xwininfo": |
| 57 | geometry = { |
| 58 | "111": (5, 7, 800, 600), |
| 59 | "222": (20, 30, 640, 480), |
| 60 | }[command[2]] |
| 61 | return _completed( |
| 62 | command, |
| 63 | stdout=( |
| 64 | f" Absolute upper-left X: {geometry[0]}\n" |
| 65 | f" Absolute upper-left Y: {geometry[1]}\n" |
| 66 | f" Width: {geometry[2]}\n" |
| 67 | f" Height: {geometry[3]}\n" |
| 68 | ), |
| 69 | ) |
| 70 | if name == "xprop": |
| 71 | window_id = command[2] |
| 72 | if window_id == "111": |
| 73 | return _completed( |
| 74 | command, |
| 75 | stdout='WM_CLASS(STRING) = "libreoffice", "libreoffice-calc"\n_NET_WM_PID(CARDINAL) = 4242\n', |
| 76 | ) |
| 77 | return _completed( |
| 78 | command, |
| 79 | stdout='WM_CLASS(STRING) = "xfce4-terminal", "Xfce4-terminal"\n_NET_WM_PID(CARDINAL) = 4343\n', |
| 80 | ) |
| 81 | raise AssertionError(f"unexpected command: {command}") |
| 82 | |
| 83 | monkeypatch.setattr(desktop_state.subprocess, "run", fake_run) |
| 84 | |
| 85 | state = desktop_state.collect_state() |
| 86 | |
| 87 | assert state["ok"] is True |
| 88 | assert state["display"] == ":120" |
| 89 | assert state["profile_dir"] == str(profile_dir) |
| 90 | assert state["size"] == {"width": 1440, "height": 900} |
| 91 | assert state["pointer"]["x"] == 12 |
| 92 | assert state["active_window"]["title"] == "LibreOffice Calc" |
| 93 | assert state["active_window"]["class"] == "libreoffice-calc" |
| 94 | assert state["active_window"]["geometry"]["width"] == 800 |
| 95 | assert [window["title"] for window in state["windows"]] == ["LibreOffice Calc", "Terminal"] |
| 96 | |
| 97 | |
| 98 | def test_desktop_state_allows_missing_active_window_when_display_is_reachable(tmp_path, monkeypatch): |
| 99 | session_dir = tmp_path / "sessions" |
| 100 | profile_dir = tmp_path / "profiles" / desktop_state.SESSION_ID |
| 101 | session_dir.mkdir(parents=True) |
| 102 | profile_dir.mkdir(parents=True) |
| 103 | (session_dir / f"{desktop_state.SESSION_ID}.json").write_text( |
| 104 | '{"display": 120, "profile_dir": "%s"}' % profile_dir, |
| 105 | encoding="utf-8", |
| 106 | ) |
| 107 | |
| 108 | monkeypatch.setattr(desktop_state, "SESSION_DIR", session_dir) |
| 109 | monkeypatch.setattr(desktop_state, "PROFILE_DIR", tmp_path / "profiles") |
| 110 | monkeypatch.setattr(desktop_state.shutil, "which", lambda name: f"/usr/bin/{name}") |
| 111 | |
| 112 | def fake_run(command, **kwargs): |
| 113 | del kwargs |
| 114 | name = Path(command[0]).name |
| 115 | if name == "xrandr": |
| 116 | return _completed(command, stdout="Screen 0: current 543 x 792, maximum 1920 x 1080\n") |
| 117 | if name == "xdotool" and command[1:3] == ["getmouselocation", "--shell"]: |
| 118 | return _completed(command, stdout="X=43\nY=244\nSCREEN=0\nWINDOW=18874412\n") |
| 119 | if name == "xdotool" and command[1] == "getactivewindow": |
| 120 | return _completed( |
| 121 | command, |
| 122 | returncode=1, |
| 123 | stderr="XGetWindowProperty[_NET_ACTIVE_WINDOW] failed (code=1)\n", |
| 124 | ) |
| 125 | if name == "xdotool" and command[1] == "search": |
| 126 | return _completed(command, stdout="18874412\n") |
| 127 | if name == "xdotool" and command[1] == "getwindowname": |
| 128 | return _completed(command, stdout="Desktop\n") |
| 129 | if name == "xwininfo": |
| 130 | return _completed( |
| 131 | command, |
| 132 | stdout=( |
| 133 | " Absolute upper-left X: 0\n" |
| 134 | " Absolute upper-left Y: 0\n" |
| 135 | " Width: 543\n" |
| 136 | " Height: 792\n" |
| 137 | ), |
| 138 | ) |
| 139 | if name == "xprop": |
| 140 | return _completed(command, stdout='WM_CLASS(STRING) = "xfdesktop", "Xfdesktop"\n') |
| 141 | raise AssertionError(f"unexpected command: {command}") |
| 142 | |
| 143 | monkeypatch.setattr(desktop_state.subprocess, "run", fake_run) |
| 144 | |
| 145 | state = desktop_state.collect_state() |
| 146 | |
| 147 | assert state["ok"] is True |
| 148 | assert state["active_window"] is None |
| 149 | assert state["errors"] == [] |
| 150 | assert [window["title"] for window in state["windows"]] == ["Desktop"] |
| 151 | |
| 152 | |
| 153 | def test_desktop_state_screenshot_capture_uses_xwd_and_pillow_when_available(tmp_path, monkeypatch): |
| 154 | monkeypatch.setattr(desktop_state, "SCREENSHOT_DIR", tmp_path) |
| 155 | capabilities = {"xwd": "/usr/bin/xwd"} |
| 156 | env = {"DISPLAY": ":120"} |
| 157 | |
| 158 | def fake_run(command, *, env, timeout): |
| 159 | raw_path = Path(command[command.index("-out") + 1]) |
| 160 | raw_path.write_bytes(b"xwd") |
| 161 | return _completed(command) |
| 162 | |
| 163 | image_module = types.ModuleType("PIL.Image") |
| 164 | |
| 165 | class FakeImage: |
| 166 | width = 320 |
| 167 | height = 240 |
| 168 | |
| 169 | def __enter__(self): |
| 170 | return self |
| 171 | |
| 172 | def __exit__(self, *_args): |
| 173 | return False |
| 174 | |
| 175 | def save(self, target): |
| 176 | Path(target).write_bytes(b"png") |
| 177 | |
| 178 | image_module.open = lambda _path: FakeImage() |
| 179 | pil_module = types.ModuleType("PIL") |
| 180 | pil_module.Image = image_module |
| 181 | |
| 182 | monkeypatch.setattr(desktop_state, "run", fake_run) |
| 183 | monkeypatch.setitem(sys.modules, "PIL", pil_module) |
| 184 | monkeypatch.setitem(sys.modules, "PIL.Image", image_module) |
| 185 | |
| 186 | screenshot = desktop_state.capture_screenshot(env, capabilities, path=tmp_path / "shot.png", errors=[]) |
| 187 | |
| 188 | assert screenshot["ok"] is True |
| 189 | assert screenshot["path"] == str(tmp_path / "shot.png") |
| 190 | assert screenshot["format"] == "png" |
| 191 | assert screenshot["ephemeral"] is False |
| 192 | assert (tmp_path / "shot.png").read_bytes() == b"png" |
| 193 | assert not (tmp_path / "shot.xwd").exists() |
| 194 | |
| 195 | |
| 196 | def test_desktop_state_shell_screenshot_path_is_context_scoped(tmp_path, monkeypatch): |
| 197 | monkeypatch.setattr(desktop_state, "BASE_DIR", tmp_path) |
| 198 | monkeypatch.setattr(desktop_state, "SCREENSHOT_DIR", tmp_path / "tmp" / "desktop" / "screenshots") |
| 199 | capabilities = {"xwd": "/usr/bin/xwd"} |
| 200 | env = {"DISPLAY": ":120"} |
| 201 | |
| 202 | def fake_run(command, *, env, timeout): |
| 203 | raw_path = Path(command[command.index("-out") + 1]) |
| 204 | raw_path.write_bytes(b"xwd") |
| 205 | return _completed(command) |
| 206 | |
| 207 | image_module = types.ModuleType("PIL.Image") |
| 208 | |
| 209 | class FakeImage: |
| 210 | width = 320 |
| 211 | height = 240 |
| 212 | |
| 213 | def __enter__(self): |
| 214 | return self |
| 215 | |
| 216 | def __exit__(self, *_args): |
| 217 | return False |
| 218 | |
| 219 | def save(self, target): |
| 220 | Path(target).write_bytes(b"png") |
| 221 | |
| 222 | image_module.open = lambda _path: FakeImage() |
| 223 | pil_module = types.ModuleType("PIL") |
| 224 | pil_module.Image = image_module |
| 225 | |
| 226 | monkeypatch.setattr(desktop_state, "run", fake_run) |
| 227 | monkeypatch.setitem(sys.modules, "PIL", pil_module) |
| 228 | monkeypatch.setitem(sys.modules, "PIL.Image", image_module) |
| 229 | stale_path = tmp_path / "tmp" / "desktop" / "screenshots" / "ctx_id" / "stale.png" |
| 230 | stale_path.parent.mkdir(parents=True) |
| 231 | stale_path.write_bytes(b"stale") |
| 232 | |
| 233 | screenshot = desktop_state.capture_screenshot( |
| 234 | env, |
| 235 | capabilities, |
| 236 | errors=[], |
| 237 | context_id="ctx/id", |
| 238 | transport="path", |
| 239 | ) |
| 240 | |
| 241 | path = Path(screenshot["path"]) |
| 242 | assert screenshot["ok"] is True |
| 243 | assert screenshot["ephemeral"] is False |
| 244 | assert screenshot["chat_scoped"] is True |
| 245 | assert screenshot["context_id"] == "ctx_id" |
| 246 | assert screenshot["a0_path"].startswith("/a0/usr/chats/ctx_id/screenshots/desktop/desktop-") |
| 247 | assert path.parent == tmp_path / "usr" / "chats" / "ctx_id" / "screenshots" / "desktop" |
| 248 | assert path.name.startswith("desktop-") |
| 249 | assert desktop_state.latest_screenshot(context_id="ctx/id")["path"] == str(path) |
| 250 | assert stale_path.exists() |
| 251 | |
| 252 | |
| 253 | def test_desktop_state_default_screenshot_returns_ephemeral_ref(tmp_path, monkeypatch): |
| 254 | monkeypatch.setattr(desktop_state, "SCREENSHOT_DIR", tmp_path) |
| 255 | capabilities = {"xwd": "/usr/bin/xwd"} |
| 256 | env = {"DISPLAY": ":120"} |
| 257 | |
| 258 | def fake_run(command, *, env, timeout): |
| 259 | raw_path = Path(command[command.index("-out") + 1]) |
| 260 | raw_path.write_bytes(b"xwd") |
| 261 | return _completed(command) |
| 262 | |
| 263 | image_module = types.ModuleType("PIL.Image") |
| 264 | |
| 265 | class FakeImage: |
| 266 | width = 320 |
| 267 | height = 240 |
| 268 | |
| 269 | def __enter__(self): |
| 270 | return self |
| 271 | |
| 272 | def __exit__(self, *_args): |
| 273 | return False |
| 274 | |
| 275 | def save(self, target): |
| 276 | Path(target).write_bytes(b"png") |
| 277 | |
| 278 | image_module.open = lambda _path: FakeImage() |
| 279 | pil_module = types.ModuleType("PIL") |
| 280 | pil_module.Image = image_module |
| 281 | |
| 282 | monkeypatch.setattr(desktop_state, "run", fake_run) |
| 283 | monkeypatch.setitem(sys.modules, "PIL", pil_module) |
| 284 | monkeypatch.setitem(sys.modules, "PIL.Image", image_module) |
| 285 | |
| 286 | screenshot = desktop_state.capture_screenshot( |
| 287 | env, |
| 288 | capabilities, |
| 289 | errors=[], |
| 290 | context_id="ctx/id", |
| 291 | ) |
| 292 | |
| 293 | assert screenshot["ok"] is True |
| 294 | assert screenshot["path"] == "" |
| 295 | assert screenshot["ephemeral"] is True |
| 296 | assert screenshot["ephemeral_ref"].startswith("a0-ephemeral-image://") |
| 297 | assert screenshot["vision_load"]["tool_args"]["paths"] == [screenshot["ephemeral_ref"]] |
| 298 | assert screenshot["context_id"] == "ctx_id" |
| 299 | assert not (tmp_path / "ctx_id").exists() |
| 300 | |
| 301 | |
| 302 | def test_desktop_prompt_context_recommends_structured_state_before_screenshots(): |
| 303 | context = desktop_state.compact_prompt_context( |
| 304 | { |
| 305 | "display": ":120", |
| 306 | "size": {"width": 1280, "height": 720}, |
| 307 | "pointer": {"x": 10, "y": 20}, |
| 308 | "active_window": {"title": "Terminal", "class": "Xfce4-terminal"}, |
| 309 | "windows": [{"title": "Terminal", "class": "Xfce4-terminal"}], |
| 310 | "screenshot": {}, |
| 311 | "context_id": "ctx_id", |
| 312 | "errors": [], |
| 313 | } |
| 314 | ) |
| 315 | |
| 316 | assert "state --json --context-id ctx_id for structured checks" in context |
| 317 | assert "observe --json --screenshot --context-id ctx_id before coordinate or visual-OCR actions" in context |
| 318 | assert "before any coordinate action" not in context |
| 319 | |
| 320 | |
| 321 | def test_virtual_desktop_system_display_normalization_rejects_portrait_viewports(): |
| 322 | assert virtual_desktop.normalize_desktop_display_size(395, 1080) == ( |
| 323 | virtual_desktop.DEFAULT_WIDTH, |
| 324 | virtual_desktop.DEFAULT_HEIGHT, |
| 325 | ) |
| 326 | assert virtual_desktop.normalize_desktop_display_size(1600, 900) == (1600, 900) |
| 327 | |
| 328 | |
| 329 | @pytest.mark.parametrize( |
| 330 | ("byte_order", "pixel_bytes"), |
| 331 | ( |
| 332 | (0, bytes.fromhex("0000ff00") + bytes.fromhex("00ff0000")), |
| 333 | (1, bytes.fromhex("00ff0000") + bytes.fromhex("0000ff00")), |
| 334 | ), |
| 335 | ) |
| 336 | def test_xwd_fallback_parser_handles_truecolor_pixels(tmp_path, byte_order, pixel_bytes): |
| 337 | from PIL import Image |
| 338 | |
| 339 | raw_path = tmp_path / "shot.xwd" |
| 340 | target = tmp_path / "shot.png" |
| 341 | header_values = [ |
| 342 | 100, # header_size |
| 343 | 7, # file_version |
| 344 | 2, # pixmap_format |
| 345 | 24, # pixmap_depth |
| 346 | 2, # pixmap_width |
| 347 | 1, # pixmap_height |
| 348 | 0, # xoffset |
| 349 | byte_order, |
| 350 | 32, # bitmap_unit |
| 351 | 1, # bitmap_bit_order |
| 352 | 32, # bitmap_pad |
| 353 | 32, # bits_per_pixel |
| 354 | 8, # bytes_per_line |
| 355 | 4, # visual_class: TrueColor |
| 356 | 0x00FF0000, # red_mask |
| 357 | 0x0000FF00, # green_mask |
| 358 | 0x000000FF, # blue_mask |
| 359 | 8, # bits_per_rgb |
| 360 | 256, # colormap_entries |
| 361 | 0, # ncolors |
| 362 | 2, # window_width |
| 363 | 1, # window_height |
| 364 | 0, # window_x |
| 365 | 0, # window_y |
| 366 | 0, # window_bdrwidth |
| 367 | ] |
| 368 | raw_path.write_bytes(struct.pack(">25I", *header_values) + pixel_bytes) |
| 369 | |
| 370 | converted = desktop_state.convert_xwd_to_image(raw_path, target) |
| 371 | |
| 372 | assert converted == {"width": 2, "height": 1} |
| 373 | with Image.open(target) as image: |
| 374 | assert image.mode == "RGB" |
| 375 | assert image.size == (2, 1) |
| 376 | assert list(image.getdata()) == [(255, 0, 0), (0, 255, 0)] |