test_browser_camofox_persistence.py
1 """Persistence tests for the Camofox browser backend. 2 3 Tests that managed persistence uses stable identity while default mode 4 uses random identity. Camofox automatically maps each userId to a 5 dedicated persistent Firefox profile on the server side. 6 """ 7 8 import json 9 from unittest.mock import MagicMock, patch 10 11 import pytest 12 13 from tools.browser_camofox import ( 14 _drop_session, 15 _get_session, 16 _managed_persistence_enabled, 17 camofox_close, 18 camofox_navigate, 19 camofox_soft_cleanup, 20 check_camofox_available, 21 get_vnc_url, 22 ) 23 from tools.browser_camofox_state import get_camofox_identity 24 25 26 def _mock_response(status=200, json_data=None): 27 resp = MagicMock() 28 resp.status_code = status 29 resp.json.return_value = json_data or {} 30 resp.raise_for_status = MagicMock() 31 return resp 32 33 34 def _enable_persistence(): 35 """Return a patch context that enables managed persistence via config.""" 36 config = {"browser": {"camofox": {"managed_persistence": True}}} 37 return patch("tools.browser_camofox.load_config", return_value=config) 38 39 40 @pytest.fixture(autouse=True) 41 def _clear_session_state(): 42 import tools.browser_camofox as mod 43 yield 44 with mod._sessions_lock: 45 mod._sessions.clear() 46 mod._vnc_url = None 47 mod._vnc_url_checked = False 48 49 50 class TestManagedPersistenceToggle: 51 def test_disabled_by_default(self): 52 config = {"browser": {"camofox": {"managed_persistence": False}}} 53 with patch("tools.browser_camofox.load_config", return_value=config): 54 assert _managed_persistence_enabled() is False 55 56 def test_enabled_via_config_yaml(self): 57 config = {"browser": {"camofox": {"managed_persistence": True}}} 58 with patch("tools.browser_camofox.load_config", return_value=config): 59 assert _managed_persistence_enabled() is True 60 61 def test_disabled_when_key_missing(self): 62 config = {"browser": {}} 63 with patch("tools.browser_camofox.load_config", return_value=config): 64 assert _managed_persistence_enabled() is False 65 66 def test_disabled_on_config_load_error(self): 67 with patch("tools.browser_camofox.load_config", side_effect=Exception("fail")): 68 assert _managed_persistence_enabled() is False 69 70 71 class TestEphemeralMode: 72 """Default behavior: random userId, no persistence.""" 73 74 def test_session_gets_random_user_id(self, tmp_path, monkeypatch): 75 monkeypatch.setenv("HERMES_HOME", str(tmp_path)) 76 monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377") 77 78 session = _get_session("task-1") 79 assert session["user_id"].startswith("hermes_") 80 assert session["managed"] is False 81 82 def test_different_tasks_get_different_user_ids(self, tmp_path, monkeypatch): 83 monkeypatch.setenv("HERMES_HOME", str(tmp_path)) 84 monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377") 85 86 s1 = _get_session("task-1") 87 s2 = _get_session("task-2") 88 assert s1["user_id"] != s2["user_id"] 89 90 def test_session_reuse_within_same_task(self, tmp_path, monkeypatch): 91 monkeypatch.setenv("HERMES_HOME", str(tmp_path)) 92 monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377") 93 94 s1 = _get_session("task-1") 95 s2 = _get_session("task-1") 96 assert s1 is s2 97 98 99 class TestManagedPersistenceMode: 100 """With managed_persistence: stable userId derived from Hermes profile.""" 101 102 def test_session_gets_stable_user_id(self, tmp_path, monkeypatch): 103 monkeypatch.setenv("HERMES_HOME", str(tmp_path)) 104 monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377") 105 106 with _enable_persistence(): 107 session = _get_session("task-1") 108 expected = get_camofox_identity("task-1") 109 assert session["user_id"] == expected["user_id"] 110 assert session["session_key"] == expected["session_key"] 111 assert session["managed"] is True 112 113 def test_same_user_id_after_session_drop(self, tmp_path, monkeypatch): 114 monkeypatch.setenv("HERMES_HOME", str(tmp_path)) 115 monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377") 116 117 with _enable_persistence(): 118 s1 = _get_session("task-1") 119 uid1 = s1["user_id"] 120 _drop_session("task-1") 121 s2 = _get_session("task-1") 122 assert s2["user_id"] == uid1 123 124 def test_same_user_id_across_tasks(self, tmp_path, monkeypatch): 125 monkeypatch.setenv("HERMES_HOME", str(tmp_path)) 126 monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377") 127 128 with _enable_persistence(): 129 s1 = _get_session("task-a") 130 s2 = _get_session("task-b") 131 # Same profile = same userId, different session keys 132 assert s1["user_id"] == s2["user_id"] 133 assert s1["session_key"] != s2["session_key"] 134 135 def test_different_profiles_get_different_user_ids(self, tmp_path, monkeypatch): 136 monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377") 137 138 with _enable_persistence(): 139 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "profile-a")) 140 s1 = _get_session("task-1") 141 uid_a = s1["user_id"] 142 _drop_session("task-1") 143 144 monkeypatch.setenv("HERMES_HOME", str(tmp_path / "profile-b")) 145 s2 = _get_session("task-1") 146 assert s2["user_id"] != uid_a 147 148 def test_navigate_uses_stable_identity(self, tmp_path, monkeypatch): 149 monkeypatch.setenv("HERMES_HOME", str(tmp_path)) 150 monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377") 151 152 requests_seen = [] 153 154 def _capture_post(url, json=None, timeout=None): 155 requests_seen.append(json) 156 return _mock_response( 157 json_data={"tabId": "tab-1", "url": "https://example.com"} 158 ) 159 160 with _enable_persistence(), \ 161 patch("tools.browser_camofox.requests.post", side_effect=_capture_post): 162 result = json.loads(camofox_navigate("https://example.com", task_id="task-1")) 163 164 assert result["success"] is True 165 expected = get_camofox_identity("task-1") 166 assert requests_seen[0]["userId"] == expected["user_id"] 167 168 def test_navigate_reuses_identity_after_close(self, tmp_path, monkeypatch): 169 monkeypatch.setenv("HERMES_HOME", str(tmp_path)) 170 monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377") 171 172 requests_seen = [] 173 174 def _capture_post(url, json=None, timeout=None): 175 requests_seen.append(json) 176 return _mock_response( 177 json_data={"tabId": f"tab-{len(requests_seen)}", "url": "https://example.com"} 178 ) 179 180 with ( 181 _enable_persistence(), 182 patch("tools.browser_camofox.requests.post", side_effect=_capture_post), 183 patch("tools.browser_camofox.requests.delete", return_value=_mock_response()), 184 ): 185 first = json.loads(camofox_navigate("https://example.com", task_id="task-1")) 186 camofox_close("task-1") 187 second = json.loads(camofox_navigate("https://example.com", task_id="task-1")) 188 189 assert first["success"] is True 190 assert second["success"] is True 191 tab_requests = [req for req in requests_seen if "userId" in req] 192 assert len(tab_requests) == 2 193 assert tab_requests[0]["userId"] == tab_requests[1]["userId"] 194 195 196 class TestVncUrlDiscovery: 197 """VNC URL is derived from the Camofox health endpoint.""" 198 199 def test_vnc_url_from_health_port(self, monkeypatch): 200 monkeypatch.setenv("CAMOFOX_URL", "http://myhost:9377") 201 health_resp = _mock_response(json_data={"ok": True, "vncPort": 6080}) 202 with patch("tools.browser_camofox.requests.get", return_value=health_resp): 203 assert check_camofox_available() is True 204 assert get_vnc_url() == "http://myhost:6080" 205 206 def test_vnc_url_none_when_headless(self, monkeypatch): 207 monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377") 208 health_resp = _mock_response(json_data={"ok": True}) 209 with patch("tools.browser_camofox.requests.get", return_value=health_resp): 210 check_camofox_available() 211 assert get_vnc_url() is None 212 213 def test_vnc_url_rejects_invalid_port(self, monkeypatch): 214 monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377") 215 health_resp = _mock_response(json_data={"ok": True, "vncPort": "bad"}) 216 with patch("tools.browser_camofox.requests.get", return_value=health_resp): 217 check_camofox_available() 218 assert get_vnc_url() is None 219 220 def test_vnc_url_only_probed_once(self, monkeypatch): 221 monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377") 222 health_resp = _mock_response(json_data={"ok": True, "vncPort": 6080}) 223 with patch("tools.browser_camofox.requests.get", return_value=health_resp) as mock_get: 224 check_camofox_available() 225 check_camofox_available() 226 # Second call still hits /health for availability but doesn't re-parse vncPort 227 assert get_vnc_url() == "http://localhost:6080" 228 229 def test_navigate_includes_vnc_hint(self, tmp_path, monkeypatch): 230 monkeypatch.setenv("HERMES_HOME", str(tmp_path)) 231 monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377") 232 import tools.browser_camofox as mod 233 mod._vnc_url = "http://localhost:6080" 234 mod._vnc_url_checked = True 235 236 with patch("tools.browser_camofox.requests.post", return_value=_mock_response( 237 json_data={"tabId": "t1", "url": "https://example.com"} 238 )): 239 result = json.loads(camofox_navigate("https://example.com", task_id="vnc-test")) 240 241 assert result["vnc_url"] == "http://localhost:6080" 242 assert "vnc_hint" in result 243 244 245 class TestCamofoxSoftCleanup: 246 """camofox_soft_cleanup drops local state only when managed persistence is on.""" 247 248 def test_returns_true_and_drops_session_when_enabled(self, tmp_path, monkeypatch): 249 monkeypatch.setenv("HERMES_HOME", str(tmp_path)) 250 monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377") 251 252 with _enable_persistence(): 253 _get_session("task-1") 254 result = camofox_soft_cleanup("task-1") 255 256 assert result is True 257 # Session should have been dropped from in-memory store 258 import tools.browser_camofox as mod 259 with mod._sessions_lock: 260 assert "task-1" not in mod._sessions 261 262 def test_returns_false_when_disabled(self, tmp_path, monkeypatch): 263 monkeypatch.setenv("HERMES_HOME", str(tmp_path)) 264 monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377") 265 266 _get_session("task-1") 267 config = {"browser": {"camofox": {"managed_persistence": False}}} 268 with patch("tools.browser_camofox.load_config", return_value=config): 269 result = camofox_soft_cleanup("task-1") 270 271 assert result is False 272 # Session should still be present — not dropped 273 import tools.browser_camofox as mod 274 with mod._sessions_lock: 275 assert "task-1" in mod._sessions 276 277 def test_does_not_call_server_delete(self, tmp_path, monkeypatch): 278 """Soft cleanup must never hit the Camofox /sessions DELETE endpoint.""" 279 monkeypatch.setenv("HERMES_HOME", str(tmp_path)) 280 monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377") 281 282 with ( 283 _enable_persistence(), 284 patch("tools.browser_camofox.requests.delete") as mock_delete, 285 ): 286 _get_session("task-1") 287 camofox_soft_cleanup("task-1") 288 289 mock_delete.assert_not_called()