Artemis-Plugins/tests/cua_shell_smoke.py

"""Opt-in native-menu and Win+B regression, under the host desktop lease."""

import argparse
import asyncio
import base64
import hashlib
import json
from pathlib import Path
import re
import subprocess
import tempfile

from artemis.plugins.models import ServerConfig
from artemis.plugins.session import PluginSession
from artemis.runtime.device_lock import DeviceExecutionLock


async def run(executable: Path, output: Path, taskbar: bool):
    output.mkdir(parents=True, exist_ok=True)
    with tempfile.TemporaryDirectory(prefix="artemis-shell-fixture-") as folder:
        state_file = Path(folder) / "state.json"
        proc = await asyncio.create_subprocess_exec(
            "powershell.exe",
            "-NoProfile",
            "-File",
            str(Path(__file__).with_name("shell-fixture.ps1")),
            "-StateFile",
            str(state_file),
            creationflags=subprocess.CREATE_NO_WINDOW,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.PIPE,
        )
        session = None
        calls = []
        try:
            async with asyncio.timeout(15):
                while not state_file.exists():
                    if proc.returncode is not None:
                        raise RuntimeError((await proc.stderr.read()).decode(errors="replace"))
                    await asyncio.sleep(0.1)
            state = json.loads(state_file.read_text(encoding="utf-8-sig"))
            target = {"pid": state["pid"], "window_id": state["hwnd"]}
            session = await PluginSession(
                ServerConfig(command=str(executable)), executable.parent
            ).start()

            async def call(name, args):
                result = await session.call(name, args)
                calls.append(
                    {
                        "name": name,
                        "arguments": args,
                        "result": result.model_dump(mode="json", by_alias=True, exclude_none=True),
                    }
                )
                assert not result.isError, "\n".join(
                    block.text[:1500] for block in result.content if block.type == "text"
                )
                return result

            def tree(result):
                return "\n".join(getattr(block, "text", "") for block in result.content)

            result = await call("get_window_state", target)
            match = re.search(r'\[(\d+)\][^\n]*"Open native menu"', tree(result))
            assert match, tree(result)
            opened = await call(
                "click",
                {
                    **target,
                    "element_token": result.structuredContent["snapshot_id"] + ":" + match[1],
                    "count": 1,
                    "delivery_mode": "foreground",
                },
            )
            observation = opened.structuredContent["observation_target"]
            menu = observation["observed"]
            assert menu["window_id"] != target["window_id"], observation
            assert observation["reason"] == "associated_native_keyboard_focus", observation
            assert opened.structuredContent["artemis"]["source"]["target_id"] == str(
                menu["window_id"]
            )
            observed = await call("get_window_state", menu)
            assert "Increment once" in tree(observed), tree(observed)
            png = next(block for block in observed.content if block.type == "image")
            image = base64.b64decode(png.data, validate=True)
            (output / "native-menu.png").write_bytes(image)
            listed = await call("list_windows", {"pid": proc.pid})
            assert any(
                window["window_id"] == menu["window_id"] and window.get("window_kind") == "menu"
                for window in listed.structuredContent["windows"]
            )
            # Escape belongs to the owner queue; no fixture or user business action.
            await call("press_key", {**target, "key": "escape", "delivery_mode": "foreground"})
            final_state = json.loads(state_file.read_text(encoding="utf-8-sig"))
            assert final_state["selected_count"] == 0, final_state
            assert final_state["menu_attempts"] == 1, final_state

            shell_verified = False
            if taskbar:
                windows = await call("list_windows", {"app": "explorer", "on_screen_only": True})
                desktop = next(
                    window
                    for window in windows.structuredContent["windows"]
                    if window.get("class_name") == "Progman"
                )
                desktop = {key: desktop[key] for key in ("pid", "window_id")}
                await call("get_window_state", desktop)
                key_result = await call(
                    "press_key",
                    {**desktop, "key": "b", "modifiers": ["win"], "delivery_mode": "foreground"},
                )
                switched = key_result.structuredContent["observation_target"]
                assert switched["observed"]["window_id"] != desktop["window_id"], switched
                assert switched["reason"] == "associated_native_keyboard_focus", switched
                assert key_result.structuredContent["post_action_focus"]["focus_root"][
                    "class_name"
                ] in {"Shell_TrayWnd", "Shell_SecondaryTrayWnd"}
                assert key_result.structuredContent["artemis"]["source"]["target_id"] == str(
                    switched["observed"]["window_id"]
                )
                shell_verified = True
            await call("bring_to_front", target)
            report = {
                "native_menu_discovered_and_captured": True,
                "menu_source_matches_actual_hwnd": True,
                "menu_png_sha256": hashlib.sha256(image).hexdigest(),
                "win_b_focus_and_capture": shell_verified,
                "fixture_business_submissions": 0,
                "calls": len(calls),
            }
            (output / "summary.json").write_text(json.dumps(report, indent=2), encoding="utf-8")
            print(json.dumps(report))
        finally:
            # Save only this fixture's protocol evidence; no pairing payloads.
            (output / "calls.json").write_text(
                json.dumps(calls, ensure_ascii=False, indent=2), encoding="utf-8"
            )
            if session:
                await session.aclose()
            if proc.returncode is None:
                proc.terminate()
                await proc.wait()
            if state_file.exists():
                (output / "fixture-state.json").write_text(
                    state_file.read_text(encoding="utf-8-sig"), encoding="utf-8"
                )


if __name__ == "__main__":
    parser = argparse.ArgumentParser(__doc__)
    parser.add_argument("executable", type=Path)
    parser.add_argument("--output", type=Path, required=True)
    parser.add_argument("--taskbar", action="store_true")
    args = parser.parse_args()
    lock = DeviceExecutionLock(
        "cua-shell-regression", description="Cua shell regression", resource_ids=["desktop:local"]
    )
    lock.acquire(timeout=0)
    try:
        asyncio.run(run(args.executable.resolve(), args.output.resolve(), args.taskbar))
    finally:
        lock.release()