Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 13 additions & 6 deletions src/browser_harness/helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,9 +49,10 @@ def _send(req):
return r


def cdp(method, session_id=None, **params):
"""Raw CDP. cdp('Page.navigate', url='...'), cdp('DOM.getDocument', depth=-1)."""
return _send({"method": method, "params": params, "session_id": session_id}).get("result", {})
def cdp(method, params=None, *, session_id=None, **kwargs):
"""Raw CDP. Kwargs and a params dict both work:
cdp('Page.navigate', url='...'), cdp('Target.closeTarget', {'targetId': tid})."""
return _send({"method": method, "params": {**(params or {}), **kwargs}, "session_id": session_id}).get("result", {})


def drain_events(): return _send({"meta": "drain_events"})["events"]
Expand Down Expand Up @@ -319,9 +320,15 @@ def new_tab(url="about:blank"):
# attach, so the brief about:blank is "complete" by the time the caller
# polls and wait_for_load() returns before navigation actually starts.
tid = cdp("Target.createTarget", url="about:blank")["targetId"]
switch_tab(tid)
if url != "about:blank":
goto_url(url)
try:
switch_tab(tid)
if url != "about:blank":
goto_url(url)
except Exception:
# a failed attach/navigation must not leak the blank tab
try: cdp("Target.closeTarget", targetId=tid)
except Exception: pass
raise
return tid

def close_tab(target=None):
Expand Down
57 changes: 57 additions & 0 deletions tests/unit/test_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -350,3 +350,60 @@ def fake_send(req):
"session filter, the background rWS/lF pair would have updated "
"last_activity and prevented the idle window from elapsing."
)


# --- cdp() call forms ---

def _capture_send():
sent = []
def fake_send(req):
sent.append(req)
return {"result": {}}
return fake_send, sent


def test_cdp_kwargs_form():
fake_send, sent = _capture_send()
with patch("browser_harness.helpers._send", side_effect=fake_send):
helpers.cdp("Page.navigate", url="https://x.test")
assert sent[0] == {"method": "Page.navigate", "params": {"url": "https://x.test"}, "session_id": None}


def test_cdp_params_dict_form():
# the form SKILL.md documents: cdp("Domain.method", params)
fake_send, sent = _capture_send()
with patch("browser_harness.helpers._send", side_effect=fake_send):
helpers.cdp("Target.closeTarget", {"targetId": "T1"})
assert sent[0]["params"] == {"targetId": "T1"}
assert sent[0]["session_id"] is None


def test_cdp_dict_and_kwargs_merge_kwargs_win():
fake_send, sent = _capture_send()
with patch("browser_harness.helpers._send", side_effect=fake_send):
helpers.cdp("Runtime.evaluate", {"expression": "1", "returnByValue": False}, returnByValue=True)
assert sent[0]["params"] == {"expression": "1", "returnByValue": True}


def test_cdp_session_id_stays_keyword():
fake_send, sent = _capture_send()
with patch("browser_harness.helpers._send", side_effect=fake_send):
helpers.cdp("Runtime.evaluate", session_id="S1", expression="1")
assert sent[0]["session_id"] == "S1"
assert sent[0]["params"] == {"expression": "1"}


# --- new_tab() must not leak its about:blank tab ---

def test_new_tab_closes_blank_tab_when_attach_fails():
calls = []
def fake_cdp(method, params=None, **kwargs):
calls.append((method, {**(params or {}), **kwargs}))
if method == "Target.createTarget":
return {"targetId": "T1"}
return {}
with patch("browser_harness.helpers.cdp", side_effect=fake_cdp), \
patch("browser_harness.helpers.switch_tab", side_effect=RuntimeError("attach failed")):
with pytest.raises(RuntimeError, match="attach failed"):
helpers.new_tab("https://x.test")
assert ("Target.closeTarget", {"targetId": "T1"}) in calls