"""Tests for Lightpanda engine support in browser_tool.py.""" import json import os from unittest.mock import MagicMock, patch import pytest from tools import browser_tool_lifecycle as bt_lifecycle from tools import browser_tool_lightpanda_fallback as bt_lightpanda_fallback from tools import browser_tool_session as bt_session from tools import browser_tool_install as bt_install from tools import browser_tool_cloud as bt_cloud from tools import browser_tool_cdp as bt_cdp # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _reset_engine_cache(): """Reset the module-level engine cache so tests start clean.""" import tools.browser_tool as bt bt._cached_browser_engine = None bt._browser_engine_resolved = False @pytest.fixture(autouse=True) def _clean_engine_cache(): """Reset engine cache before and after each test.""" _reset_engine_cache() yield _reset_engine_cache() # --------------------------------------------------------------------------- # _get_browser_engine # --------------------------------------------------------------------------- class TestGetBrowserEngine: """Test engine resolution from config and env vars.""" def test_config_lightpanda(self): """Config browser.engine = 'lightpanda' is respected.""" from tools.browser_tool_cloud import _get_browser_engine cfg = {"browser": {"engine": "lightpanda"}} with patch("hermes_cli.config.read_raw_config", return_value=cfg): assert _get_browser_engine() == "lightpanda" # --------------------------------------------------------------------------- # _should_inject_engine # --------------------------------------------------------------------------- class TestShouldInjectEngine: """Test whether --engine flag is injected based on mode.""" def test_auto_never_injects(self): from tools.browser_tool_cloud import _should_inject_engine assert _should_inject_engine("auto") is False def test_lightpanda_injects_in_local_mode(self): from tools.browser_tool_cloud import _should_inject_engine with patch("tools.browser_tool._is_camofox_mode", return_value=False), \ patch("tools.browser_tool_cdp._get_cdp_override", return_value=""), \ patch("tools.browser_tool_cloud._get_cloud_provider", return_value=None): assert _should_inject_engine("lightpanda") is True def test_chrome_injects_in_local_mode(self): from tools.browser_tool_cloud import _should_inject_engine with patch("tools.browser_tool._is_camofox_mode", return_value=False), \ patch("tools.browser_tool_cdp._get_cdp_override", return_value=""), \ patch("tools.browser_tool_cloud._get_cloud_provider", return_value=None): assert _should_inject_engine("chrome") is True def test_no_inject_in_camofox_mode(self): from tools.browser_tool_cloud import _should_inject_engine with patch("tools.browser_tool._is_camofox_mode", return_value=True): assert _should_inject_engine("lightpanda") is False def test_no_inject_with_cdp_override(self): from tools.browser_tool_cloud import _should_inject_engine with patch("tools.browser_tool._is_camofox_mode", return_value=False), \ patch("tools.browser_tool_cdp._get_cdp_override_raw", return_value="ws://localhost:9222"): assert _should_inject_engine("lightpanda") is False # --------------------------------------------------------------------------- # _needs_lightpanda_fallback # --------------------------------------------------------------------------- class TestNeedsLightpandaFallback: """Test fallback detection for Lightpanda results.""" def test_non_lightpanda_never_falls_back(self): from tools.browser_tool_lightpanda_fallback import _needs_lightpanda_fallback result = {"success": False, "error": "timeout"} assert _needs_lightpanda_fallback("chrome", "open", result) is False assert _needs_lightpanda_fallback("auto", "open", result) is False def test_failed_command_triggers_fallback(self): from tools.browser_tool_lightpanda_fallback import _needs_lightpanda_fallback result = {"success": False, "error": "page.goto: Timeout"} assert _needs_lightpanda_fallback("lightpanda", "open", result) is True def test_empty_snapshot_triggers_fallback(self): from tools.browser_tool_lightpanda_fallback import _needs_lightpanda_fallback result = {"success": True, "data": {"snapshot": ""}} assert _needs_lightpanda_fallback("lightpanda", "snapshot", result) is True def test_unknown_command_does_not_trigger_fallback(self): """Commands not in the whitelist should not trigger fallback.""" from tools.browser_tool_lightpanda_fallback import _needs_lightpanda_fallback result = {"success": False, "error": "nope"} assert _needs_lightpanda_fallback("lightpanda", "some_future_cmd", result) is False # --------------------------------------------------------------------------- # Config integration # --------------------------------------------------------------------------- class TestLightpandaRequirements: """Lightpanda should expose browser tools without local Chromium.""" def test_lightpanda_local_mode_does_not_require_chromium(self): with patch("tools.browser_tool._is_camofox_mode", return_value=False), \ patch("tools.browser_tool_cdp._get_cdp_override", return_value=""), \ patch("tools.browser_tool_install._find_agent_browser", return_value="/usr/bin/agent-browser"), \ patch("tools.browser_tool_cloud._get_cloud_provider", return_value=None), \ patch("tools.browser_tool_cloud._get_browser_engine", return_value="lightpanda"), \ patch("tools.browser_tool_install._chromium_installed", return_value=False): assert bt_install.check_browser_requirements() is True def test_chrome_local_mode_still_requires_chromium(self): with patch("tools.browser_tool._is_camofox_mode", return_value=False), \ patch("tools.browser_tool_cdp._get_cdp_override", return_value=""), \ patch("tools.browser_tool_install._find_agent_browser", return_value="/usr/bin/agent-browser"), \ patch("tools.browser_tool_cloud._get_cloud_provider", return_value=None), \ patch("tools.browser_tool_cloud._get_browser_engine", return_value="auto"), \ patch("tools.browser_tool_install._chromium_installed", return_value=False): assert bt_install.check_browser_requirements() is False # --------------------------------------------------------------------------- # cleanup_all_browsers resets engine cache # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Chrome fallback behavior # --------------------------------------------------------------------------- class TestChromeFallback: """Chrome fallback must hand off from Lightpanda without leaking engine policy.""" def test_chrome_fallback_injects_required_sandbox_args(self, tmp_path): captured_envs = [] mock_proc = MagicMock() mock_proc.wait.return_value = None mock_proc.returncode = 1 def capture_popen(_cmd, **kwargs): captured_envs.append(kwargs["env"]) return mock_proc # Keep the fallback's socket dir under this test's private tmp_path. # Using the real shared tmpdir raced concurrent orphan reapers from # sibling pytest processes (atexit _emergency_cleanup_all_sessions), # which rmtree'd the fresh pidless dir mid-command — the CI flake # this test kept hitting before the reaper grace fix. with patch("tools.browser_tool_session._run_browser_command", return_value={ "success": True, "data": {"url": "https://example.com/"} }), \ patch("tools.browser_tool._socket_safe_tmpdir", return_value=str(tmp_path)), \ patch("tools.browser_tool_install._find_agent_browser", return_value="/usr/bin/agent-browser"), \ patch("tools.browser_tool_install._chromium_installed", return_value=True), \ patch("tools.browser_tool_session._needs_chromium_sandbox_bypass", return_value=True), \ patch("subprocess.Popen", side_effect=capture_popen): result = bt_lightpanda_fallback._run_chrome_fallback_command( "task1", "screenshot", [], timeout=30 ) assert result["success"] is False assert captured_envs assert all( env.get("AGENT_BROWSER_ARGS") == "--no-sandbox,--disable-dev-shm-usage" for env in captured_envs ) # --------------------------------------------------------------------------- # fallback warning annotation # --------------------------------------------------------------------------- class TestLightpandaFallbackWarning: """Verify Chrome fallback results are annotated for users.""" def test_fallback_result_gets_user_visible_warning(self): from tools.browser_tool_lightpanda_fallback import _annotate_lightpanda_fallback result = {"success": True, "data": {"snapshot": "- heading \"Hello\" [ref=e1]"}} annotated = _annotate_lightpanda_fallback( result, "Lightpanda returned an empty/too-short snapshot; retried with Chrome.", ) assert annotated["browser_engine"] == "chrome" assert "Lightpanda fallback" in annotated["fallback_warning"] assert annotated["browser_engine_fallback"] == { "from": "lightpanda", "to": "chrome", "reason": "Lightpanda returned an empty/too-short snapshot; retried with Chrome.", } assert annotated["data"]["fallback_warning"] == annotated["fallback_warning"] assert annotated["data"]["browser_engine"] == "chrome" def test_browser_navigate_surfaces_fallback_warning(self): import json import tools.browser_tool as bt result = bt_lightpanda_fallback._annotate_lightpanda_fallback( {"success": True, "data": {"title": "Fallback OK", "url": "https://example.com/"}}, "synthetic Lightpanda failure; retried with Chrome.", ) with patch("tools.browser_tool_cloud._is_local_backend", return_value=True), \ patch("tools.browser_tool_cloud._get_cloud_provider", return_value=None), \ patch("tools.browser_tool_session._get_session_info", return_value={ "session_name": "test", "_first_nav": False, "features": {"local": True, "proxies": True} }), \ patch("tools.browser_tool_session._run_browser_command", side_effect=[ result, {"success": True, "data": {"snapshot": "- heading \"Fallback OK\" [ref=e1]", "refs": {"e1": {}}}}, ]): response = json.loads(bt.browser_navigate("https://example.com", task_id="warn-test")) assert response["success"] is True assert response["browser_engine"] == "chrome" assert "Lightpanda fallback" in response["fallback_warning"] assert response["browser_engine_fallback"]["from"] == "lightpanda" assert response["browser_engine_fallback"]["to"] == "chrome" bt._last_active_session_key.pop("warn-test", None) def test_browser_vision_lightpanda_response_has_structured_fallback(self, tmp_path): import json import tools.browser_tool as bt chrome_shot = tmp_path / "chrome-structured.png" chrome_shot.write_bytes(b"\x89PNG" + b"0" * 128) class _Msg: content = "Example Domain screenshot" class _Choice: message = _Msg() class _Response: choices = [_Choice()] with patch("tools.browser_tool_cloud._get_browser_engine", return_value="lightpanda"), \ patch("tools.browser_tool_cloud._should_inject_engine", return_value=True), \ patch("tools.browser_tool_lightpanda_fallback._chrome_fallback_screenshot", return_value={ "success": True, "data": {"path": str(chrome_shot)} }), \ patch("hermes_constants.get_hermes_dir", return_value=tmp_path), \ patch("agent.auxiliary_client.call_llm", return_value=_Response()): response = json.loads(bt.browser_vision("what is this?", task_id="vision-structured")) assert response["success"] is True assert response["browser_engine"] == "chrome" assert response["browser_engine_fallback"]["from"] == "lightpanda" assert response["browser_engine_fallback"]["to"] == "chrome" assert response["browser_engine_fallback"]["reason"] # --------------------------------------------------------------------------- # _engine_override parameter # --------------------------------------------------------------------------- class TestEngineOverride: """Verify _engine_override bypasses the cached engine.""" @patch("tools.browser_tool_session._get_session_info") @patch("tools.browser_tool_install._find_agent_browser", return_value="/usr/bin/agent-browser") @patch("tools.browser_tool_cloud._is_local_mode", return_value=True) @patch("tools.browser_tool_install._chromium_installed", return_value=True) @patch("tools.browser_tool_cloud._get_cloud_provider", return_value=None) @patch("tools.browser_tool_cdp._get_cdp_override", return_value="") @patch("tools.browser_tool._is_camofox_mode", return_value=False) def test_override_prevents_engine_injection( self, _camofox, _cdp, _cloud, _chromium, _local, _find, _session ): """When _engine_override='auto', --engine flag is NOT injected.""" import tools.browser_tool as bt # Set the global cache to lightpanda bt._cached_browser_engine = "lightpanda" bt._browser_engine_resolved = True _session.return_value = {"session_name": "test-sess"} # Track the cmd_parts that Popen receives captured_cmds = [] mock_proc = MagicMock() mock_proc.wait.return_value = None mock_proc.returncode = 0 def capture_popen(cmd, **kwargs): captured_cmds.append(cmd) return mock_proc # We need to mock the file operations too with patch("subprocess.Popen", side_effect=capture_popen), \ patch("os.open", return_value=99), \ patch("os.close"), \ patch("os.unlink"), \ patch("os.makedirs"), \ patch("builtins.open", MagicMock(return_value=MagicMock( __enter__=MagicMock(return_value=MagicMock(read=MagicMock(return_value='{"success": true, "data": {}}'))), __exit__=MagicMock(return_value=False), ))), \ patch("tools.interrupt.is_interrupted", return_value=False), \ patch("tools.browser_tool_lifecycle._write_owner_pid"): bt_session._run_browser_command("task1", "snapshot", [], _engine_override="auto") # Should NOT contain "--engine" since override is "auto" assert len(captured_cmds) == 1 assert "--engine" not in captured_cmds[0] @patch("tools.browser_tool_session._get_session_info") @patch("tools.browser_tool_install._find_agent_browser", return_value="/usr/bin/agent-browser") @patch("tools.browser_tool_cloud._is_local_mode", return_value=True) @patch("tools.browser_tool_install._chromium_installed", return_value=True) @patch("tools.browser_tool_cloud._get_cloud_provider", return_value=None) @patch("tools.browser_tool_cdp._get_cdp_override", return_value="") @patch("tools.browser_tool._is_camofox_mode", return_value=False) def test_no_override_uses_cached_engine( self, _camofox, _cdp, _cloud, _chromium, _local, _find, _session ): """Lightpanda gets neither auto-injected nor inherited Chrome arguments.""" import tools.browser_tool as bt bt._cached_browser_engine = "lightpanda" bt._browser_engine_resolved = True _session.return_value = {"session_name": "test-sess"} captured_cmds = [] captured_envs = [] mock_proc = MagicMock() mock_proc.wait.return_value = None mock_proc.returncode = 0 def capture_popen(cmd, **kwargs): captured_cmds.append(cmd) captured_envs.append(kwargs["env"]) return mock_proc # Return a substantive snapshot so the LP fallback does NOT trigger. mock_stdout = '{"success": true, "data": {"snapshot": "- heading \\"Hello\\" [ref=e1]", "refs": {"e1": {}}}}' with patch("subprocess.Popen", side_effect=capture_popen), \ patch("os.open", return_value=99), \ patch("os.close"), \ patch("os.unlink"), \ patch("os.makedirs"), \ patch("builtins.open", MagicMock(return_value=MagicMock( __enter__=MagicMock(return_value=MagicMock(read=MagicMock(return_value=mock_stdout))), __exit__=MagicMock(return_value=False), ))), \ patch("tools.interrupt.is_interrupted", return_value=False), \ patch("tools.browser_tool_session._needs_chromium_sandbox_bypass", return_value=True), \ patch("tools.browser_tool_lifecycle._write_owner_pid"), \ patch.dict(os.environ, {}, clear=False): # AppArmor/root detection would normally auto-inject Chromium args. bt_session._run_browser_command("task1", "snapshot", []) # User-supplied current and legacy Chromium knobs must also be removed. with patch.dict(os.environ, { "AGENT_BROWSER_ARGS": "--no-sandbox", "AGENT_BROWSER_CHROME_FLAGS": "--disable-dev-shm-usage", }): bt_session._run_browser_command("task1", "snapshot", []) assert len(captured_cmds) == 2 for command, environment in zip(captured_cmds, captured_envs): assert "--engine" in command engine_idx = command.index("--engine") assert command[engine_idx + 1] == "lightpanda" assert "AGENT_BROWSER_ARGS" not in environment assert "AGENT_BROWSER_CHROME_FLAGS" not in environment def test_hybrid_local_sidecar_injects_engine_even_with_cloud_provider(self): """A task::local sidecar is local even when global cloud config exists.""" import tools.browser_tool as bt bt._cached_browser_engine = "lightpanda" bt._browser_engine_resolved = True captured_cmds = [] mock_provider = MagicMock() mock_proc = MagicMock() mock_proc.wait.return_value = None mock_proc.returncode = 0 def capture_popen(cmd, **kwargs): captured_cmds.append(cmd) return mock_proc mock_stdout = json.dumps({ "success": True, "data": {"snapshot": '- heading "Hello" [ref=e1]', "refs": {"e1": {}}}, }) with patch("tools.browser_tool_session._get_session_info", return_value={"session_name": "local-sidecar"}), \ patch("tools.browser_tool_install._find_agent_browser", return_value="/usr/bin/agent-browser"), \ patch("tools.browser_tool_cloud._is_local_mode", return_value=False), \ patch("tools.browser_tool_install._chromium_installed", return_value=True), \ patch("tools.browser_tool_cloud._get_cloud_provider", return_value=mock_provider), \ patch("tools.browser_tool_cdp._get_cdp_override", return_value=""), \ patch("tools.browser_tool._is_camofox_mode", return_value=False), \ patch("subprocess.Popen", side_effect=capture_popen), \ patch("os.open", return_value=99), \ patch("os.close"), \ patch("os.unlink"), \ patch("os.makedirs"), \ patch("builtins.open", MagicMock(return_value=MagicMock( __enter__=MagicMock(return_value=MagicMock(read=MagicMock(return_value=mock_stdout))), __exit__=MagicMock(return_value=False), ))), \ patch("tools.interrupt.is_interrupted", return_value=False), \ patch("tools.browser_tool_lifecycle._write_owner_pid"): bt_session._run_browser_command("task::local", "snapshot", []) assert len(captured_cmds) == 1 assert "--engine" in captured_cmds[0] assert captured_cmds[0][captured_cmds[0].index("--engine") + 1] == "lightpanda" # --------------------------------------------------------------------------- # lightpanda_engine_status — is the engine in effect, or shadowed? # --------------------------------------------------------------------------- class TestLightpandaEngineStatus: def _gates(self, monkeypatch, **overrides): import tools.browser_tool as bt gates = dict( _using_lightpanda_engine=lambda: True, _get_cdp_override_raw=lambda: "", _is_camofox_mode=lambda: False, _get_cloud_provider=lambda: None, _is_browser_use_cli_mode=lambda: True, _use_real_profile=lambda: False, ) gates.update(overrides) homes = { "_using_lightpanda_engine": bt_lightpanda_fallback, "_get_cdp_override_raw": bt_cdp, "_get_cloud_provider": bt_cloud, "_use_real_profile": bt_cloud, } for name, fn in gates.items(): monkeypatch.setattr(homes.get(name, bt), name, fn) monkeypatch.setattr( "tools.browser_use_cli.is_legacy_browser_use_cloud_config", lambda cfg: False ) return bt def test_not_lightpanda(self, monkeypatch): self._gates(monkeypatch, _using_lightpanda_engine=lambda: False) assert bt_lightpanda_fallback.lightpanda_engine_status() == (False, "") def test_used_in_browser_use_mode(self, monkeypatch): self._gates(monkeypatch) used, reason = bt_lightpanda_fallback.lightpanda_engine_status() assert used is True assert "lightpanda serve" in reason def test_used_with_builtin_tools(self, monkeypatch): self._gates(monkeypatch, _is_browser_use_cli_mode=lambda: False) used, reason = bt_lightpanda_fallback.lightpanda_engine_status() assert used is True assert "--engine lightpanda" in reason def test_shadowed_by_cdp_override(self, monkeypatch): self._gates(monkeypatch, _get_cdp_override_raw=lambda: "ws://x") used, reason = bt_lightpanda_fallback.lightpanda_engine_status() assert used is False and "CDP override" in reason def test_shadowed_by_camofox(self, monkeypatch): self._gates(monkeypatch, _is_camofox_mode=lambda: True) used, reason = bt_lightpanda_fallback.lightpanda_engine_status() assert used is False and "Camofox" in reason def test_shadowed_by_cloud_provider(self, monkeypatch): provider = MagicMock() provider.display_name = "Browserbase" self._gates(monkeypatch, _get_cloud_provider=lambda: provider) used, reason = bt_lightpanda_fallback.lightpanda_engine_status() assert used is False and "Browserbase" in reason def test_shadowed_by_legacy_browser_use_cloud(self, monkeypatch): self._gates(monkeypatch) monkeypatch.setattr( "tools.browser_use_cli.is_legacy_browser_use_cloud_config", lambda cfg: True ) used, reason = bt_lightpanda_fallback.lightpanda_engine_status() assert used is False and "Browser Use cloud" in reason def test_shadowed_by_real_profile(self, monkeypatch): self._gates(monkeypatch, _use_real_profile=lambda: True) used, reason = bt_lightpanda_fallback.lightpanda_engine_status() assert used is False and "use_real_profile" in reason def test_real_profile_wins_over_cloud_provider(self, monkeypatch): """browser_exec resolves real-profile before the backend, so with both set the real-profile toggle is the actual shadow.""" provider = MagicMock() provider.display_name = "Browserbase" self._gates( monkeypatch, _use_real_profile=lambda: True, _get_cloud_provider=lambda: provider, ) used, reason = bt_lightpanda_fallback.lightpanda_engine_status() assert used is False and "use_real_profile" in reason # --------------------------------------------------------------------------- # Browser Use mode session lifecycle # --------------------------------------------------------------------------- class _FakeServer: def __init__(self, port=4321, alive=True): self.port = port self.cdp_url = f"http://127.0.0.1:{port}" self._alive = alive def is_alive(self): return self._alive class TestLightpandaSessionCreation: def _common(self, monkeypatch, *, bu_mode=True, local_backend=True, launch=None): import tools.browser_tool as bt calls = [] def fake_launch(session_name, *, block_private_networks=False): calls.append((session_name, block_private_networks)) if launch is not None: return launch return _FakeServer(), None monkeypatch.setattr("tools.browser_tool_real_profile._real_profile_cdp", lambda: (None, None)) monkeypatch.setattr(bt, "_is_browser_use_cli_mode", lambda: bu_mode) monkeypatch.setattr("tools.browser_tool_lightpanda_fallback._using_lightpanda_engine", lambda: True) monkeypatch.setattr(bt_cloud, "_is_local_backend", lambda: local_backend) monkeypatch.setattr("tools.browser_lightpanda.launch_lightpanda", fake_launch) return bt, calls def test_spawns_lightpanda_in_browser_use_mode(self, monkeypatch): bt, calls = self._common(monkeypatch) info = bt_session._create_local_session("task-1") assert info["session_name"].startswith("lp_") assert info["cdp_url"] == "http://127.0.0.1:4321" assert info["features"] == {"local": True, "lightpanda": True} assert info["bb_session_id"] is None assert calls == [(info["session_name"], False)] def test_blocks_private_networks_for_containerised_terminal(self, monkeypatch): bt, calls = self._common(monkeypatch, local_backend=False) bt_session._create_local_session("task-1") assert calls[0][1] is True def test_ignores_engine_outside_browser_use_mode(self, monkeypatch): bt, calls = self._common(monkeypatch, bu_mode=False) info = bt_session._create_local_session("task-1") assert info["features"] == {"local": True} assert info["cdp_url"] is None assert calls == [] def test_launch_failure_raises(self, monkeypatch): bt, _ = self._common(monkeypatch, launch=(None, "no lightpanda binary was found")) with pytest.raises(RuntimeError, match="no lightpanda binary"): bt_session._create_local_session("task-1") class TestLightpandaSessionLifecycle: def setup_method(self): import tools.browser_tool as bt self.bt = bt self.orig_sessions = bt._active_sessions.copy() self.orig_activity = bt._session_last_activity.copy() self.orig_cleanup_done = bt._cleanup_done bt._active_sessions.clear() bt._session_last_activity.clear() def teardown_method(self): bt = self.bt bt._active_sessions.clear() bt._active_sessions.update(self.orig_sessions) bt._session_last_activity.clear() bt._session_last_activity.update(self.orig_activity) bt._cleanup_done = self.orig_cleanup_done def _seed(self, key="task-1", name="lp_dead"): info = { "session_name": name, "bb_session_id": None, "cdp_url": "http://127.0.0.1:1", "features": {"local": True, "lightpanda": True}, } self.bt._active_sessions[key] = info self.bt._session_last_activity[key] = 1.0 return info def test_dead_process_is_detected(self, monkeypatch): info = self._seed() monkeypatch.setattr("tools.browser_lightpanda.get_server", lambda name: None) assert bt_session._local_backend_process_dead(info) is True monkeypatch.setattr( "tools.browser_lightpanda.get_server", lambda name: _FakeServer(alive=False) ) assert bt_session._local_backend_process_dead(info) is True monkeypatch.setattr("tools.browser_lightpanda.get_server", lambda name: _FakeServer()) assert bt_session._local_backend_process_dead(info) is False assert bt_session._local_backend_process_dead({"features": {"local": True}}) is False def test_get_session_info_respawns_dead_lightpanda(self, monkeypatch): bt = self.bt stale = self._seed() fresh = { "session_name": "lp_fresh", "bb_session_id": None, "cdp_url": "http://127.0.0.1:2", "features": {"local": True, "lightpanda": True}, } cleaned = [] def fake_cleanup(key): cleaned.append(key) bt._active_sessions.pop(key, None) monkeypatch.setattr("tools.browser_tool_lifecycle._start_browser_cleanup_thread", lambda: None) monkeypatch.setattr( bt, "_browser_session_backend", lambda key: MagicMock(ensure_healthy=lambda: True), ) monkeypatch.setattr("tools.browser_lightpanda.get_server", lambda name: None) monkeypatch.setattr(bt_lifecycle, "_cleanup_single_browser_session", fake_cleanup) monkeypatch.setattr("tools.browser_tool_cdp._get_cdp_override", lambda: "") monkeypatch.setattr(bt_cloud, "_get_cloud_provider", lambda: None) monkeypatch.setattr("tools.browser_tool_session._create_local_session", lambda *a, **k: fresh) supervised = [] monkeypatch.setattr("tools.browser_tool_cdp._ensure_cdp_supervisor", supervised.append) info = bt_session._get_session_info("task-1") assert cleaned == ["task-1"] assert info["session_name"] == "lp_fresh" assert bt._active_sessions["task-1"]["session_name"] == "lp_fresh" assert info["session_name"] != stale["session_name"] # Browser Use mode hides the browser_* tools that read supervisor # state; a Lightpanda session never attaches one. assert supervised == [] def test_cleanup_stops_lightpanda_without_agent_browser_close(self, monkeypatch): bt = self.bt self._seed() stopped = [] monkeypatch.setattr("tools.browser_lightpanda.stop_lightpanda", stopped.append) with patch("tools.browser_tool._maybe_stop_recording"), \ patch("tools.browser_tool_session._run_browser_command") as run, \ patch("tools.browser_tool.os.path.exists", return_value=False): bt_lifecycle.cleanup_browser("task-1") run.assert_not_called() assert stopped == ["lp_dead"] assert "task-1" not in bt._active_sessions assert "task-1" not in bt._session_last_activity def test_emergency_cleanup_stops_all_lightpanda(self, monkeypatch): bt = self.bt bt._cleanup_done = False with patch("tools.browser_lightpanda.stop_all_lightpanda") as stop_all, \ patch("tools.browser_tool_real_profile._terminate_real_profile_chrome"), \ patch("tools.browser_tool_lifecycle.cleanup_all_browsers"), \ patch("tools.browser_tool_lifecycle._reap_orphaned_browser_sessions"): bt_lifecycle._emergency_cleanup_all_sessions() stop_all.assert_called_once() def test_orphan_reaper_sweeps_lightpanda_records(self, tmp_path): with patch("tools.browser_lightpanda.reap_orphaned_lightpanda") as reap, \ patch("tools.browser_tool._socket_safe_tmpdir", return_value=str(tmp_path)): bt_lifecycle._reap_orphaned_browser_sessions() reap.assert_called_once() def test_orphan_reaper_survives_non_dict_record(self, tmp_path, monkeypatch): """A parseable-but-non-object state file is swept like an unreadable one — never an AttributeError that wedges the reaper.""" from tools import browser_lightpanda monkeypatch.setenv("HERMES_HOME", str(tmp_path)) state_dir = browser_lightpanda._state_dir() bad = state_dir / "broken.json" bad.write_text('"not a record"', encoding="utf-8") dead_owner = state_dir / "dead.json" dead_owner.write_text(json.dumps({"owner_pid": 2**22 + 7}), encoding="utf-8") assert browser_lightpanda.reap_orphaned_lightpanda() == 0 assert not bad.exists() and not dead_owner.exists()