1  """Tests for cloud browser provider runtime fallback to local Chromium.
  2  
  3  Covers the fallback logic in _get_session_info() when a cloud provider
  4  is configured but fails at runtime (issue #10883).
  5  """
  6  import logging
  7  from unittest.mock import Mock, patch
  8  
  9  import pytest
 10  
 11  import tools.browser_tool as browser_tool
 12  
 13  
 14  def _reset_session_state(monkeypatch):
 15      """Clear caches so each test starts fresh."""
 16      monkeypatch.setattr(browser_tool, "_active_sessions", {})
 17      monkeypatch.setattr(browser_tool, "_cached_cloud_provider", None)
 18      monkeypatch.setattr(browser_tool, "_cloud_provider_resolved", False)
 19      monkeypatch.setattr(browser_tool, "_start_browser_cleanup_thread", lambda: None)
 20      monkeypatch.setattr(browser_tool, "_update_session_activity", lambda t: None)
 21  
 22  
 23  class TestCloudProviderRuntimeFallback:
 24      """Tests for _get_session_info cloud → local fallback."""
 25  
 26      def test_cloud_failure_falls_back_to_local(self, monkeypatch):
 27          """When cloud provider.create_session raises, fall back to local."""
 28          _reset_session_state(monkeypatch)
 29  
 30          provider = Mock()
 31          provider.create_session.side_effect = RuntimeError("401 Unauthorized")
 32          monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: provider)
 33          monkeypatch.setattr(browser_tool, "_get_cdp_override", lambda: None)
 34  
 35          session = browser_tool._get_session_info("task-1")
 36  
 37          assert session["fallback_from_cloud"] is True
 38          assert "401 Unauthorized" in session["fallback_reason"]
 39          assert session["fallback_provider"] == "Mock"
 40          assert session["features"]["local"] is True
 41          assert session["cdp_url"] is None
 42  
 43      def test_cloud_success_no_fallback(self, monkeypatch):
 44          """When cloud succeeds, no fallback markers are present."""
 45          _reset_session_state(monkeypatch)
 46  
 47          provider = Mock()
 48          provider.create_session.return_value = {
 49              "session_name": "cloud-sess",
 50              "bb_session_id": "bb_123",
 51              "cdp_url": None,
 52              "features": {"browser_use": True},
 53          }
 54          monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: provider)
 55          monkeypatch.setattr(browser_tool, "_get_cdp_override", lambda: None)
 56  
 57          session = browser_tool._get_session_info("task-2")
 58  
 59          assert session["session_name"] == "cloud-sess"
 60          assert "fallback_from_cloud" not in session
 61          assert "fallback_reason" not in session
 62  
 63      def test_cloud_and_local_both_fail(self, monkeypatch):
 64          """When both cloud and local fail, raise RuntimeError with both contexts."""
 65          _reset_session_state(monkeypatch)
 66  
 67          provider = Mock()
 68          provider.create_session.side_effect = RuntimeError("cloud boom")
 69          monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: provider)
 70          monkeypatch.setattr(browser_tool, "_get_cdp_override", lambda: None)
 71          monkeypatch.setattr(
 72              browser_tool, "_create_local_session",
 73              Mock(side_effect=OSError("no chromium")),
 74          )
 75  
 76          with pytest.raises(RuntimeError, match="cloud boom.*local.*no chromium"):
 77              browser_tool._get_session_info("task-3")
 78  
 79      def test_no_provider_uses_local_directly(self, monkeypatch):
 80          """When no cloud provider is configured, local mode is used with no fallback markers."""
 81          _reset_session_state(monkeypatch)
 82  
 83          monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: None)
 84          monkeypatch.setattr(browser_tool, "_get_cdp_override", lambda: None)
 85  
 86          session = browser_tool._get_session_info("task-4")
 87  
 88          assert session["features"]["local"] is True
 89          assert "fallback_from_cloud" not in session
 90  
 91      def test_cdp_override_bypasses_provider(self, monkeypatch):
 92          """CDP override takes priority — cloud provider is never consulted."""
 93          _reset_session_state(monkeypatch)
 94  
 95          provider = Mock()
 96          monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: provider)
 97          monkeypatch.setattr(browser_tool, "_get_cdp_override", lambda: "ws://host:9222/devtools/browser/abc")
 98  
 99          session = browser_tool._get_session_info("task-5")
100  
101          provider.create_session.assert_not_called()
102          assert session["cdp_url"] == "ws://host:9222/devtools/browser/abc"
103  
104      def test_fallback_logs_warning_with_provider_name(self, monkeypatch, caplog):
105          """Fallback emits a warning log with the provider class name and error."""
106          _reset_session_state(monkeypatch)
107  
108          BrowserUseProviderFake = type("BrowserUseProvider", (), {
109              "create_session": Mock(side_effect=ConnectionError("timeout")),
110          })
111          provider = BrowserUseProviderFake()
112          monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: provider)
113          monkeypatch.setattr(browser_tool, "_get_cdp_override", lambda: None)
114  
115          with caplog.at_level(logging.WARNING, logger="tools.browser_tool"):
116              session = browser_tool._get_session_info("task-6")
117  
118          assert session["fallback_from_cloud"] is True
119          assert any("BrowserUseProvider" in r.message and "timeout" in r.message
120                      for r in caplog.records)
121  
122      def test_cloud_failure_does_not_poison_next_task(self, monkeypatch):
123          """A fallback for one task_id doesn't affect a new task_id when cloud recovers."""
124          _reset_session_state(monkeypatch)
125  
126          call_count = 0
127  
128          def create_session_flaky(task_id):
129              nonlocal call_count
130              call_count += 1
131              if call_count == 1:
132                  raise RuntimeError("transient failure")
133              return {
134                  "session_name": "cloud-ok",
135                  "bb_session_id": "bb_999",
136                  "cdp_url": None,
137                  "features": {"browser_use": True},
138              }
139  
140          provider = Mock()
141          provider.create_session.side_effect = create_session_flaky
142          monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: provider)
143          monkeypatch.setattr(browser_tool, "_get_cdp_override", lambda: None)
144  
145          # First call fails → fallback
146          s1 = browser_tool._get_session_info("task-a")
147          assert s1["fallback_from_cloud"] is True
148  
149          # Second call (different task) → cloud succeeds
150          s2 = browser_tool._get_session_info("task-b")
151          assert "fallback_from_cloud" not in s2
152          assert s2["session_name"] == "cloud-ok"
153  
154      def test_cloud_returns_invalid_session_triggers_fallback(self, monkeypatch):
155          """Cloud provider returning None or empty dict triggers fallback."""
156          _reset_session_state(monkeypatch)
157  
158          provider = Mock()
159          provider.create_session.return_value = None
160          monkeypatch.setattr(browser_tool, "_get_cloud_provider", lambda: provider)
161          monkeypatch.setattr(browser_tool, "_get_cdp_override", lambda: None)
162  
163          session = browser_tool._get_session_info("task-7")
164  
165          assert session["fallback_from_cloud"] is True
166          assert "invalid session" in session["fallback_reason"]