提交完善

src/jiangchang_desktop_sdk/e2e_helpers.py

@@ -25,13 +25,19 @@ from __future__ import annotations
 import base64
 import re
 import time
+from dataclasses import dataclass
 from pathlib import Path
-from typing import Optional, Tuple
+from typing import Iterator, Optional, Tuple
 
 from playwright.sync_api import Page
 
 
 __all__ = [
+    "SkillDataDir",
+    "resolve_skill_data_dir",
+    "require_logged_in_skill_data_dir",
+    "get_runtime_env",
+    "assert_skill_env_file",
     "drop_file_into_composer",
     "wait_for_attachment_ready",
     "wait_for_composer_enabled",
@@ -40,11 +46,209 @@ __all__ = [
     "wait_for_agent_complete",
 ]
 
+_NEWLINE = object()
+
+_ANONYMOUS_USER_IDS = frozenset({"", "_anon", "anon", "anonymous"})
+
+_DEFAULT_PLACEHOLDER_VALUES = frozenset({
+    "",
+    "xxx",
+    "sk-xxx",
+    "your-key",
+    "your_api_key",
+    "todo",
+    "changeme",
+    "replace-me",
+    "<your-key>",
+})
+
+@dataclass(frozen=True)
+class SkillDataDir:
+    """匠厂宿主解析出的 skill 用户数据目录上下文。"""
+
+    root: Path
+    path: Path
+    user_id: str
+    skill_slug: str
+
 
 def _normalize_ws(text: str) -> str:
     return re.sub(r"\s+", "", text)
 
 
+def _invoke_ipc(page: Page, channel: str, payload: dict | None = None) -> dict:
+    """通过匠厂 preload 暴露的 IPC 调用主进程 handler。"""
+    result = page.evaluate(
+        """
+        async ({ channel, payload }) => {
+          const electron = window.electron;
+          if (!electron || !electron.ipcRenderer
+              || typeof electron.ipcRenderer.invoke !== 'function') {
+            throw new Error(
+              'window.electron.ipcRenderer.invoke 不可用：请在匠厂桌面客户端页面内运行 E2E，'
+              + '且 preload 已暴露 jiangchang IPC'
+            );
+          }
+          try {
+            if (payload === null || payload === undefined) {
+              return await electron.ipcRenderer.invoke(channel);
+            }
+            return await electron.ipcRenderer.invoke(channel, payload);
+          } catch (err) {
+            const msg = err && err.message ? err.message : String(err);
+            throw new Error(channel + ' IPC 调用失败: ' + msg);
+          }
+        }
+        """,
+        {"channel": channel, "payload": payload},
+    )
+    if not isinstance(result, dict):
+        raise RuntimeError(
+            f"{channel} 返回非对象: {result!r}"
+        )
+    return result
+
+
+def resolve_skill_data_dir(
+    page: Page,
+    skill_slug: str,
+    *,
+    create: bool = True,
+) -> SkillDataDir:
+    """通过宿主 IPC 解析当前登录用户下的 skill 数据目录（不拼路径、不用 _anon 默认值）。"""
+    raw = _invoke_ipc(
+        page,
+        "jiangchang:resolve-skill-data-dir",
+        {"skillSlug": skill_slug},
+    )
+    missing = [k for k in ("root", "path", "userId", "skillSlug") if not raw.get(k)]
+    if missing:
+        raise RuntimeError(
+            "jiangchang:resolve-skill-data-dir 返回结构不完整，"
+            f"缺少字段 {missing}，实际: {raw!r}"
+        )
+
+    data_path = Path(str(raw["path"]))
+    if create:
+        data_path.mkdir(parents=True, exist_ok=True)
+
+    return SkillDataDir(
+        root=Path(str(raw["root"])),
+        path=data_path,
+        user_id=str(raw["userId"]),
+        skill_slug=str(raw["skillSlug"]),
+    )
+
+
+def require_logged_in_skill_data_dir(page: Page, skill_slug: str) -> SkillDataDir:
+    """解析 skill 数据目录，并断言当前宿主已同步真实登录用户（非匿名）。"""
+    ctx = resolve_skill_data_dir(page, skill_slug)
+    uid = (ctx.user_id or "").strip().lower()
+    if uid in _ANONYMOUS_USER_IDS:
+        raise AssertionError(
+            "当前宿主没有同步登录用户 ID（收到 "
+            f"{ctx.user_id!r}），不能运行桌面 E2E。"
+            "请先在匠厂客户端登录，或检查 jiangchangUserId 是否已同步到设置。"
+        )
+    return ctx
+
+
+def get_runtime_env(page: Page) -> dict:
+    """诊断用：读取宿主主进程解析的运行时环境（勿用于拼路径）。"""
+    return _invoke_ipc(page, "jiangchang:get-runtime-env", None)
+
+
+def _strip_env_value(raw: str) -> str:
+    value = raw.strip()
+    if len(value) >= 2 and value[0] == value[-1] and value[0] in ('"', "'"):
+        return value[1:-1]
+    return value
+
+
+def _parse_env_file(env_path: Path) -> dict[str, str]:
+    """解析简单 KEY=value .env（无 shell 展开）。"""
+    parsed: dict[str, str] = {}
+    text = env_path.read_text(encoding="utf-8")
+    for line in text.splitlines():
+        stripped = line.strip()
+        if not stripped or stripped.startswith("#"):
+            continue
+        if "=" not in stripped:
+            continue
+        key, _, raw_value = stripped.partition("=")
+        key = key.strip()
+        if not key:
+            continue
+        parsed[key] = _strip_env_value(raw_value)
+    return parsed
+
+
+def _is_placeholder_value(value: str, extra: set[str] | None) -> bool:
+    normalized = value.strip().lower()
+    placeholders = _DEFAULT_PLACEHOLDER_VALUES
+    if extra:
+        placeholders = placeholders | {p.strip().lower() for p in extra}
+    return normalized in placeholders
+
+
+def assert_skill_env_file(
+    skill_data_dir: SkillDataDir,
+    required_keys: list[str] | tuple[str, ...],
+    *,
+    env_rel_path: str = "config/.env",
+    placeholder_values: set[str] | None = None,
+) -> Path:
+    """Preflight：检查 skill 用户数据目录下 config/.env 是否已配置所需密钥。"""
+    env_path = skill_data_dir.path / env_rel_path
+    if not env_path.is_file():
+        raise AssertionError(
+            f"缺少配置文件: {env_path}\n"
+            "E2E 测试不会自动生成真实密钥，请先一次性配置用户数据目录 config/.env"
+        )
+
+    env_map = _parse_env_file(env_path)
+    invalid: list[str] = []
+    for key in required_keys:
+        if key not in env_map:
+            invalid.append(key)
+            continue
+        value = env_map[key]
+        if not value.strip() or _is_placeholder_value(value, placeholder_values):
+            invalid.append(key)
+
+    if invalid:
+        raise AssertionError(
+            f"config/.env 配置不完整或仍为占位符: {env_path}\n"
+            f"缺失/无效的 key: {', '.join(invalid)}\n"
+            "E2E 测试不会自动生成真实密钥，请先一次性配置用户数据目录 config/.env"
+        )
+    return env_path
+
+
+def _iter_text_and_newlines(prompt: str) -> Iterator[str | object]:
+    """按真实输入顺序产出文本片段与换行标记（\\r\\n / \\n / \\r 均视为单次换行）。"""
+    i = 0
+    n = len(prompt)
+    text_start = 0
+    while i < n:
+        if i + 1 < n and prompt[i] == "\r" and prompt[i + 1] == "\n":
+            if i > text_start:
+                yield prompt[text_start:i]
+            yield _NEWLINE
+            i += 2
+            text_start = i
+        elif prompt[i] in "\n\r":
+            if i > text_start:
+                yield prompt[text_start:i]
+            yield _NEWLINE
+            i += 1
+            text_start = i
+        else:
+            i += 1
+    if text_start < n:
+        yield prompt[text_start:]
+
+
 # ─── 附件上传（绕开 SDK send_file 未实现） ──────────────────────────
 
 def drop_file_into_composer(page: Page, file_path: Path) -> None:
@@ -138,11 +342,16 @@ def wait_for_composer_enabled(page: Page, timeout_s: float = 15.0) -> None:
 
 
 def send_prompt_via_composer(page: Page, prompt: str, *, delay_ms: int = 25) -> None:
-    """拟人输入 + Enter 发送。不做任何等待，返回即调用结束。"""
+    """拟人逐字输入 + Shift+Enter 换行 + Enter 发送（禁止 DOM/剪贴板/伪造事件）。"""
     textarea = page.locator('[data-testid="chat-composer-input"]')
     textarea.click()
-    textarea.fill("")
-    textarea.press_sequentially(prompt, delay=delay_ms)
+    textarea.press("Control+A")
+    textarea.press("Backspace")
+    for segment in _iter_text_and_newlines(prompt):
+        if segment is _NEWLINE:
+            textarea.press("Shift+Enter")
+        else:
+            textarea.press_sequentially(str(segment), delay=delay_ms)
     textarea.press("Enter")