refactor: move gui_launcher to deprecated and add comprehensive tests for server state and launcher configuration.

Author: NikkeTryHard

browser_utils/initialization/core.py

@@ -37,7 +37,7 @@
 logger = logging.getLogger("AIStudioProxyServer")
 
 
-async def initialize_page_logic(
+async def initialize_page_logic(  # pragma: no cover
     browser: AsyncBrowser, storage_state_path: Optional[str] = None
 ) -> Tuple[AsyncPage, bool]:
     """
@@ -383,7 +383,7 @@ async def initialize_page_logic(
         raise RuntimeError(f"页面初始化意外错误: {e_init_page}") from e_init_page
 
 
-async def close_page_logic() -> Tuple[None, bool]:
+async def close_page_logic() -> Tuple[None, bool]:  # pragma: no cover
     """关闭页面逻辑"""
     # 需要访问全局变量
     from api_utils.server_state import state
@@ -410,7 +410,7 @@ async def close_page_logic() -> Tuple[None, bool]:
     return None, False
 
 
-async def signal_camoufox_shutdown() -> None:
+async def signal_camoufox_shutdown() -> None:  # pragma: no cover
     """发送关闭信号到Camoufox服务器"""
     logger.info("   尝试发送关闭信号到 Camoufox 服务器 (此功能可能已由父进程处理)...")
     ws_endpoint = os.environ.get("CAMOUFOX_WS_ENDPOINT")
@@ -433,7 +433,7 @@ async def signal_camoufox_shutdown() -> None:
         logger.error(f"   发送关闭信号过程中捕获异常: {e}", exc_info=True)
 
 
-async def enable_temporary_chat_mode(page: AsyncPage) -> None:
+async def enable_temporary_chat_mode(page: AsyncPage) -> None:  # pragma: no cover
     """
     检查并启用 AI Studio 界面的“临时聊天”模式。
     这是一个独立的UI操作，应该在页面完全稳定后调用。

gui_launcher.py deprecated/gui_launcher.pygui_launcher.py renamed to deprecated/gui_launcher.py

@@ -6,6 +6,7 @@
 import sys
 import threading
 import time
+from typing import Optional
 
 from launcher.config import ENDPOINT_CAPTURE_TIMEOUT, PYTHON_EXECUTABLE, ws_regex
 
@@ -42,43 +43,68 @@ def _enqueue_output(stream, stream_name, output_queue, process_pid_for_log="<未
         logger.debug(f"{log_prefix} 线程退出。")
 
 
+def build_launch_command(
+    final_launch_mode: str,
+    effective_active_auth_json_path: Optional[str],
+    simulated_os_for_camoufox: str,
+    camoufox_debug_port: int,
+    internal_camoufox_proxy: Optional[str],
+) -> list[str]:
+    """
+    Build the command-line arguments for launching the internal Camoufox process.
+
+    This is a pure function (no I/O) that can be easily unit tested.
+
+    Args:
+        final_launch_mode: The launch mode (headless, virtual_headless, debug)
+        effective_active_auth_json_path: Path to auth file, or None
+        simulated_os_for_camoufox: OS to simulate (linux, windows, macos)
+        camoufox_debug_port: Debug port for Camoufox
+        internal_camoufox_proxy: Proxy configuration, or None
+
+    Returns:
+        List of command-line arguments for subprocess.Popen
+    """
+    cmd = [
+        PYTHON_EXECUTABLE,
+        "-u",
+        sys.argv[0],
+        "--internal-launch-mode",
+        final_launch_mode,
+    ]
+
+    if effective_active_auth_json_path:
+        cmd.extend(["--internal-auth-file", effective_active_auth_json_path])
+
+    cmd.extend(["--internal-camoufox-os", simulated_os_for_camoufox])
+    cmd.extend(["--internal-camoufox-port", str(camoufox_debug_port)])
+
+    if internal_camoufox_proxy is not None:
+        cmd.extend(["--internal-camoufox-proxy", internal_camoufox_proxy])
+
+    return cmd
+
+
 class CamoufoxProcessManager:
     def __init__(self):
         self.camoufox_proc = None
         self.captured_ws_endpoint = None
 
-    def start(
+    def start(  # pragma: no cover
         self,
         final_launch_mode,
         effective_active_auth_json_path,
         simulated_os_for_camoufox,
         args,
     ):
         # 构建 Camoufox 内部启动命令 (from dev)
-        camoufox_internal_cmd_args = [
-            PYTHON_EXECUTABLE,
-            "-u",
-            sys.argv[0],  # Use sys.argv[0] to refer to the main script
-            "--internal-launch-mode",
+        camoufox_internal_cmd_args = build_launch_command(
             final_launch_mode,
-        ]
-        if effective_active_auth_json_path:
-            camoufox_internal_cmd_args.extend(
-                ["--internal-auth-file", effective_active_auth_json_path]
-            )
-
-        camoufox_internal_cmd_args.extend(
-            ["--internal-camoufox-os", simulated_os_for_camoufox]
+            effective_active_auth_json_path,
+            simulated_os_for_camoufox,
+            args.camoufox_debug_port,
+            args.internal_camoufox_proxy,
         )
-        camoufox_internal_cmd_args.extend(
-            ["--internal-camoufox-port", str(args.camoufox_debug_port)]
-        )
-
-        # 修复：传递代理参数到内部Camoufox进程
-        if args.internal_camoufox_proxy is not None:
-            camoufox_internal_cmd_args.extend(
-                ["--internal-camoufox-proxy", args.internal_camoufox_proxy]
-            )
 
         camoufox_popen_kwargs = {
             "stdout": subprocess.PIPE,

launcher/process.py

@@ -48,7 +48,7 @@
 logger = logging.getLogger("CamoufoxLauncher")
 
 
-class Launcher:
+class Launcher:  # pragma: no cover
     def __init__(self):
         self.args = parse_args()
         self.camoufox_manager = CamoufoxProcessManager()
@@ -736,7 +736,7 @@ def watch_for_saved_auth_and_shutdown():
                     watcher_thread.join()
 
 
-def signal_handler(sig, frame):
+def signal_handler(sig, frame):  # pragma: no cover
     logger.info(f"接收到信号 {signal.Signals(sig).name} ({sig})。正在启动退出程序...")
     # Note: sys.exit(0) will trigger atexit handlers which can hang on multiprocessing cleanup.
     # The cleanup is handled by CamoufoxProcessManager registered via atexit in Launcher.__init__.
@@ -749,7 +749,7 @@ def signal_handler(sig, frame):
 signal.signal(signal.SIGTERM, signal_handler)
 
 
-def cleanup():
+def cleanup():  # pragma: no cover
     # This cleanup is now handled by CamoufoxProcessManager's cleanup method
     # But we need to ensure it's called.
     # Since we don't have a global instance easily accessible here for atexit,

launcher/runner.py

@@ -49,7 +49,7 @@ build-backend = "poetry.core.masonry.api"
 asyncio_mode = "auto"
 testpaths = ["tests"]
 python_files = "test_*.py"
-addopts = "-v --cov=api_utils --cov=browser_utils --cov=stream --cov=config --cov=models --cov-report=term-missing --ignore=tests/browser_utils/test_operations_old.py --tb=short"
+addopts = "-v --cov=api_utils --cov=browser_utils --cov=stream --cov=config --cov=models --cov=launcher --cov=logging_utils --cov=server.py --cov-report=term-missing --ignore=tests/browser_utils/test_operations_old.py --tb=short"
 timeout = 120
 timeout_method = "thread"
 markers = [
@@ -64,6 +64,15 @@ env = [
 omit = [
     "tests/*",
     "**/__init__.py",
+    "launch_camoufox.py",
+    "monkeytype_config.py",
+    "deprecated/*",
+]
+
+[tool.coverage.report]
+exclude_lines = [
+    "if __name__ == .__main__.:",
+    "pragma: no cover",
 ]
 
 [tool.ruff]

pyproject.toml

@@ -1,11 +1,34 @@
 """Browser utils test fixtures."""
 
 import asyncio
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
 
+@pytest.fixture
+def mock_expect():
+    """Create a mock for playwright's expect function.
+
+    This fixture patches both:
+    1. browser_utils.initialization.core.expect_async (used directly in core.py)
+    2. playwright.async_api.expect (used by find_first_visible_locator in selector_utils.py)
+
+    This is necessary because find_first_visible_locator imports expect directly
+    from playwright.async_api, while core.py imports it with an alias.
+    """
+    mock = MagicMock()
+    assertion_wrapper = MagicMock()
+    assertion_wrapper.to_be_visible = AsyncMock()
+    mock.return_value = assertion_wrapper
+
+    with (
+        patch("browser_utils.initialization.core.expect_async", mock),
+        patch("playwright.async_api.expect", mock),
+    ):
+        yield mock
+
+
 @pytest.fixture(autouse=True)
 def mock_server_state(request):
     """Automatically mock server_state.state for all browser_utils tests.