diff --git a/src/browser_harness/helpers.py b/src/browser_harness/helpers.py index 2014887b..f794536a 100644 --- a/src/browser_harness/helpers.py +++ b/src/browser_harness/helpers.py @@ -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"] @@ -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): diff --git a/tests/unit/test_helpers.py b/tests/unit/test_helpers.py index 4a45ee07..c65b56d5 100644 --- a/tests/unit/test_helpers.py +++ b/tests/unit/test_helpers.py @@ -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