diff --git a/README.md b/README.md index a949bea..0a19ea3 100644 --- a/README.md +++ b/README.md @@ -54,6 +54,30 @@ finally: client.disconnect() ``` +### 桌面 E2E（`jiangchang_desktop_sdk.e2e_helpers`） + +Skill 的桌面端到端测试应通过宿主 IPC 获取**真实**当前用户数据目录，不要在 pytest 进程里手工设置 `JIANGCHANG_USER_ID` / `CLAW_USER_ID` 来伪造用户（否则会回退到 `_anon`）。 + +推荐 preflight 流程： + +1. `require_logged_in_skill_data_dir(page, skill_slug)` — 经 `jiangchang:resolve-skill-data-dir` 解析路径，并断言已登录（非 `_anon` / `anon` / `anonymous`）。 +2. `assert_skill_env_file(ctx, ["API_KEY", ...])` — 检查 `{skill_data}/config/.env` 中必填项已配置且非占位符；E2E 不会自动生成真实密钥。 +3. `send_prompt_via_composer(page, prompt)` — 普通文本逐字键入；内容中的 `\n` / `\r\n` / `\r` 用 **Shift+Enter** 输入换行；最后用 **Enter** 发送（禁止 DOM 注入、剪贴板、JS 伪造 input/change）。 + +诊断时可调用 `get_runtime_env(page)` 查看主进程解析的环境，**不要**用它拼用户数据路径。 + +```python +from jiangchang_desktop_sdk.e2e_helpers import ( + assert_skill_env_file, + require_logged_in_skill_data_dir, + send_prompt_via_composer, +) + +ctx = require_logged_in_skill_data_dir(page, "demo-skill") +assert_skill_env_file(ctx, ["API_KEY"]) +send_prompt_via_composer(page, "第一行\n第二行") +``` + ## 5. 发版流程（维护者） ### main 分支自动预发布 diff --git a/src/jiangchang_desktop_sdk/e2e_helpers.py b/src/jiangchang_desktop_sdk/e2e_helpers.py index 7e36d45..1c23a21 100644 --- a/src/jiangchang_desktop_sdk/e2e_helpers.py +++ b/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", + "", +}) + +@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") diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..a548115 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +# Package marker for unittest discovery. diff --git a/tests/test_e2e_helpers.py b/tests/test_e2e_helpers.py new file mode 100644 index 0000000..4291b6e --- /dev/null +++ b/tests/test_e2e_helpers.py @@ -0,0 +1,125 @@ +# -*- coding: utf-8 -*- +"""e2e_helpers 轻量单元测试（不依赖匠厂桌面 / IPC）。""" +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from jiangchang_desktop_sdk.e2e_helpers import ( + SkillDataDir, + _iter_text_and_newlines, + _is_placeholder_value, + _parse_env_file, + assert_skill_env_file, +) + + +def _segments(prompt: str) -> list[str | None]: + out: list[str | None] = [] + for seg in _iter_text_and_newlines(prompt): + if seg is not None and not isinstance(seg, str): + out.append(None) + else: + out.append(seg) # type: ignore[arg-type] + return out + + +class TestIterTextAndNewlines(unittest.TestCase): + def test_lf(self) -> None: + self.assertEqual(_segments("a\nb"), ["a", None, "b"]) + + def test_crlf(self) -> None: + self.assertEqual(_segments("a\r\nb"), ["a", None, "b"]) + + def test_cr(self) -> None: + self.assertEqual(_segments("a\rb"), ["a", None, "b"]) + + def test_multiple_newlines(self) -> None: + self.assertEqual(_segments("x\n\ny"), ["x", None, None, "y"]) + + +class TestParseEnvFile(unittest.TestCase): + def _write(self, path: Path, content: str) -> None: + path.write_text(content, encoding="utf-8") + + def test_basic_and_comments(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + env_path = Path(tmp) / ".env" + self._write( + env_path, + "# comment\n\nAPI_KEY=real-secret\n# KEY=ignored\n", + ) + parsed = _parse_env_file(env_path) + self.assertEqual(parsed["API_KEY"], "real-secret") + self.assertNotIn("KEY", parsed) + + def test_quoted_values(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + env_path = Path(tmp) / ".env" + self._write(env_path, 'A="double"\nB=\'single\'\n') + parsed = _parse_env_file(env_path) + self.assertEqual(parsed["A"], "double") + self.assertEqual(parsed["B"], "single") + + def test_placeholder_detection(self) -> None: + self.assertTrue(_is_placeholder_value("TODO", None)) + self.assertTrue(_is_placeholder_value("sk-xxx", None)) + self.assertTrue(_is_placeholder_value(" Changeme ", None)) + self.assertFalse(_is_placeholder_value("sk-live-abc", None)) + + +class TestAssertSkillEnvFile(unittest.TestCase): + def test_missing_file(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + ctx = SkillDataDir( + root=Path(tmp), + path=Path(tmp) / "user" / "demo-skill", + user_id="12428", + skill_slug="demo-skill", + ) + with self.assertRaises(AssertionError) as cm: + assert_skill_env_file(ctx, ["API_KEY"]) + self.assertIn("config/.env", str(cm.exception)) + + def test_missing_and_placeholder_keys(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + skill_dir = Path(tmp) / "12428" / "demo-skill" + env_dir = skill_dir / "config" + env_dir.mkdir(parents=True) + (env_dir / ".env").write_text( + "API_KEY=your-key\nOTHER=real\n", + encoding="utf-8", + ) + ctx = SkillDataDir( + root=Path(tmp), + path=skill_dir, + user_id="12428", + skill_slug="demo-skill", + ) + with self.assertRaises(AssertionError) as cm: + assert_skill_env_file(ctx, ["API_KEY", "MISSING", "OTHER"]) + msg = str(cm.exception) + self.assertIn("API_KEY", msg) + self.assertIn("MISSING", msg) + self.assertNotIn("OTHER", msg) + + def test_pass_returns_path(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + skill_dir = Path(tmp) / "1" / "demo-skill" + env_dir = skill_dir / "config" + env_dir.mkdir(parents=True) + env_file = env_dir / ".env" + env_file.write_text("API_KEY=sk-live-ok\n", encoding="utf-8") + ctx = SkillDataDir( + root=Path(tmp), + path=skill_dir, + user_id="1", + skill_slug="demo-skill", + ) + got = assert_skill_env_file(ctx, ["API_KEY"]) + self.assertEqual(got, env_file) + + +if __name__ == "__main__": + unittest.main()