195 lines
7.7 KiB
Python
195 lines
7.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Exercise the real controld adapter in a disposable desktop, not raw Cua calls."""
|
|
import argparse
|
|
import base64
|
|
import importlib.machinery
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import statistics
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
loader = importlib.machinery.SourceFileLoader("smoke", "/usr/local/bin/lazyboy-cua-smoke")
|
|
spec = importlib.util.spec_from_loader(loader.name, loader)
|
|
smoke = importlib.util.module_from_spec(spec)
|
|
loader.exec_module(smoke)
|
|
TIMINGS = {}
|
|
|
|
|
|
def api(path, body=None, display=":1", expect_error=False):
|
|
request = urllib.request.Request(
|
|
"http://127.0.0.1:7070" + path,
|
|
data=None if body is None else json.dumps(body).encode(),
|
|
headers={"Authorization": "Bearer " + os.environ["LAZYBOY_CONTROL_TOKEN"],
|
|
"Content-Type": "application/json", "x-lazyboy-display": display},
|
|
)
|
|
started = time.monotonic()
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=150) as response:
|
|
result = json.load(response)
|
|
except urllib.error.HTTPError as error:
|
|
if expect_error:
|
|
assert error.code == 400, error.code
|
|
return None
|
|
raise AssertionError(f"{path}: HTTP {error.code}: {error.read()[:300]!r}") from error
|
|
finally:
|
|
if not expect_error:
|
|
TIMINGS.setdefault(path + ":" + str((body or {}).get("action", "")), []).append(
|
|
(time.monotonic() - started) * 1000
|
|
)
|
|
if expect_error:
|
|
assert not result.get("ok", True), result
|
|
else:
|
|
assert result.get("ok", True), result
|
|
return result
|
|
|
|
|
|
def wait_health():
|
|
deadline = time.monotonic() + 120
|
|
last = None
|
|
while time.monotonic() < deadline:
|
|
try:
|
|
health = api("/controller/health")
|
|
if health.get("backend") == "cua" and health.get("healthy"):
|
|
return health
|
|
last = health
|
|
except Exception as error:
|
|
last = str(error)
|
|
time.sleep(0.5)
|
|
raise AssertionError(f"controld/cua health was not ready: {last}")
|
|
|
|
|
|
def observe():
|
|
result = api("/observe", {})
|
|
png = base64.b64decode(result["png_base64"])
|
|
assert png.startswith(b"\x89PNG"), "no desktop screenshot"
|
|
return result
|
|
|
|
|
|
def element(result, label):
|
|
found = [el for el in result.get("elements", []) if el["title"] == label]
|
|
assert len(found) == 1, (label, result.get("elements"))
|
|
return found[0]
|
|
|
|
|
|
def act(action, expect_error=False):
|
|
return act_many([action], expect_error=expect_error)
|
|
|
|
|
|
def act_many(actions, expect_error=False):
|
|
return api("/act", {"actions": actions, "observe": True, "settle_ms": 100},
|
|
expect_error=expect_error)
|
|
|
|
|
|
def native_round():
|
|
gtk = smoke.launch_gtk()
|
|
try:
|
|
before = observe()
|
|
# Agent enrichment requests browser state after native observation.
|
|
api("/browser", {"action": "snapshot", "ensure": True})
|
|
target = element(before, "Smoke Click")
|
|
assert target["kind"] == "a11y", target
|
|
click = {"kind": "ref", "verb": "click", "refKind": "a11y", "target": target["selector"]}
|
|
after = act(click)
|
|
smoke.wait_file(smoke.CLICKED, "clicked")
|
|
# A successful mutation invalidates the previous observation's handles.
|
|
act(click, expect_error=True)
|
|
after = observe()
|
|
entry = element(after, "Smoke Entry")
|
|
save = element(after, "Smoke Save")
|
|
act_many([
|
|
{"kind": "ref", "verb": "setvalue", "refKind": "a11y",
|
|
"target": entry["selector"], "text": "中文 hello-cua"},
|
|
{"kind": "ref", "verb": "click", "refKind": "a11y", "target": save["selector"]},
|
|
])
|
|
smoke.wait_file(smoke.TYPED, "中文 hello-cua")
|
|
act({"kind": "focus", "title": "LazyBoy Cua Smoke"})
|
|
page = observe()
|
|
gtk_windows = [el for el in page.get("elements", [])
|
|
if el.get("kind") == "window"
|
|
and "Cua Smoke" in el.get("title", "")
|
|
and "Chromium" not in el.get("title", "")]
|
|
assert gtk_windows, page.get("elements")
|
|
win = gtk_windows[0]
|
|
start_x = win["x"] + 40
|
|
start_y = win["y"] + max(int(win["h"]), 80) - 40
|
|
last_error = None
|
|
for _ in range(3):
|
|
(smoke.ROOT / "cua-smoke-drag.json").unlink(missing_ok=True)
|
|
actions = []
|
|
for kind, offset in (("down", 0), ("move", 160), ("up", 160)):
|
|
if actions:
|
|
actions.append({"kind": "wait", "ms": 40})
|
|
actions.append({"kind": "pointer", "type": kind, "button": "left",
|
|
"x": start_x + offset, "y": start_y})
|
|
api("/act", {"actions": actions, "observe": False, "settle_ms": 150})
|
|
try:
|
|
events = json.loads(smoke.wait_file(
|
|
smoke.ROOT / "cua-smoke-drag.json", "button-press-event", timeout=3))
|
|
break
|
|
except smoke.SmokeError as error:
|
|
last_error = error
|
|
act({"kind": "focus", "title": "LazyBoy Cua Smoke"})
|
|
else:
|
|
raise last_error
|
|
assert "button-press-event" in events and "motion-notify-event" in events and "button-release-event" in events, events
|
|
|
|
finally:
|
|
gtk.terminate()
|
|
gtk.wait(timeout=5)
|
|
|
|
|
|
def browser_round():
|
|
page = api("/browser", {"action": "navigate", "url": smoke.FIXTURE_URL, "ensure": True})
|
|
target = element(page, "Smoke Click")
|
|
page = api("/browser", {"action": "click", "selector": target["selector"]})
|
|
assert "clicked-ok" in page["text"], page
|
|
entry = element(page, "Smoke Entry")
|
|
page = api("/browser", {"action": "type", "selector": entry["selector"], "text": "中文 hello-cua"})
|
|
save = element(page, "Smoke Save")
|
|
page = api("/browser", {"action": "click", "selector": save["selector"]})
|
|
assert "typed:中文 hello-cua" in page["text"], page
|
|
api("/browser", {"action": "click", "selector": "p999999:999999"}, expect_error=True)
|
|
# The browser and human view are the same existing profile/window.
|
|
assert smoke.find_browser_window(smoke.list_windows())
|
|
assert smoke.run(["xdpyinfo", "-display", ":1"]).returncode == 0
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--repeat", type=int, default=10)
|
|
parser.add_argument("--check-persistence", action="store_true")
|
|
args = parser.parse_args()
|
|
assert args.repeat > 0
|
|
smoke.REPORT.mkdir(parents=True, exist_ok=True)
|
|
smoke.wait_ready(120)
|
|
health = wait_health()
|
|
server = smoke.start_fixture_server()
|
|
summary = {"backend": health, "requested": args.repeat, "passed": 0}
|
|
try:
|
|
if args.check_persistence:
|
|
page = api("/browser", {"action": "navigate", "url": smoke.FIXTURE_URL, "ensure": True})
|
|
blob = (page.get("title") or "") + "\n" + (page.get("text") or "")
|
|
assert "session-retained" in blob, page
|
|
summary.update(requested=1, passed=1, persistent_cookie=True)
|
|
print("persistent browser cookie survived lifecycle change", flush=True)
|
|
return
|
|
for i in range(args.repeat):
|
|
native_round()
|
|
browser_round()
|
|
summary["passed"] += 1
|
|
print(f"adapter iteration {i + 1}/{args.repeat} passed", flush=True)
|
|
finally:
|
|
server.terminate()
|
|
server.wait(timeout=5)
|
|
summary["timings_ms"] = {key: {"count": len(values), "median": statistics.median(values),
|
|
"max": max(values)} for key, values in TIMINGS.items()}
|
|
(smoke.REPORT / ("persistence.json" if args.check_persistence else "adapter.json")).write_text(json.dumps(summary, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|