fix: 修复已有浏览器 CDP 连接

tests/test_cdp_browser.py

@@ -0,0 +1,95 @@
+# -*- coding: utf-8 -*-
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+import config
+from tools.cdp_browser import CDPBrowserManager
+
+
+@pytest.mark.asyncio
+async def test_existing_browser_connects_directly_to_devtools_browser(monkeypatch):
+    monkeypatch.setattr(config, "CDP_CONNECT_EXISTING", True)
+    monkeypatch.setattr(config, "BROWSER_LAUNCH_TIMEOUT", 60)
+
+    manager = CDPBrowserManager()
+    manager.debug_port = 9222
+    manager._get_browser_websocket_url = AsyncMock(  # type: ignore[method-assign]
+        side_effect=AssertionError("existing browser mode must not call /json/version")
+    )
+
+    browser = MagicMock()
+    browser.is_connected.return_value = True
+    browser.contexts = []
+
+    playwright = MagicMock()
+    playwright.chromium.connect_over_cdp = AsyncMock(return_value=browser)
+
+    await manager._connect_via_cdp(playwright)
+
+    playwright.chromium.connect_over_cdp.assert_awaited_once_with(
+        "ws://localhost:9222/devtools/browser",
+        timeout=60000,
+    )
+
+
+@pytest.mark.asyncio
+async def test_existing_browser_falls_back_to_discovered_websocket_url(monkeypatch):
+    monkeypatch.setattr(config, "CDP_CONNECT_EXISTING", True)
+    monkeypatch.setattr(config, "BROWSER_LAUNCH_TIMEOUT", 60)
+
+    manager = CDPBrowserManager()
+    manager.debug_port = 9222
+    manager._get_browser_websocket_url = AsyncMock(  # type: ignore[method-assign]
+        return_value="ws://localhost:9222/devtools/browser/generated-id"
+    )
+
+    browser = MagicMock()
+    browser.is_connected.return_value = True
+    browser.contexts = []
+
+    playwright = MagicMock()
+    playwright.chromium.connect_over_cdp = AsyncMock(
+        side_effect=[RuntimeError("direct websocket failed"), browser]
+    )
+
+    await manager._connect_via_cdp(playwright)
+
+    manager._get_browser_websocket_url.assert_awaited_once_with(9222)
+    assert playwright.chromium.connect_over_cdp.await_args_list[0].args == (
+        "ws://localhost:9222/devtools/browser",
+    )
+    assert playwright.chromium.connect_over_cdp.await_args_list[0].kwargs == {
+        "timeout": 60000,
+    }
+    assert playwright.chromium.connect_over_cdp.await_args_list[1].args == (
+        "ws://localhost:9222/devtools/browser/generated-id",
+    )
+    assert playwright.chromium.connect_over_cdp.await_args_list[1].kwargs == {
+        "timeout": 60000,
+    }
+
+
+@pytest.mark.asyncio
+async def test_launched_browser_uses_discovered_websocket_url(monkeypatch):
+    monkeypatch.setattr(config, "CDP_CONNECT_EXISTING", False)
+
+    manager = CDPBrowserManager()
+    manager.debug_port = 9223
+    manager._get_browser_websocket_url = AsyncMock(  # type: ignore[method-assign]
+        return_value="ws://localhost:9223/devtools/browser/generated-id"
+    )
+
+    browser = MagicMock()
+    browser.is_connected.return_value = True
+    browser.contexts = []
+
+    playwright = MagicMock()
+    playwright.chromium.connect_over_cdp = AsyncMock(return_value=browser)
+
+    await manager._connect_via_cdp(playwright)
+
+    manager._get_browser_websocket_url.assert_awaited_once_with(9223)
+    playwright.chromium.connect_over_cdp.assert_awaited_once_with(
+        "ws://localhost:9223/devtools/browser/generated-id"
+    )