Artemis-Plugins/tests/browser_smoke.py

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

import argparse
import asyncio
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


async def main(executable, chrome):
    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)">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",
            ],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            creationflags=subprocess.CREATE_NO_WINDOW,
        )
        try:
            await asyncio.sleep(3)
            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)

                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"
                    )
                    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
                print(
                    "PASS: isolated browser form changed, submitted once and read back through MCP"
                )
        finally:
            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)
    options = parser.parse_args()
    asyncio.run(main(options.executable.resolve(), options.chrome))