Artemis-Plugins/tests/browser_smoke.py

"""Opt-in isolated Chrome form test against the packaged Windows MCP server."""

import argparse
import asyncio
import base64
import io
import json
import re
import subprocess
import tempfile
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

import psutil
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from PIL import Image


async def main(executable, chrome, occlude=False):
    received = []

    class Handler(BaseHTTPRequestHandler):
        def log_message(self, *args):
            pass

        def do_GET(self):
            body = b"""<!doctype html><meta charset="utf-8"><title>ARTEMIS Browser Fixture</title>
            <style>body{font:24px sans-serif;margin:40px}input,button{font:inherit;padding:12px;display:block;margin:20px 0}</style>
            <h1>ARTEMIS Browser Fixture</h1><label>Test parameter<input aria-label="Test parameter" id="value"></label>
            <button aria-label="Apply parameter" onclick="fetch('/save',{method:'POST',body:document.getElementById('value').value}).then(()=>{document.getElementById('result').textContent='Saved: '+document.getElementById('value').value;document.body.style.backgroundColor='#10e020'})">Apply parameter</button>
            <div id="result" aria-live="polite">Pending</div>"""
            self.send_response(200)
            self.send_header("Content-Type", "text/html; charset=utf-8")
            self.end_headers()
            self.wfile.write(body)

        def do_POST(self):
            received.append(
                self.rfile.read(int(self.headers["Content-Length"])).decode()
            )
            self.send_response(200)
            self.end_headers()

    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    with tempfile.TemporaryDirectory(prefix="artemis-browser-smoke-") as folder:
        process = await asyncio.create_subprocess_exec(
            *[
                str(chrome),
                "--user-data-dir=" + folder,
                "--no-first-run",
                "--no-default-browser-check",
                "--disable-extensions",
                "--app=http://127.0.0.1:" + str(server.server_port),
                "--window-size=800,600",
                "--window-position=100,100",
            ],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            creationflags=subprocess.CREATE_NO_WINDOW,
        )
        cover = None
        try:
            await asyncio.sleep(3)
            if occlude:
                ready = Path(folder) / "cover.ready"
                cover = await asyncio.create_subprocess_exec(
                    "powershell.exe",
                    "-NoProfile",
                    "-NonInteractive",
                    "-File",
                    str(Path(__file__).with_name("capture-fixture.ps1")),
                    "-Title",
                    "ARTEMIS Browser Occlusion Fixture",
                    "-Cover",
                    "-Width",
                    "1000",
                    "-Height",
                    "800",
                    "-ReadyFile",
                    str(ready),
                    stdout=subprocess.DEVNULL,
                    stderr=subprocess.DEVNULL,
                    creationflags=subprocess.CREATE_NO_WINDOW,
                )
                async with asyncio.timeout(10):
                    while not ready.exists():
                        await asyncio.sleep(0.05)
                await asyncio.sleep(1)
            async with (
                stdio_client(
                    StdioServerParameters(command=str(executable), args=["mcp"])
                ) as streams,
                ClientSession(*streams) as session,
            ):
                await session.initialize()
                apps = await session.call_tool("list_apps", {})
                own = next(
                    b.text
                    for b in apps.content
                    if b.type == "text" and "ARTEMIS Browser Fixture" in b.text
                )
                line = next(
                    line
                    for line in own.splitlines()
                    if "ARTEMIS Browser Fixture" in line
                )
                pid = re.search(r"pid=(\d+)", line).group(1)

                captured = []

                async def observe():
                    response = await session.call_tool("get_app_state", {"app": pid})
                    assert not response.isError, response
                    content = "\n".join(
                        b.text for b in response.content if b.type == "text"
                    )
                    assert any(b.type == "image" for b in response.content), (
                        "Browser capture unavailable"
                    )
                    captured.append(
                        next(b.data for b in response.content if b.type == "image")
                    )
                    return content

                state = await observe()
                print(state)
                entry = re.search(
                    r"^\s*(\d+) (?:entry|text field|edit|textbox|编辑)[^\n]*Test parameter",
                    state,
                    re.MULTILINE | re.IGNORECASE,
                )
                if entry is None:
                    raise AssertionError("No editable browser form control exposed")
                result = await session.call_tool(
                    "set_value",
                    {
                        "app": pid,
                        "element_index": entry.group(1),
                        "value": "ARTEMIS 中文参数",
                    },
                )
                assert not result.isError, result
                state = await observe()
                button = re.search(
                    r"^\s*(\d+) (?:button|按钮) Apply parameter",
                    state,
                    re.MULTILINE | re.IGNORECASE,
                )
                assert button, state
                result = await session.call_tool(
                    "click",
                    {
                        "app": pid,
                        "element_index": button.group(1),
                        "click_method": "auto",
                    },
                )
                assert not result.isError, result
                state = await observe()
                assert received == ["ARTEMIS 中文参数"], json.dumps(received)
                assert "Saved: ARTEMIS 中文参数" in state
                with Image.open(
                    io.BytesIO(base64.b64decode(captured[-1]))
                ) as screenshot:
                    pixels = list(screenshot.convert("RGB").getdata())
                    green = sum(g > 180 and r < 80 and b < 80 for r, g, b in pixels)
                    assert green > len(pixels) / 3, (
                        "Screenshot did not reflect the submitted page change"
                    )
                print(
                    "PASS: isolated browser form changed, submitted once and read back through MCP"
                )
        finally:
            if cover:
                if cover.returncode is None:
                    cover.terminate()
                await asyncio.wait_for(cover.wait(), 10)
            try:
                parent = psutil.Process(process.pid)
                children = parent.children(recursive=True)
                for child in children:
                    try:
                        child.terminate()
                    except psutil.NoSuchProcess:
                        pass
                parent.terminate()
                psutil.wait_procs([parent, *children], timeout=10)
            except psutil.NoSuchProcess:
                pass
            server.shutdown()
            server.server_close()


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("executable", type=Path)
    parser.add_argument("--chrome", type=Path, required=True)
    parser.add_argument("--occlude", action="store_true")
    options = parser.parse_args()
    asyncio.run(main(options.executable.resolve(), options.chrome, options.occlude))