116 lines
4.1 KiB
Python
116 lines
4.1 KiB
Python
"""Validate the benchmark's independent oracle and Cua foreground escalation."""
|
|
|
|
import asyncio
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from artemis.plugins.models import ServerConfig
|
|
from artemis.plugins.session import PluginSession
|
|
|
|
|
|
async def main(executable):
|
|
with tempfile.TemporaryDirectory(
|
|
prefix="artemis-benchmark-validation-"
|
|
) as temporary:
|
|
state_file = Path(temporary) / "state.json"
|
|
proc = await asyncio.create_subprocess_exec(
|
|
"powershell.exe",
|
|
"-NoProfile",
|
|
"-File",
|
|
str(Path(__file__).with_name("benchmark-fixture.ps1")),
|
|
"-StateFile",
|
|
str(state_file),
|
|
creationflags=subprocess.CREATE_NO_WINDOW,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
session = None
|
|
try:
|
|
for _ in range(100):
|
|
if state_file.exists():
|
|
break
|
|
await asyncio.sleep(0.1)
|
|
assert state_file.exists()
|
|
session = await PluginSession(
|
|
ServerConfig(command=str(executable)), executable.parent
|
|
).start()
|
|
listing = await session.call("list_windows", {"pid": proc.pid})
|
|
target = {
|
|
"pid": proc.pid,
|
|
"window_id": listing.structuredContent["windows"][0]["window_id"],
|
|
}
|
|
|
|
async def element(label, tool, **extra):
|
|
state = await session.call("get_window_state", target)
|
|
text = "\n".join(getattr(c, "text", "") for c in state.content)
|
|
match = re.search(r'\[(\d+)\][^\n]*"' + re.escape(label) + '"', text)
|
|
assert match, text
|
|
token = state.structuredContent["snapshot_id"] + ":" + match[1]
|
|
result = await session.call(
|
|
tool, {**target, "element_token": token, **extra}
|
|
)
|
|
assert not result.isError, result
|
|
|
|
await element("Network name", "set_value", value="ARTEMIS-LAB")
|
|
await element("Night mode", "click")
|
|
await element("Apply", "click")
|
|
assert json.loads(state_file.read_text("utf-8-sig"))["applied"]
|
|
state = await session.call("get_window_state", target)
|
|
pixel = await session.call(
|
|
"click_at",
|
|
{
|
|
**target,
|
|
"x": 512,
|
|
"y": 256,
|
|
"capture_id": state.structuredContent["capture_id"],
|
|
},
|
|
)
|
|
assert not pixel.isError, pixel
|
|
scrolled = await session.call(
|
|
"scroll",
|
|
{
|
|
**target,
|
|
"direction": "down",
|
|
"amount": 4,
|
|
"by": "page",
|
|
"x": 400,
|
|
"y": 470,
|
|
},
|
|
)
|
|
if scrolled.isError:
|
|
assert (
|
|
scrolled.structuredContent["artemis"]["execution"] == "not_started"
|
|
), scrolled
|
|
scrolled = await session.call(
|
|
"scroll",
|
|
{
|
|
**target,
|
|
"direction": "down",
|
|
"amount": 4,
|
|
"by": "page",
|
|
"x": 400,
|
|
"y": 470,
|
|
"delivery_mode": "foreground",
|
|
},
|
|
)
|
|
assert not scrolled.isError, scrolled
|
|
await element("Finish", "click")
|
|
result = json.loads(state_file.read_text("utf-8-sig"))
|
|
print(json.dumps(result))
|
|
assert (
|
|
result["applied"] and result["finished"] and result["wrong_clicks"] == 0
|
|
)
|
|
finally:
|
|
if session:
|
|
await session.aclose()
|
|
proc.terminate()
|
|
await asyncio.wait_for(proc.wait(), 10)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main(Path(sys.argv[1]).resolve()))
|