browser-use--browser-harness
Sarath S Menon
fefca43ab5
Add browser_connections() / active_browser_connections() to admin, which probe each daemon socket via a new connection_status meta message. The daemon tracks target_id (set on attach and on set_session) and returns the live page title/URL via Target.getTargetInfo. switch_tab now forwards target_id in set_session so the daemon stays in sync after tab switches. run_doctor shows the count, each daemon name, and the truncated active-page title and URL. Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
139 行
5.0 KiB
Python
139 行
5.0 KiB
Python
from browser_harness import admin
|
|
|
|
|
|
class FakeSocket:
|
|
def __init__(self, response=b'{"target_id":"target-1","session_id":"session-1","page":null}\n'):
|
|
self.response = response
|
|
self.closed = False
|
|
self.sent = b""
|
|
|
|
def sendall(self, data):
|
|
self.sent += data
|
|
|
|
def recv(self, _size):
|
|
out, self.response = self.response, b""
|
|
return out
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
|
|
def test_local_chrome_mode_is_false_when_env_provides_remote_cdp():
|
|
assert not admin._is_local_chrome_mode({"BU_CDP_WS": "ws://example.test/devtools/browser/1"})
|
|
|
|
|
|
def test_local_chrome_mode_is_false_when_process_env_provides_remote_cdp(monkeypatch):
|
|
monkeypatch.setenv("BU_CDP_WS", "ws://example.test/devtools/browser/1")
|
|
|
|
assert not admin._is_local_chrome_mode()
|
|
|
|
|
|
def test_handshake_timeout_needs_chrome_remote_debugging_prompt():
|
|
msg = "CDP WS handshake failed: timed out during opening handshake"
|
|
|
|
assert admin._needs_chrome_remote_debugging_prompt(msg)
|
|
|
|
|
|
def test_handshake_403_needs_chrome_remote_debugging_prompt():
|
|
msg = "CDP WS handshake failed: server rejected WebSocket connection: HTTP 403"
|
|
|
|
assert admin._needs_chrome_remote_debugging_prompt(msg)
|
|
|
|
|
|
def test_stale_websocket_does_not_open_chrome_inspect():
|
|
msg = "no close frame received or sent"
|
|
|
|
assert not admin._needs_chrome_remote_debugging_prompt(msg)
|
|
|
|
|
|
def test_daemon_endpoint_names_discovers_valid_socket_names(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(admin.ipc, "IS_WINDOWS", False)
|
|
monkeypatch.setattr(admin.ipc, "_TMP", tmp_path)
|
|
(tmp_path / "bu-default.sock").touch()
|
|
(tmp_path / "bu-remote_1.sock").touch()
|
|
(tmp_path / "bu-invalid.name.sock").touch()
|
|
(tmp_path / "not-bu-default.sock").touch()
|
|
|
|
assert admin._daemon_endpoint_names() == ["default", "remote_1"]
|
|
|
|
|
|
def test_active_browser_connections_counts_only_healthy_daemons(monkeypatch):
|
|
monkeypatch.setattr(admin, "_daemon_endpoint_names", lambda: ["default", "stale", "remote"])
|
|
|
|
def fake_connect(name, timeout=1.0):
|
|
if name == "stale":
|
|
raise ConnectionRefusedError()
|
|
if name == "remote":
|
|
return FakeSocket(b'{"error":"no close frame received or sent"}\n')
|
|
return FakeSocket()
|
|
|
|
monkeypatch.setattr(admin.ipc, "connect", fake_connect)
|
|
|
|
assert admin.active_browser_connections() == 1
|
|
|
|
|
|
def test_browser_connections_returns_attached_page(monkeypatch):
|
|
monkeypatch.setattr(admin, "_daemon_endpoint_names", lambda: ["default"])
|
|
response = (
|
|
b'{"target_id":"target-1","session_id":"session-1",'
|
|
b'"page":{"targetId":"target-1","title":"Cat - Wikipedia","url":"https://en.wikipedia.org/wiki/Cat"}}\n'
|
|
)
|
|
monkeypatch.setattr(admin.ipc, "connect", lambda name, timeout=1.0: FakeSocket(response))
|
|
|
|
assert admin.browser_connections() == [
|
|
{
|
|
"name": "default",
|
|
"page": {"title": "Cat - Wikipedia", "url": "https://en.wikipedia.org/wiki/Cat"},
|
|
}
|
|
]
|
|
|
|
|
|
def test_run_doctor_prints_active_browser_connections_and_active_pages(monkeypatch, capsys):
|
|
monkeypatch.setattr(admin, "_version", lambda: "0.1.0")
|
|
monkeypatch.setattr(admin, "_install_mode", lambda: "git")
|
|
monkeypatch.setattr(admin, "_chrome_running", lambda: True)
|
|
monkeypatch.setattr(admin, "daemon_alive", lambda: True)
|
|
monkeypatch.setattr(admin, "browser_connections", lambda: [
|
|
{
|
|
"name": "default",
|
|
"page": {"title": "Example", "url": "https://example.test"},
|
|
},
|
|
{
|
|
"name": "cats",
|
|
"page": {"title": "Cat - Wikipedia", "url": "https://en.wikipedia.org/wiki/Cat"},
|
|
},
|
|
])
|
|
monkeypatch.setattr(admin, "_latest_release_tag", lambda: "0.1.0")
|
|
monkeypatch.setattr("shutil.which", lambda _cmd: None)
|
|
monkeypatch.delenv("BROWSER_USE_API_KEY", raising=False)
|
|
|
|
assert admin.run_doctor() == 0
|
|
|
|
out = capsys.readouterr().out
|
|
assert "[ok ] active browser connections — 2" in out
|
|
assert " default — active page: Example — https://example.test" in out
|
|
assert " cats — active page: Cat - Wikipedia — https://en.wikipedia.org/wiki/Cat" in out
|
|
|
|
|
|
def test_doctor_page_output_truncates_long_text(monkeypatch, capsys):
|
|
monkeypatch.setattr(admin, "_version", lambda: "0.1.0")
|
|
monkeypatch.setattr(admin, "_install_mode", lambda: "git")
|
|
monkeypatch.setattr(admin, "_chrome_running", lambda: True)
|
|
monkeypatch.setattr(admin, "daemon_alive", lambda: True)
|
|
monkeypatch.setattr(admin, "DOCTOR_TEXT_LIMIT", 20)
|
|
monkeypatch.setattr(admin, "browser_connections", lambda: [
|
|
{
|
|
"name": "default",
|
|
"page": {"title": "A very long page title", "url": "https://example.test/very/long/path"},
|
|
}
|
|
])
|
|
monkeypatch.setattr(admin, "_latest_release_tag", lambda: "0.1.0")
|
|
monkeypatch.setattr("shutil.which", lambda _cmd: None)
|
|
monkeypatch.delenv("BROWSER_USE_API_KEY", raising=False)
|
|
|
|
assert admin.run_doctor() == 0
|
|
|
|
out = capsys.readouterr().out
|
|
assert "A very long page ..." in out
|
|
assert "https://example.t..." in out
|