mirror of
https://github.com/zhayujie/chatgpt-on-wechat.git
synced 2026-10-26 01:54:36 +08:00
fix(browser): use native actions for snapshot references
Signed-off-by: Rudy Celekli <rudy@gradiahq.com>
This commit is contained in:
2 files changed
+86
-34
No files matched your search
@@ -14,6 +14,7 @@ import uuid
|
||||
import queue
|
||||
import signal
|
||||
import threading
|
||||
from contextlib import contextmanager
|
||||
from typing import Optional, Dict, Any, List, Callable
|
||||
|
||||
from common.log import logger
|
||||
@@ -1165,22 +1166,28 @@ class BrowserService:
|
||||
timeout: int = 5000) -> Dict[str, Any]:
|
||||
return self._submit(self._do_click, ref, selector, timeout)
|
||||
|
||||
@contextmanager
|
||||
def _ref_element(self, ref):
|
||||
handle = self._page.evaluate_handle(
|
||||
"ref => (window.__cowRefMap && window.__cowRefMap[ref]) || null", ref
|
||||
)
|
||||
try:
|
||||
element = handle.as_element()
|
||||
if element is None:
|
||||
raise ValueError(f"ref {ref} not found. Run snapshot first.")
|
||||
yield element
|
||||
finally:
|
||||
handle.dispose()
|
||||
|
||||
def _do_click(self, ref, selector, timeout) -> Dict[str, Any]:
|
||||
page = self._page
|
||||
try:
|
||||
if ref is not None:
|
||||
result = page.evaluate(f"""
|
||||
() => {{
|
||||
const el = window.__cowRefMap && window.__cowRefMap[{ref}];
|
||||
if (!el) return {{ error: "ref {ref} not found. Run snapshot first." }};
|
||||
el.click();
|
||||
return {{ clicked: true, tag: el.tagName.toLowerCase() }};
|
||||
}}
|
||||
""")
|
||||
if result.get("error"):
|
||||
return result
|
||||
with self._ref_element(ref) as element:
|
||||
tag = element.evaluate("el => el.tagName.toLowerCase()")
|
||||
element.click(timeout=timeout)
|
||||
page.wait_for_timeout(500)
|
||||
return result
|
||||
return {"clicked": True, "tag": tag}
|
||||
elif selector:
|
||||
page.click(selector, timeout=timeout)
|
||||
return {"clicked": True, "selector": selector}
|
||||
@@ -1197,18 +1204,8 @@ class BrowserService:
|
||||
page = self._page
|
||||
try:
|
||||
if ref is not None:
|
||||
result = page.evaluate(f"""
|
||||
() => {{
|
||||
const el = window.__cowRefMap && window.__cowRefMap[{ref}];
|
||||
if (!el) return {{ error: "ref {ref} not found. Run snapshot first." }};
|
||||
el.focus();
|
||||
el.value = "";
|
||||
return {{ tag: el.tagName.toLowerCase(), name: el.name || "" }};
|
||||
}}
|
||||
""")
|
||||
if result.get("error"):
|
||||
return result
|
||||
page.keyboard.type(text)
|
||||
with self._ref_element(ref) as element:
|
||||
element.fill(text, timeout=timeout)
|
||||
return {"filled": True, "ref": ref, "text": text}
|
||||
elif selector:
|
||||
page.fill(selector, text, timeout=timeout)
|
||||
@@ -1226,17 +1223,10 @@ class BrowserService:
|
||||
page = self._page
|
||||
try:
|
||||
if ref is not None:
|
||||
result = page.evaluate(f"""
|
||||
() => {{
|
||||
const el = window.__cowRefMap && window.__cowRefMap[{ref}];
|
||||
if (!el || el.tagName.toLowerCase() !== "select")
|
||||
return {{ error: "ref {ref} is not a <select> element" }};
|
||||
el.value = {repr(value)};
|
||||
el.dispatchEvent(new Event("change", {{ bubbles: true }}));
|
||||
return {{ selected: true, value: el.value }};
|
||||
}}
|
||||
""")
|
||||
return result
|
||||
with self._ref_element(ref) as element:
|
||||
element.select_option(value, timeout=timeout)
|
||||
selected = element.evaluate("el => el.value")
|
||||
return {"selected": True, "value": selected}
|
||||
elif selector:
|
||||
page.select_option(selector, value, timeout=timeout)
|
||||
return {"selected": True, "selector": selector, "value": value}
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
"""Opt-in real browser proof that snapshot refs use normal input actions.
|
||||
|
||||
Run with COW_BROWSER_NATIVE_TEST=1 and an installed system Chrome/Edge plus
|
||||
Playwright. This never installs a browser or opens the user's existing profile.
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
|
||||
import pytest
|
||||
|
||||
pytestmark = pytest.mark.skipif(os.environ.get("COW_BROWSER_NATIVE_TEST") != "1", reason="opt-in installed browser test")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("action,kind,disabled", [
|
||||
("click", "button", False), ("click", "button", True),
|
||||
("fill", "input", False), ("fill", "input", True),
|
||||
("fill", "div", False),
|
||||
("select", "select", False), ("select", "select", True),
|
||||
])
|
||||
def test_ref_actions_follow_browser_editability(tmp_path, action, kind, disabled):
|
||||
pytest.importorskip("playwright.sync_api")
|
||||
from agent.tools.browser.browser_env import detect_system_chrome
|
||||
from agent.tools.browser.browser_tool import BrowserTool
|
||||
|
||||
if detect_system_chrome() is None:
|
||||
pytest.skip("no installed system Chrome/Edge")
|
||||
disabled_attr = " disabled" if disabled else ""
|
||||
if kind == "button":
|
||||
control = f'<button onclick="window.events++"{disabled_attr}>Click target</button>'
|
||||
elif kind == "input":
|
||||
control = f'<input value="initial" oninput="window.events++"{disabled_attr}>'
|
||||
elif kind == "div":
|
||||
control = '<div contenteditable="true" oninput="window.events++">initial</div>'
|
||||
else:
|
||||
control = f'<select onchange="window.events++"{disabled_attr}><option value="initial">Initial</option><option value="new">New</option></select>'
|
||||
page = tmp_path / "controls.html"
|
||||
page.write_text(f'<html><body><script>window.events=0</script>{control}</body></html>', encoding="utf-8")
|
||||
previous = BrowserTool._shared_service
|
||||
BrowserTool._shared_service = None
|
||||
tool = BrowserTool({"cwd": str(tmp_path), "user_data_dir": str(tmp_path / "profile"), "headless": True, "idle_timeout": 0, "startup_timeout": 20, "engine": "system-chrome"})
|
||||
try:
|
||||
navigation = tool.execute({"action": "navigate", "url": page.as_uri(), "timeout": 15000})
|
||||
assert navigation.status == "success", navigation.result
|
||||
match = re.search(r"\[(\d+)\] " + kind + r"\b", navigation.result)
|
||||
assert match, navigation.result
|
||||
outcome = tool.execute({"action": action, "ref": int(match.group(1)), "text": "new", "value": "new", "timeout": 500})
|
||||
assert outcome.status == ("error" if disabled else "success"), outcome.result
|
||||
state = tool._get_service().evaluate("() => ({events: window.events, value: document.querySelector('input,select')?.value, text: document.querySelector('[contenteditable]')?.textContent})")["result"]
|
||||
if disabled:
|
||||
assert state["events"] == 0
|
||||
if kind != "button":
|
||||
assert state["value"] == "initial"
|
||||
elif kind == "button":
|
||||
assert state["events"] == 1
|
||||
else:
|
||||
assert state["events"] > 0
|
||||
assert state["text" if kind == "div" else "value"] == "new"
|
||||
finally:
|
||||
if tool._service is not None:
|
||||
tool._service.close()
|
||||
BrowserTool._shared_service = previous
|
||||
Reference in new issue
Block a user