143 lines
5.7 KiB
Python
143 lines
5.7 KiB
Python
"""Opt-in Windows smoke: opens two disposable fixture windows, never user apps.
|
|
|
|
Run with ARTEMIS's Python environment and the built Windows executable path.
|
|
"""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import base64
|
|
import io
|
|
import re
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from mcp import ClientSession, StdioServerParameters
|
|
from mcp.client.stdio import stdio_client
|
|
from PIL import Image
|
|
|
|
|
|
async def main(executable: Path):
|
|
processes = []
|
|
with tempfile.TemporaryDirectory(prefix="artemis-capture-smoke-") as directory:
|
|
folder = Path(directory)
|
|
title = "ARTEMIS-Capture-" + uuid4().hex[:8]
|
|
|
|
async def launch(cover=False):
|
|
ready = folder / ("cover.ready" if cover else "target.ready")
|
|
command = [
|
|
"powershell.exe",
|
|
"-NoProfile",
|
|
"-NonInteractive",
|
|
"-File",
|
|
str(Path(__file__).with_name("capture-fixture.ps1")),
|
|
"-Title",
|
|
title + ("-cover" if cover else ""),
|
|
"-ReadyFile",
|
|
str(ready),
|
|
]
|
|
if cover:
|
|
command.append("-Cover")
|
|
process = await asyncio.create_subprocess_exec(
|
|
*command,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
creationflags=subprocess.CREATE_NO_WINDOW,
|
|
)
|
|
processes.append(process)
|
|
for _ in range(200):
|
|
if ready.exists():
|
|
await asyncio.sleep(0.5)
|
|
return
|
|
if process.returncode is not None:
|
|
raise RuntimeError("Fixture failed to start")
|
|
await asyncio.sleep(0.05)
|
|
raise TimeoutError("Fixture was not ready")
|
|
|
|
try:
|
|
await launch()
|
|
async with (
|
|
stdio_client(
|
|
StdioServerParameters(command=str(executable), args=["mcp"])
|
|
) as streams,
|
|
ClientSession(*streams) as session,
|
|
):
|
|
await session.initialize()
|
|
|
|
async def observe():
|
|
# PID is discovered from our owned fixture; never target another app.
|
|
result = await session.call_tool(
|
|
"get_app_state", {"app": str(processes[0].pid)}
|
|
)
|
|
assert not result.isError, result
|
|
block = next((b for b in result.content if b.type == "image"), None)
|
|
assert block is not None, (
|
|
"Target window did not return a usable image"
|
|
)
|
|
with Image.open(
|
|
io.BytesIO(base64.b64decode(block.data))
|
|
) as picture:
|
|
rgb = picture.convert("RGB")
|
|
pixels = list(rgb.getdata())
|
|
green = sum(g > 220 and r < 40 and b < 40 for r, g, b in pixels)
|
|
magenta = sum(
|
|
r > 220 and b > 220 and g < 40 for r, g, b in pixels
|
|
)
|
|
assert green / len(pixels) > 0.5, (
|
|
"Capture lost the target surface"
|
|
)
|
|
assert magenta == 0, "Capture included the overlapping window"
|
|
content = "\n".join(
|
|
b.text for b in result.content if b.type == "text"
|
|
)
|
|
bounds = re.search(
|
|
r"Capture coordinate target[^\n]*Frame:.*x: (\d+), y: (\d+), width: (\d+), height: (\d+)",
|
|
content,
|
|
)
|
|
assert bounds, (
|
|
"Fixture button is missing from the accessibility tree"
|
|
)
|
|
x, y, w, h = map(int, bounds.groups())
|
|
r, g, b = rgb.getpixel((x + 8, y + h // 2))
|
|
assert r > 180 and g < 80 and b < 80, (
|
|
"Accessibility coordinates do not match screenshot pixels at this DPI"
|
|
)
|
|
return picture.size, (x + w // 2, y + h // 2)
|
|
|
|
before = await observe()
|
|
await launch(cover=True)
|
|
after = await observe()
|
|
assert before == after
|
|
clicked = await session.call_tool(
|
|
"click",
|
|
{
|
|
"app": str(processes[0].pid),
|
|
"click_method": "app_post",
|
|
"x": after[1][0],
|
|
"y": after[1][1],
|
|
},
|
|
)
|
|
assert not clicked.isError, clicked
|
|
for _ in range(40):
|
|
if (folder / "target.ready.clicked").exists():
|
|
break
|
|
await asyncio.sleep(0.05)
|
|
assert (folder / "target.ready.clicked").read_text() == "clicked", (
|
|
"Click did not reach the target control"
|
|
)
|
|
print(
|
|
f"PASS: target surface preserved behind overlapping fixture ({before[0][0]}x{before[0][1]}), coordinate alignment and child-control click verified"
|
|
)
|
|
finally:
|
|
for process in reversed(processes):
|
|
if process.returncode is None:
|
|
process.terminate()
|
|
await asyncio.wait_for(process.wait(), timeout=10)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("executable", type=Path)
|
|
asyncio.run(main(parser.parse_args().executable.resolve()))
|