1  from unittest.mock import Mock, patch
  2  
  3  
  4  HOST = "example-host"
  5  PORT = 9223
  6  WS_URL = f"ws://{HOST}:{PORT}/devtools/browser/abc123"
  7  HTTP_URL = f"http://{HOST}:{PORT}"
  8  VERSION_URL = f"{HTTP_URL}/json/version"
  9  
 10  
 11  class TestResolveCdpOverride:
 12      def test_keeps_full_devtools_websocket_url(self):
 13          from tools.browser_tool import _resolve_cdp_override
 14  
 15          assert _resolve_cdp_override(WS_URL) == WS_URL
 16  
 17      def test_resolves_http_discovery_endpoint_to_websocket(self):
 18          from tools.browser_tool import _resolve_cdp_override
 19  
 20          response = Mock()
 21          response.raise_for_status.return_value = None
 22          response.json.return_value = {"webSocketDebuggerUrl": WS_URL}
 23  
 24          with patch("tools.browser_tool.requests.get", return_value=response) as mock_get:
 25              resolved = _resolve_cdp_override(HTTP_URL)
 26  
 27          assert resolved == WS_URL
 28          mock_get.assert_called_once_with(VERSION_URL, timeout=10)
 29  
 30      def test_resolves_bare_ws_hostport_to_discovery_websocket(self):
 31          from tools.browser_tool import _resolve_cdp_override
 32  
 33          response = Mock()
 34          response.raise_for_status.return_value = None
 35          response.json.return_value = {"webSocketDebuggerUrl": WS_URL}
 36  
 37          with patch("tools.browser_tool.requests.get", return_value=response) as mock_get:
 38              resolved = _resolve_cdp_override(f"ws://{HOST}:{PORT}")
 39  
 40          assert resolved == WS_URL
 41          mock_get.assert_called_once_with(VERSION_URL, timeout=10)
 42  
 43      def test_falls_back_to_raw_url_when_discovery_fails(self):
 44          from tools.browser_tool import _resolve_cdp_override
 45  
 46          with patch("tools.browser_tool.requests.get", side_effect=RuntimeError("boom")):
 47              assert _resolve_cdp_override(HTTP_URL) == HTTP_URL
 48  
 49      def test_normalizes_provider_returned_http_cdp_url_when_creating_session(self, monkeypatch):
 50          import tools.browser_tool as browser_tool
 51  
 52          provider = Mock()
 53          provider.create_session.return_value = {
 54              "session_name": "cloud-session",
 55              "bb_session_id": "bu_123",
 56              "cdp_url": "https://cdp.browser-use.example/session",
 57              "features": {"browser_use": True},
 58          }
 59  
 60          response = Mock()
 61          response.raise_for_status.return_value = None
 62          response.json.return_value = {"webSocketDebuggerUrl": WS_URL}
 63  
 64          monkeypatch.setattr(browser_tool, "_active_sessions", {})
 65          monkeypatch.setattr(browser_tool, "_session_last_activity", {})
 66          monkeypatch.setattr(browser_tool, "_start_browser_cleanup_thread", lambda: None)
 67          monkeypatch.setattr(browser_tool, "_update_session_activity", lambda task_id: None)
 68          monkeypatch.setattr(browser_tool, "_get_cdp_override", lambda: "")
 69          monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: provider)
 70  
 71          with patch("tools.browser_tool.requests.get", return_value=response) as mock_get:
 72              session_info = browser_tool._get_session_info("task-browser-use")
 73  
 74          assert session_info["cdp_url"] == WS_URL
 75          provider.create_session.assert_called_once_with("task-browser-use")
 76          mock_get.assert_called_once_with(
 77              "https://cdp.browser-use.example/session/json/version",
 78              timeout=10,
 79          )
 80  
 81  
 82  class TestGetCdpOverride:
 83      def test_prefers_env_var_over_config(self, monkeypatch):
 84          import tools.browser_tool as browser_tool
 85  
 86          monkeypatch.setenv("BROWSER_CDP_URL", HTTP_URL)
 87          monkeypatch.setattr(
 88              browser_tool,
 89              "read_raw_config",
 90              lambda: {"browser": {"cdp_url": "http://config-host:9222"}},
 91              raising=False,
 92          )
 93  
 94          response = Mock()
 95          response.raise_for_status.return_value = None
 96          response.json.return_value = {"webSocketDebuggerUrl": WS_URL}
 97  
 98          with patch("tools.browser_tool.requests.get", return_value=response) as mock_get:
 99              resolved = browser_tool._get_cdp_override()
100  
101          assert resolved == WS_URL
102          mock_get.assert_called_once_with(VERSION_URL, timeout=10)
103  
104      def test_uses_config_browser_cdp_url_when_env_missing(self, monkeypatch):
105          import tools.browser_tool as browser_tool
106  
107          monkeypatch.delenv("BROWSER_CDP_URL", raising=False)
108  
109          response = Mock()
110          response.raise_for_status.return_value = None
111          response.json.return_value = {"webSocketDebuggerUrl": WS_URL}
112  
113          with patch("hermes_cli.config.read_raw_config", return_value={"browser": {"cdp_url": HTTP_URL}}), \
114               patch("tools.browser_tool.requests.get", return_value=response) as mock_get:
115              resolved = browser_tool._get_cdp_override()
116  
117          assert resolved == WS_URL
118          mock_get.assert_called_once_with(VERSION_URL, timeout=10)