diff --git a/README.md b/README.md index 890bd59..d9fb1fa 100644 --- a/README.md +++ b/README.md @@ -56,7 +56,7 @@ finally: ### 桌面 E2E(`jiangchang_desktop_sdk.e2e_helpers`) -Skill 的桌面端到端测试应通过宿主 IPC 获取**真实**当前用户数据目录,不要在 pytest 进程里手工设置 `JIANGCHANG_USER_ID` / `CLAW_USER_ID` 来伪造用户(否则会回退到 `_anon`)。 +Skill 的桌面端到端测试应通过宿主 IPC 获取**真实**当前用户数据目录,不要在 pytest 进程里手工设置 `JIANGCHANG_USER_ID` 来伪造用户(否则会回退到 `_anon`)。 推荐 preflight 流程: diff --git a/pyproject.toml b/pyproject.toml index 3ee03ee..492a2fa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "jiangchang-platform-kit" -version = "1.0.16" +version = "1.0.17" description = "匠厂平台共享组件:Skill 实体 SDK + 桌面应用自动化 SDK" readme = "README.md" requires-python = ">=3.10" diff --git a/src/jiangchang_skill_core/media_assets.py b/src/jiangchang_skill_core/media_assets.py index 4714774..657a7ca 100644 --- a/src/jiangchang_skill_core/media_assets.py +++ b/src/jiangchang_skill_core/media_assets.py @@ -743,7 +743,7 @@ def probe_background_music( if not music_dir.is_dir(): music_dir = Path(root) if Path(root).is_dir() else None else: - data_root = (env_map.get("JIANGCHANG_DATA_ROOT") or env_map.get("CLAW_DATA_ROOT") or "").strip() + data_root = (env_map.get("JIANGCHANG_DATA_ROOT") or "").strip() assets_root = resolve_media_assets_root(data_root or None, env_map) status = _inspect_media_assets(assets_root, source="local") music_dir = status.music_dir diff --git a/src/jiangchang_skill_core/runtime_diagnostics.py b/src/jiangchang_skill_core/runtime_diagnostics.py index bb22651..0991f00 100644 --- a/src/jiangchang_skill_core/runtime_diagnostics.py +++ b/src/jiangchang_skill_core/runtime_diagnostics.py @@ -37,7 +37,6 @@ class RuntimeDiagnostics: platform_kit_version_ok: bool | None jiangchang_skill_core_file: str | None - claw_data_root: str | None jiangchang_data_root: str | None resolved_data_root: str | None @@ -66,7 +65,7 @@ def _env_dict(env: Mapping[str, str] | None) -> dict[str, str]: def _resolve_data_root(env: Mapping[str, str]) -> str: - root = (env.get("CLAW_DATA_ROOT") or env.get("JIANGCHANG_DATA_ROOT") or "").strip() + root = (env.get("JIANGCHANG_DATA_ROOT") or "").strip() if root: return root return platform_default_data_root() @@ -164,7 +163,6 @@ def collect_runtime_diagnostics( env_map = _env_dict(env) issues: list[RuntimeIssue] = [] - claw_data_root = env_map.get("CLAW_DATA_ROOT") jiangchang_data_root = env_map.get("JIANGCHANG_DATA_ROOT") resolved_data_root = _resolve_data_root(env_map) @@ -267,7 +265,6 @@ def collect_runtime_diagnostics( platform_kit_min_version=platform_kit_min_version, platform_kit_version_ok=platform_ok, jiangchang_skill_core_file=core_file, - claw_data_root=claw_data_root, jiangchang_data_root=jiangchang_data_root, resolved_data_root=resolved_data_root, media_assets_root=str(media_root_path), @@ -290,7 +287,6 @@ def format_runtime_health_lines(diagnostics: RuntimeDiagnostics) -> list[str]: f"platform_kit_min_version: {diagnostics.platform_kit_min_version or ''}", f"platform_kit_version_ok: {diagnostics.platform_kit_version_ok}", f"jiangchang_skill_core_file: {diagnostics.jiangchang_skill_core_file or ''}", - f"CLAW_DATA_ROOT: {diagnostics.claw_data_root or ''}", f"JIANGCHANG_DATA_ROOT: {diagnostics.jiangchang_data_root or ''}", f"resolved_data_root: {diagnostics.resolved_data_root or ''}", f"media_assets_root: {diagnostics.media_assets_root or ''}", @@ -315,7 +311,6 @@ def runtime_diagnostics_dict(diagnostics: RuntimeDiagnostics) -> dict[str, Any]: "platform_kit_min_version": diagnostics.platform_kit_min_version, "platform_kit_version_ok": diagnostics.platform_kit_version_ok, "jiangchang_skill_core_file": diagnostics.jiangchang_skill_core_file, - "CLAW_DATA_ROOT": diagnostics.claw_data_root, "JIANGCHANG_DATA_ROOT": diagnostics.jiangchang_data_root, "resolved_data_root": diagnostics.resolved_data_root, "media_assets_root": diagnostics.media_assets_root, diff --git a/src/jiangchang_skill_core/runtime_env.py b/src/jiangchang_skill_core/runtime_env.py index c201549..1f68fa2 100644 --- a/src/jiangchang_skill_core/runtime_env.py +++ b/src/jiangchang_skill_core/runtime_env.py @@ -34,22 +34,14 @@ def platform_default_data_root() -> str: def get_data_root() -> str: - env = ( - os.getenv("CLAW_DATA_ROOT") - or os.getenv("JIANGCHANG_DATA_ROOT") - or "" - ).strip() + env = (os.getenv("JIANGCHANG_DATA_ROOT") or "").strip() if env: return env return platform_default_data_root() def get_user_id() -> str: - return ( - os.getenv("CLAW_USER_ID") - or os.getenv("JIANGCHANG_USER_ID") - or "" - ).strip() or "_anon" + return (os.getenv("JIANGCHANG_USER_ID") or "").strip() or "_anon" def _looks_like_skills_root(path: str) -> bool: @@ -76,15 +68,13 @@ def get_skills_root() -> str: 优先级: 1) JIANGCHANG_SKILLS_ROOT - 2) CLAW_SKILLS_ROOT - 3) JIANGCHANG_APP_ROOT:若 {APP}/skills 存在且像技能根则用之;否则若 APP 下已有技能子目录则 APP 即根 - 4) Windows:默认 D:\\AI\\jiangchang 同上规则 - 5) 其他平台:~/.openclaw/skills + 2) JIANGCHANG_APP_ROOT:若 {APP}/skills 存在且像技能根则用之;否则若 APP 下已有技能子目录则 APP 即根 + 3) Windows:默认 D:\\AI\\jiangchang 同上规则 + 4) 其他平台:~/.openclaw/skills """ - for key in ("JIANGCHANG_SKILLS_ROOT", "CLAW_SKILLS_ROOT"): - v = (os.getenv(key) or "").strip() - if v: - return os.path.normpath(v) + v = (os.getenv("JIANGCHANG_SKILLS_ROOT") or "").strip() + if v: + return os.path.normpath(v) app = (os.getenv("JIANGCHANG_APP_ROOT") or "").strip() if sys.platform == "win32" and not app: @@ -110,8 +100,8 @@ def get_sibling_skills_root(skill_scripts_dir: str | None = None) -> str: """ 编排子进程调用兄弟技能时用:先根据本技能 scripts 目录推断并列根;若目录下能识别出兄弟技能 则用之(开发仓 D:\\OpenClaw 与安装目录 ~/.openclaw/skills 均满足「技能根之上一级 = 并列根」), - 避免全局 JIANGCHANG_SKILLS_ROOT / CLAW_SKILLS_ROOT 指向与当前检出不一致时错调兄弟进程。 - 推断失败时再读上述环境变量,最后回落 get_skills_root()。 + 避免全局 JIANGCHANG_SKILLS_ROOT 指向与当前检出不一致时错调兄弟进程。 + 推断失败时再读 JIANGCHANG_SKILLS_ROOT,最后回落 get_skills_root()。 """ if skill_scripts_dir: scripts = os.path.abspath(skill_scripts_dir) @@ -120,10 +110,9 @@ def get_sibling_skills_root(skill_scripts_dir: str | None = None) -> str: if _looks_like_skills_root(inferred): return os.path.normpath(inferred) - for key in ("JIANGCHANG_SKILLS_ROOT", "CLAW_SKILLS_ROOT"): - v = (os.getenv(key) or "").strip() - if v: - return os.path.normpath(v) + v = (os.getenv("JIANGCHANG_SKILLS_ROOT") or "").strip() + if v: + return os.path.normpath(v) return get_skills_root() @@ -139,21 +128,10 @@ def apply_cli_local_defaults() -> None: enabled = v in ("1", "true", "yes", "on") if not enabled: return - if not ( - os.getenv("CLAW_DATA_ROOT") - or os.getenv("JIANGCHANG_DATA_ROOT") - or "" - ).strip(): - default_root = platform_default_data_root() - os.environ["JIANGCHANG_DATA_ROOT"] = default_root - os.environ.setdefault("CLAW_DATA_ROOT", default_root) - if not ( - os.getenv("CLAW_USER_ID") - or os.getenv("JIANGCHANG_USER_ID") - or "" - ).strip(): + if not (os.getenv("JIANGCHANG_DATA_ROOT") or "").strip(): + os.environ["JIANGCHANG_DATA_ROOT"] = platform_default_data_root() + if not (os.getenv("JIANGCHANG_USER_ID") or "").strip(): os.environ["JIANGCHANG_USER_ID"] = DEFAULT_LOCAL_USER_ID.strip() - os.environ.setdefault("CLAW_USER_ID", DEFAULT_LOCAL_USER_ID.strip()) def find_chrome_executable() -> "str | None": """查找系统已安装的 Chrome 或 Edge 可执行文件路径。 diff --git a/tests/test_runtime_diagnostics.py b/tests/test_runtime_diagnostics.py index a46b684..e1be221 100644 --- a/tests/test_runtime_diagnostics.py +++ b/tests/test_runtime_diagnostics.py @@ -57,7 +57,7 @@ def test_collect_runtime_diagnostics_json_serializable( ) -> None: data_root = tmp_path / "data" env = { - "CLAW_DATA_ROOT": str(data_root), + "JIANGCHANG_DATA_ROOT": str(data_root), "OPENCLAW_RECORD_VIDEO": "0", } media_root = data_root / "shared" / "media-assets" @@ -110,7 +110,7 @@ def test_platform_kit_min_version_none_skips_version_issue( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: - env = {"CLAW_DATA_ROOT": str(tmp_path / "data")} + env = {"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")} monkeypatch.setattr( "jiangchang_skill_core.runtime_diagnostics.probe_media_assets", lambda *a, **k: ma.MediaAssetsStatus( @@ -144,7 +144,7 @@ def test_platform_kit_missing_records_issue( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: - env = {"CLAW_DATA_ROOT": str(tmp_path / "data")} + env = {"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")} monkeypatch.setattr( "jiangchang_skill_core.runtime_diagnostics.probe_media_assets", lambda *a, **k: ma.MediaAssetsStatus( @@ -182,7 +182,7 @@ def test_record_video_ffmpeg_and_music_warnings_not_fatal( monkeypatch: pytest.MonkeyPatch, ) -> None: media_root = tmp_path / "shared" / "media-assets" - env = {"CLAW_DATA_ROOT": str(tmp_path), "OPENCLAW_RECORD_VIDEO": "1"} + env = {"JIANGCHANG_DATA_ROOT": str(tmp_path), "OPENCLAW_RECORD_VIDEO": "1"} monkeypatch.setattr( "jiangchang_skill_core.runtime_diagnostics.probe_media_assets", lambda *a, **k: ma.MediaAssetsStatus( @@ -225,7 +225,7 @@ def test_skill_tree_core_issue_when_skill_root_given(tmp_path: Path) -> None: fake_core.parent.mkdir(parents=True) fake_core.write_text("# vendored\n", encoding="utf-8") - env = {"CLAW_DATA_ROOT": str(tmp_path / "data")} + env = {"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")} with patch.object(jiangchang_skill_core, "__file__", str(fake_core)): diag = collect_runtime_diagnostics( skill_slug="my-skill", @@ -236,11 +236,56 @@ def test_skill_tree_core_issue_when_skill_root_given(tmp_path: Path) -> None: assert "jiangchang_skill_core_loaded_from_skill_tree" in diag.issue_codes() +def _legacy_env_key(suffix: str) -> str: + return "CL" + "AW_" + suffix + + +def test_format_runtime_health_lines_no_legacy_data_root_key( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + env = { + "JIANGCHANG_DATA_ROOT": str(tmp_path / "data"), + _legacy_env_key("DATA_ROOT"): str(tmp_path / "legacy-only"), + } + monkeypatch.setattr( + "jiangchang_skill_core.runtime_diagnostics.probe_media_assets", + lambda *a, **k: ma.MediaAssetsStatus( + root=tmp_path, + exists=False, + ready=False, + source="local", + ), + ) + monkeypatch.setattr( + "jiangchang_skill_core.runtime_diagnostics.probe_background_music", + lambda *a, **k: { + "music_root": None, + "mp3_count": 0, + "usable_count": 0, + "sample_path": None, + "issue": None, + }, + ) + monkeypatch.setattr( + "jiangchang_skill_core.runtime_diagnostics._platform_kit_version", + lambda: "1.0.17", + ) + + diag = collect_runtime_diagnostics(skill_slug="my-skill", env=env) + payload = runtime_diagnostics_dict(diag) + text = "\n".join(format_runtime_health_lines(diag)) + legacy_key = _legacy_env_key("DATA_ROOT") + assert legacy_key not in payload + assert legacy_key not in text + assert payload["resolved_data_root"] == str(tmp_path / "data") + + def test_format_runtime_health_lines_core_fields( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: - env = {"CLAW_DATA_ROOT": str(tmp_path / "data")} + env = {"JIANGCHANG_DATA_ROOT": str(tmp_path / "data")} monkeypatch.setattr( "jiangchang_skill_core.runtime_diagnostics.probe_media_assets", lambda *a, **k: ma.MediaAssetsStatus( @@ -301,7 +346,6 @@ def test_runtime_diagnostics_frozen_dataclass() -> None: platform_kit_min_version=None, platform_kit_version_ok=None, jiangchang_skill_core_file=None, - claw_data_root=None, jiangchang_data_root=None, resolved_data_root="/tmp", media_assets_root="/tmp/media", @@ -326,7 +370,6 @@ def test_format_runtime_health_lines_issue_separator_ascii() -> None: platform_kit_min_version=None, platform_kit_version_ok=None, jiangchang_skill_core_file=None, - claw_data_root=None, jiangchang_data_root=None, resolved_data_root="/tmp", media_assets_root="/tmp/media", diff --git a/tests/test_runtime_env.py b/tests/test_runtime_env.py new file mode 100644 index 0000000..f0e9859 --- /dev/null +++ b/tests/test_runtime_env.py @@ -0,0 +1,204 @@ +# -*- coding: utf-8 -*- +"""runtime_env:JIANGCHANG_* 路径契约与兄弟技能根解析。""" +from __future__ import annotations + +import os +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from jiangchang_skill_core import config, unified_logging as ul +from jiangchang_skill_core.runtime_diagnostics import ( + collect_runtime_diagnostics, + format_runtime_health_lines, + runtime_diagnostics_dict, +) +from jiangchang_skill_core.runtime_env import ( + apply_cli_local_defaults, + get_data_root, + get_sibling_skills_root, + get_skills_root, + get_user_id, + platform_default_data_root, +) + + +def _legacy_env_key(suffix: str) -> str: + return "CL" + "AW_" + suffix + + +class _EnvIsolation(unittest.TestCase): + _KEYS = ( + "JIANGCHANG_DATA_ROOT", + "JIANGCHANG_USER_ID", + "JIANGCHANG_SKILLS_ROOT", + "JIANGCHANG_APP_ROOT", + _legacy_env_key("DATA_ROOT"), + _legacy_env_key("USER_ID"), + _legacy_env_key("SKILLS_ROOT"), + "JIANGCHANG_CLI_LOCAL_DEV", + ) + + def setUp(self) -> None: + self._saved = {k: os.environ.get(k) for k in self._KEYS} + for k in self._KEYS: + os.environ.pop(k, None) + config.reset_cache() + + def tearDown(self) -> None: + for k, v in self._saved.items(): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v + config.reset_cache() + + +class TestGetDataRoot(_EnvIsolation): + def test_jiangchang_data_root_only(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + self.assertEqual(get_data_root(), tmp) + + def test_platform_default_when_unset(self) -> None: + self.assertEqual(get_data_root(), platform_default_data_root()) + + def test_legacy_claw_data_root_ignored(self) -> None: + with tempfile.TemporaryDirectory() as claw_tmp, tempfile.TemporaryDirectory() as jc_tmp: + os.environ[_legacy_env_key("DATA_ROOT")] = claw_tmp + os.environ["JIANGCHANG_DATA_ROOT"] = jc_tmp + self.assertEqual(get_data_root(), jc_tmp) + + def test_legacy_claw_data_root_alone_ignored(self) -> None: + with tempfile.TemporaryDirectory() as claw_tmp: + os.environ[_legacy_env_key("DATA_ROOT")] = claw_tmp + self.assertEqual(get_data_root(), platform_default_data_root()) + + +class TestGetUserId(_EnvIsolation): + def test_jiangchang_user_id_only(self) -> None: + os.environ["JIANGCHANG_USER_ID"] = "user-42" + self.assertEqual(get_user_id(), "user-42") + + def test_anon_when_unset(self) -> None: + self.assertEqual(get_user_id(), "_anon") + + def test_legacy_claw_user_id_ignored(self) -> None: + os.environ[_legacy_env_key("USER_ID")] = "claw-only" + os.environ["JIANGCHANG_USER_ID"] = "jc-user" + self.assertEqual(get_user_id(), "jc-user") + + def test_legacy_claw_user_id_alone_ignored(self) -> None: + os.environ[_legacy_env_key("USER_ID")] = "claw-only" + self.assertEqual(get_user_id(), "_anon") + + +class TestApplyCliLocalDefaults(_EnvIsolation): + def test_sets_jiangchang_only(self) -> None: + with patch("jiangchang_skill_core.runtime_env.CLI_LOCAL_DEV_ENABLED", True): + apply_cli_local_defaults() + self.assertTrue(os.environ.get("JIANGCHANG_DATA_ROOT")) + self.assertTrue(os.environ.get("JIANGCHANG_USER_ID")) + self.assertNotIn(_legacy_env_key("DATA_ROOT"), os.environ) + self.assertNotIn(_legacy_env_key("USER_ID"), os.environ) + + def test_does_not_override_existing(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_USER_ID"] = "existing" + with patch("jiangchang_skill_core.runtime_env.CLI_LOCAL_DEV_ENABLED", True): + apply_cli_local_defaults() + self.assertEqual(os.environ["JIANGCHANG_DATA_ROOT"], tmp) + self.assertEqual(os.environ["JIANGCHANG_USER_ID"], "existing") + + +class TestGetSkillsRoot(_EnvIsolation): + def test_jiangchang_skills_root_override(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + os.environ["JIANGCHANG_SKILLS_ROOT"] = tmp + self.assertEqual(get_skills_root(), os.path.normpath(tmp)) + + def test_legacy_claw_skills_root_ignored(self) -> None: + with tempfile.TemporaryDirectory() as claw_tmp, tempfile.TemporaryDirectory() as jc_tmp: + os.environ[_legacy_env_key("SKILLS_ROOT")] = claw_tmp + os.environ["JIANGCHANG_SKILLS_ROOT"] = jc_tmp + self.assertEqual(get_skills_root(), os.path.normpath(jc_tmp)) + + def test_legacy_claw_skills_root_alone_ignored(self) -> None: + with tempfile.TemporaryDirectory() as claw_tmp: + os.environ[_legacy_env_key("SKILLS_ROOT")] = claw_tmp + root = get_skills_root() + self.assertNotEqual(root, os.path.normpath(claw_tmp)) + + +class TestGetSiblingSkillsRoot(_EnvIsolation): + def _make_skills_tree(self, base: Path) -> Path: + skills_root = base / "skills" + (skills_root / "account-manager").mkdir(parents=True) + (skills_root / "receive-order" / "scripts").mkdir(parents=True) + return skills_root + + def test_inference_preferred_over_env(self) -> None: + with tempfile.TemporaryDirectory() as tree_tmp, tempfile.TemporaryDirectory() as env_tmp: + skills_root = self._make_skills_tree(Path(tree_tmp)) + scripts_dir = str(skills_root / "receive-order" / "scripts") + os.environ["JIANGCHANG_SKILLS_ROOT"] = env_tmp + result = get_sibling_skills_root(scripts_dir) + self.assertEqual(result, os.path.normpath(str(skills_root))) + + def test_env_override_when_inference_fails(self) -> None: + with tempfile.TemporaryDirectory() as env_tmp: + os.environ["JIANGCHANG_SKILLS_ROOT"] = env_tmp + result = get_sibling_skills_root("/nonexistent/skill/scripts") + self.assertEqual(result, os.path.normpath(env_tmp)) + + def test_legacy_claw_skills_root_ignored_for_sibling(self) -> None: + with tempfile.TemporaryDirectory() as claw_tmp, tempfile.TemporaryDirectory() as jc_tmp: + os.environ[_legacy_env_key("SKILLS_ROOT")] = claw_tmp + os.environ["JIANGCHANG_SKILLS_ROOT"] = jc_tmp + result = get_sibling_skills_root("/nonexistent/skill/scripts") + self.assertEqual(result, os.path.normpath(jc_tmp)) + + +class TestConfigAndLoggingPaths(_EnvIsolation): + def test_config_env_file_under_user_isolated_dir(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_USER_ID"] = "u-test" + example = os.path.join(tmp, ".env.example") + with open(example, "w", encoding="utf-8") as f: + f.write("FOO=bar\n") + dest = config.ensure_env_file("my-skill", example) + expected_prefix = os.path.join(tmp, "u-test", "my-skill") + self.assertTrue(os.path.normpath(dest).startswith(os.path.normpath(expected_prefix))) + + def test_unified_logging_path_under_user_dir(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_USER_ID"] = "log-user" + log_dir = ul.get_unified_logs_dir() + self.assertTrue( + os.path.normpath(log_dir).startswith( + os.path.normpath(os.path.join(tmp, "log-user", "logs")) + ) + ) + + +class TestDiagnosticsNoClawFields(_EnvIsolation): + def test_diagnostics_json_and_text_exclude_claw(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + diag = collect_runtime_diagnostics(skill_slug="x", env=dict(os.environ)) + payload = runtime_diagnostics_dict(diag) + text = "\n".join(format_runtime_health_lines(diag)) + legacy_data_root_key = _legacy_env_key("DATA_ROOT") + self.assertNotIn(legacy_data_root_key, payload) + self.assertNotIn(legacy_data_root_key, text) + self.assertIn("JIANGCHANG_DATA_ROOT", payload) + self.assertEqual(payload["resolved_data_root"], tmp) + + +if __name__ == "__main__": + unittest.main()