import os import tempfile import time from unittest.mock import patch import pytest from PIL import Image from browser_harness import helpers def _run(fake_png, width, height, **kwargs): fake = lambda method, **_: {"data": fake_png(width, height)} with patch("browser_harness.helpers.cdp", side_effect=fake), tempfile.TemporaryDirectory() as d: path = os.path.join(d, "shot.png") helpers.capture_screenshot(path, **kwargs) return Image.open(path).size def test_max_dim_downsizes_oversized_image(fake_png): assert max(_run(fake_png, 4592, 2286, max_dim=1800)) == 1800 def test_max_dim_skips_when_image_already_small(fake_png): assert _run(fake_png, 800, 400, max_dim=1800) == (800, 400) def test_max_dim_default_is_no_resize(fake_png): assert _run(fake_png, 4592, 2286) == (4592, 2286) def _seed_skill(tmp_path): site = tmp_path / "domain-skills" / "example" site.mkdir(parents=True) (site / "scraping.md").write_text("hi") def test_goto_url_omits_domain_skills_by_default(tmp_path, monkeypatch): monkeypatch.delenv("BH_DOMAIN_SKILLS", raising=False) monkeypatch.setattr(helpers, "AGENT_WORKSPACE", tmp_path) _seed_skill(tmp_path) with patch("browser_harness.helpers.cdp", return_value={"frameId": "f"}): result = helpers.goto_url("https://www.example.com/") assert result == {"frameId": "f"} def test_goto_url_includes_domain_skills_when_enabled(tmp_path, monkeypatch): monkeypatch.setenv("BH_DOMAIN_SKILLS", "1") monkeypatch.setattr(helpers, "AGENT_WORKSPACE", tmp_path) _seed_skill(tmp_path) with patch("browser_harness.helpers.cdp", return_value={"frameId": "f"}): result = helpers.goto_url("https://www.example.com/") assert result == {"frameId": "f", "domain_skills": ["scraping.md"]} def test_page_info_raises_clear_error_on_js_exception(): def fake_send(req): return {} def fake_cdp(method, **kwargs): return { "result": { "type": "object", "subtype": "error", "description": "ReferenceError: location is not defined", }, "exceptionDetails": { "text": "Uncaught", "lineNumber": 0, "columnNumber": 16, }, } with patch("browser_harness.helpers._send", side_effect=fake_send), \ patch("browser_harness.helpers.cdp", side_effect=fake_cdp): with pytest.raises(RuntimeError, match="ReferenceError"): helpers.page_info() # --- fill_input --- def test_fill_input_focuses_types_and_fires_events(): cdp_calls = [] js_calls = [] def fake_cdp(method, **kwargs): cdp_calls.append((method, kwargs)) return {} def fake_js(expr, **kwargs): js_calls.append(expr) return True # focus call must return True (element found) with patch("browser_harness.helpers.cdp", side_effect=fake_cdp), \ patch("browser_harness.helpers.js", side_effect=fake_js): helpers.fill_input("#my-input", "hello") assert any("#my-input" in e for e in js_calls) key_downs = [m for m, _ in cdp_calls if m == "Input.dispatchKeyEvent"] assert len(key_downs) > 0 assert any("input" in e and "change" in e for e in js_calls) def test_fill_input_raises_when_element_not_found(): def fake_js(expr, **kwargs): return False # element not found with patch("browser_harness.helpers.js", side_effect=fake_js): with pytest.raises(RuntimeError, match="element not found"): helpers.fill_input("#missing", "hello") def test_fill_input_clear_first_sends_select_all_then_backspace(): import sys key_events = [] def fake_cdp(method, **kwargs): if method == "Input.dispatchKeyEvent": key_events.append(kwargs) return {} def fake_js(expr, **kwargs): return True # element found with patch("browser_harness.helpers.cdp", side_effect=fake_cdp), \ patch("browser_harness.helpers.js", side_effect=fake_js): helpers.fill_input("#inp", "x", clear_first=True) # The "a" must be dispatched with the platform-correct modifier (Meta=4 on # macOS, Ctrl=2 elsewhere). Without the modifier, the field would never get # selected — it would just receive a literal "a". expected_mod = 4 if sys.platform == "darwin" else 2 a_events = [e for e in key_events if e.get("key") == "a"] assert a_events, "expected an 'a' key event for select-all" assert all(e.get("modifiers") == expected_mod for e in a_events), \ f"select-all 'a' must carry modifiers={expected_mod}; got {[e.get('modifiers') for e in a_events]}" # Crucial: no `char` event for the "a" — emitting one makes Chrome treat # Cmd/Ctrl+A as a printable letter instead of a shortcut. assert not any(e.get("type") == "char" and e.get("text") == "a" for e in key_events), \ "select-all must not emit a 'char' event with text='a' (would cancel the shortcut)" # Backspace still fires (via press_key, which uses keyDown). keys_down = [e.get("key") for e in key_events if e.get("type") in ("keyDown", "rawKeyDown")] assert "Backspace" in keys_down def test_fill_input_no_clear_skips_ctrl_a(): key_events = [] def fake_cdp(method, **kwargs): if method == "Input.dispatchKeyEvent": key_events.append(kwargs) return {} def fake_js(expr, **kwargs): return True # element found with patch("browser_harness.helpers.cdp", side_effect=fake_cdp), \ patch("browser_harness.helpers.js", side_effect=fake_js): helpers.fill_input("#inp", "x", clear_first=False) keys_seen = [e.get("key") for e in key_events if e.get("type") == "keyDown"] assert "Backspace" not in keys_seen # --- wait_for_element --- def test_wait_for_element_returns_true_when_found_immediately(): def fake_js(expr, **kwargs): return True with patch("browser_harness.helpers.js", side_effect=fake_js): assert helpers.wait_for_element("#target", timeout=2.0) is True def test_wait_for_element_returns_false_on_timeout(): def fake_js(expr, **kwargs): return False with patch("browser_harness.helpers.js", side_effect=fake_js), \ patch("browser_harness.helpers.time") as mock_time: # simulate time advancing past the deadline immediately start = time.time() mock_time.time.side_effect = [start, start + 5.0] mock_time.sleep = lambda _: None assert helpers.wait_for_element("#missing", timeout=1.0) is False def test_wait_for_element_visible_uses_check_visibility(): js_exprs = [] def fake_js(expr, **kwargs): js_exprs.append(expr) return True with patch("browser_harness.helpers.js", side_effect=fake_js): helpers.wait_for_element("#btn", visible=True) # Prefers checkVisibility (walks ancestor chain) with a computed-style # fallback for older Chrome. assert any("checkVisibility" in e for e in js_exprs) assert any("getComputedStyle" in e for e in js_exprs) # must NOT use offsetParent (fails for position:fixed elements) assert not any("offsetParent" in e for e in js_exprs) def test_wait_for_element_non_visible_uses_simple_check(): js_exprs = [] def fake_js(expr, **kwargs): js_exprs.append(expr) return True with patch("browser_harness.helpers.js", side_effect=fake_js): helpers.wait_for_element("#btn", visible=False) assert any("querySelector" in e and "offsetParent" not in e for e in js_exprs) # --- wait_for_network_idle --- def test_wait_for_network_idle_returns_true_when_no_events(): call_count = 0 def fake_send(req): nonlocal call_count call_count += 1 return {"events": []} with patch("browser_harness.helpers._send", side_effect=fake_send), \ patch("browser_harness.helpers.time") as mock_time: start = 1000.0 # first call: not idle yet; second call: idle window elapsed mock_time.time.side_effect = [start, start, start, start + 0.6, start + 0.6] mock_time.sleep = lambda _: None result = helpers.wait_for_network_idle(timeout=5.0, idle_ms=500) assert result is True def test_wait_for_network_idle_waits_for_inflight_request(): # Verifies inflight tracking: must not return True until loadingFinished, # even though >idle_ms elapses between requestWillBeSent and loadingFinished. # An event-silence-only implementation would return True at iter2 (wrong). events_seq = [ [{"method": "Network.requestWillBeSent", "params": {"requestId": "req1"}}], [], # >500ms elapsed — old impl returns True here; new must NOT [{"method": "Network.loadingFinished", "params": {"requestId": "req1"}}], [], # idle_ms after loadingFinished → return True ] idx = 0 def fake_send(req): nonlocal idx evs = events_seq[min(idx, len(events_seq) - 1)] idx += 1 return {"events": evs} with patch("browser_harness.helpers._send", side_effect=fake_send), \ patch("browser_harness.helpers.time") as mock_time: start = 1000.0 # inflight non-empty → short-circuit skips time.time() in idle check for iter1/iter2 mock_time.time.side_effect = [ start, start, # deadline + last_activity init start + 0.1, # iter1 while-check start + 0.1, # iter1 rWS last_activity update # iter1 idle-check: inflight non-empty → short-circuit start + 0.7, # iter2 while-check (>500ms since rWS but request still in flight) # iter2 idle-check: inflight non-empty → short-circuit start + 0.8, # iter3 while-check start + 0.8, # iter3 lF last_activity update start + 0.8, # iter3 idle-check: 0ms < 500 → not idle start + 1.4, # iter4 while-check start + 1.4, # iter4 idle-check: 600ms >= 500 → True ] mock_time.sleep = lambda _: None result = helpers.wait_for_network_idle(timeout=5.0, idle_ms=500) assert result is True assert idx == 4 # did not short-circuit at iter2 despite silence > idle_ms def test_wait_for_network_idle_returns_false_on_timeout(): # Continuous rWS keeps inflight non-empty → idle check short-circuits every iteration. # time.time() is only called for while-check and rWS last_activity (not idle check). def fake_send(req): return {"events": [{"method": "Network.requestWillBeSent", "params": {"requestId": "r"}}]} with patch("browser_harness.helpers._send", side_effect=fake_send), \ patch("browser_harness.helpers.time") as mock_time: start = 1000.0 mock_time.time.side_effect = [ start, start, # deadline + last_activity init start + 0.1, # iter1 while-check (in deadline) start + 0.1, # iter1 rWS last_activity update # iter1 idle-check: inflight non-empty → short-circuit start + 20.0, # iter2 while-check (past deadline → exit) ] mock_time.sleep = lambda _: None result = helpers.wait_for_network_idle(timeout=10.0, idle_ms=500) assert result is False def test_wait_for_network_idle_filters_events_to_active_session(): """Background tabs (e.g. a polling page the agent switched away from) keep emitting Network events into the daemon's global buffer. The wait must filter by session_id of the currently-attached tab — otherwise it would see the background tab's traffic and either fail to return idle or wait on the wrong tab's requests.""" active = "session-ACTIVE" background = "session-BACKGROUND" # First /drain_events/ payload: rWS + lF on the BACKGROUND session that we # must ignore, plus zero events on the active session. With filtering, the # active session sees no traffic and the idle window can elapse. events_seq = [ [ {"session_id": background, "method": "Network.requestWillBeSent", "params": {"requestId": "bg1"}}, {"session_id": background, "method": "Network.loadingFinished", "params": {"requestId": "bg1"}}, ], [], # second drain — quiet on both sessions; idle window should fire here ] drain_idx = 0 def fake_send(req): nonlocal drain_idx if req.get("meta") == "session": return {"session_id": active} if req.get("meta") == "drain_events": evs = events_seq[min(drain_idx, len(events_seq) - 1)] drain_idx += 1 return {"events": evs} return {} with patch("browser_harness.helpers._send", side_effect=fake_send), \ patch("browser_harness.helpers.time") as mock_time: start = 1000.0 # No inflight on active session → idle check uses time.time(). mock_time.time.side_effect = [start, start, start, start + 0.6, start + 0.6] mock_time.sleep = lambda _: None result = helpers.wait_for_network_idle(timeout=5.0, idle_ms=500) assert result is True, ( "wait_for_network_idle must return True even when the BACKGROUND " "session is busy, as long as the ACTIVE session is idle. Without the " "session filter, the background rWS/lF pair would have updated " "last_activity and prevented the idle window from elapsing." )