"""app/harness/workspace 单元测试。 覆盖 Workspace 生命周期管理、manifest 读写、 VersionedSkillStore / VersionedPromptStore Protocol 合规。 """ from __future__ import annotations import json import shutil from pathlib import Path import pytest from app.harness.workspace import ( ResolvedPaths, VersionedPromptStore, VersionedSkillStore, archive_workspace, init_workspace, init_workspace_from_seed, list_video_ids, load_manifest, read_best, record_run, resolve_paths, update_best, update_manifest, ) from core.evolution.protocols import PromptStore, SkillStore # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture() def store_dir(tmp_path: Path) -> Path: """构建一个最小 Store 目录结构供测试使用。""" sd = tmp_path / "store" sd.mkdir() # videos：两个含 tree.json 的视频目录 for vid in ("vid_A", "vid_B"): vdir = sd / "videos" / vid vdir.mkdir(parents=True) (vdir / "tree.json").write_text("{}") # skills/v1 + prompts/v1 s1 = sd / "skills" / "v1" s1.mkdir(parents=True) (s1 / "temporal-reasoning.md").write_text("skill content A") (s1 / "spatial-analysis.md").write_text("skill content B") p1 = sd / "prompts" / "v1" p1.mkdir(parents=True) (p1 / "system.md").write_text("system prompt v1") (p1 / "extract.md").write_text("extract prompt v1") # questions/benchmarks/Video-MME q = sd / "questions" / "benchmarks" / "Video-MME" q.mkdir(parents=True) (q / "q1.json").write_text('{"id": "q1"}') # seed "initial" seed_dir = sd / "seeds" / "initial" seed_dir.mkdir(parents=True) shutil.copytree(s1, seed_dir / "skills") shutil.copytree(p1, seed_dir / "prompts") baseline_db = seed_dir / "baseline.db" baseline_db.write_bytes(b"") (seed_dir / "seed.json").write_text( json.dumps({"baseline_run_id": "baseline-001", "parent": None}) ) return sd @pytest.fixture() def workspace_dir(tmp_path: Path) -> Path: """返回一个不存在的 workspace 路径。""" return tmp_path / "ws_test" # --------------------------------------------------------------------------- # ResolvedPaths # --------------------------------------------------------------------------- def test_resolved_paths_frozen() -> None: """ResolvedPaths 实例应不可变。""" rp = ResolvedPaths( store_dir=Path("/s"), videos_dir=Path("/v"), questions_dir=Path("/q"), skills_dir=Path("/sk"), prompts_dir=Path("/p"), workspace_dir=Path("/w"), db_path=Path("/d"), analyses_dir=Path("/a"), runs_dir=Path("/r"), ) with pytest.raises(AttributeError): rp.store_dir = Path("/other") # type: ignore[misc] # --------------------------------------------------------------------------- # init_workspace # --------------------------------------------------------------------------- def test_init_workspace(store_dir: Path, workspace_dir: Path) -> None: """init_workspace 应创建 manifest、拷贝权重。""" init_workspace( workspace_dir, store_dir, questions="benchmarks/Video-MME", skills_version="v1", prompts_version="v1", ) # manifest 存在 manifest = json.loads((workspace_dir / "manifest.json").read_text()) assert manifest["current"]["skills"] == "skills/v1" assert manifest["current"]["prompts"] == "prompts/v1" assert manifest["current"]["questions"] == "questions/benchmarks/Video-MME" # 权重已拷入 workspace assert (workspace_dir / "skills" / "v1" / "temporal-reasoning.md").exists() assert (workspace_dir / "prompts" / "v1" / "system.md").exists() # analyses, runs 目录已创建 assert (workspace_dir / "analyses").is_dir() assert (workspace_dir / "runs").is_dir() # 重复创建应报错 with pytest.raises(FileExistsError): init_workspace( workspace_dir, store_dir, questions="benchmarks/Video-MME", skills_version="v1", prompts_version="v1", ) # --------------------------------------------------------------------------- # init_workspace_from_seed # --------------------------------------------------------------------------- def test_init_workspace_from_seed(store_dir: Path, workspace_dir: Path) -> None: """从种子初始化应拷权重 + baseline.db + 返回 baseline_run_id。""" run_id = init_workspace_from_seed( workspace_dir, store_dir, seed_name="initial", questions="benchmarks/Video-MME", ) assert run_id == "baseline-001" # 权重在 workspace assert (workspace_dir / "skills" / "v1" / "temporal-reasoning.md").exists() assert (workspace_dir / "prompts" / "v1" / "system.md").exists() # baseline.db -> harness.db assert (workspace_dir / "harness.db").exists() # manifest 正确 manifest = json.loads((workspace_dir / "manifest.json").read_text()) assert manifest["current"]["skills"] == "skills/v1" def test_init_workspace_from_seed_missing_questions(store_dir: Path, workspace_dir: Path) -> None: """questions ref 不存在应 fail-fast（FileNotFoundError），不创建 workspace。""" with pytest.raises(FileNotFoundError, match="questions ref"): init_workspace_from_seed( workspace_dir, store_dir, seed_name="initial", questions="benchmarks/NONEXISTENT", ) # workspace 不应被创建 assert not workspace_dir.exists() # --------------------------------------------------------------------------- # load_manifest # --------------------------------------------------------------------------- def test_load_manifest(store_dir: Path, workspace_dir: Path) -> None: """正常加载 manifest。""" init_workspace( workspace_dir, store_dir, questions="benchmarks/Video-MME", skills_version="v1", prompts_version="v1", ) manifest = load_manifest(workspace_dir) assert "current" in manifest assert "history" in manifest assert manifest["current"]["skills"] == "skills/v1" def test_load_manifest_missing(workspace_dir: Path) -> None: """manifest 不存在应 FileNotFoundError。""" workspace_dir.mkdir(parents=True) with pytest.raises(FileNotFoundError): load_manifest(workspace_dir) # --------------------------------------------------------------------------- # update_manifest # --------------------------------------------------------------------------- def test_update_manifest_invalid_key(store_dir: Path, workspace_dir: Path) -> None: """非法 key 应 KeyError。""" init_workspace( workspace_dir, store_dir, questions="benchmarks/Video-MME", skills_version="v1", prompts_version="v1", ) with pytest.raises(KeyError, match="无效"): update_manifest(workspace_dir, bad_key="skills/v2") def test_update_manifest_valid(store_dir: Path, workspace_dir: Path) -> None: """合法 key 应更新 manifest current。""" init_workspace( workspace_dir, store_dir, questions="benchmarks/Video-MME", skills_version="v1", prompts_version="v1", ) update_manifest(workspace_dir, skills="skills/v2") manifest = load_manifest(workspace_dir) assert manifest["current"]["skills"] == "skills/v2" # --------------------------------------------------------------------------- # record_run # --------------------------------------------------------------------------- def test_record_run_idempotent(store_dir: Path, workspace_dir: Path) -> None: """同 run_id 调用两次不应重复追加 history。""" init_workspace( workspace_dir, store_dir, questions="benchmarks/Video-MME", skills_version="v1", prompts_version="v1", ) run_dir1 = record_run(workspace_dir, "run_001") run_dir2 = record_run(workspace_dir, "run_001") # 返回路径一致 assert run_dir1 == run_dir2 # history 只有一条 manifest = load_manifest(workspace_dir) matched = [h for h in manifest["history"] if h["run_id"] == "run_001"] assert len(matched) == 1 # run 目录存在 assert run_dir1.is_dir() # per-video wiki 目录存在 assert (run_dir1 / "vid_A" / "wiki").is_dir() assert (run_dir1 / "vid_B" / "wiki").is_dir() # --------------------------------------------------------------------------- # update_best / read_best # --------------------------------------------------------------------------- def test_update_best_independent_of_current(store_dir: Path, workspace_dir: Path) -> None: """best 应独立于 current——更新 best 不影响 current。""" init_workspace( workspace_dir, store_dir, questions="benchmarks/Video-MME", skills_version="v1", prompts_version="v1", ) # 初始无 best assert read_best(workspace_dir) is None # 设置 best update_best( workspace_dir, skills="skills/v3", prompts="prompts/v3", val_acc=0.85, run_id="run_005", epoch=3, ) best = read_best(workspace_dir) assert best is not None assert best["skills"] == "skills/v3" assert best["val_acc"] == 0.85 assert best["epoch"] == 3 # current 不受影响 manifest = load_manifest(workspace_dir) assert manifest["current"]["skills"] == "skills/v1" # --------------------------------------------------------------------------- # archive_workspace # --------------------------------------------------------------------------- def test_archive_workspace(store_dir: Path, workspace_dir: Path) -> None: """归档应把 workspace 移到 .archive/ 下。""" init_workspace( workspace_dir, store_dir, questions="benchmarks/Video-MME", skills_version="v1", prompts_version="v1", ) archived = archive_workspace(workspace_dir) # 原目录不存在 assert not workspace_dir.exists() # 归档目录存在并包含 manifest assert archived.is_dir() assert (archived / "manifest.json").exists() # 归档路径在 .archive 下 assert archived.parent.name == ".archive" def test_archive_workspace_missing(workspace_dir: Path) -> None: """归档不存在的 workspace 应 FileNotFoundError。""" with pytest.raises(FileNotFoundError): archive_workspace(workspace_dir) # --------------------------------------------------------------------------- # list_video_ids # --------------------------------------------------------------------------- def test_list_video_ids(store_dir: Path, workspace_dir: Path) -> None: """列出 workspace 引用的所有视频 ID。""" init_workspace( workspace_dir, store_dir, questions="benchmarks/Video-MME", skills_version="v1", prompts_version="v1", ) ids = list_video_ids(workspace_dir) assert ids == ["vid_A", "vid_B"] # --------------------------------------------------------------------------- # resolve_paths # --------------------------------------------------------------------------- def test_resolve_paths(store_dir: Path, workspace_dir: Path) -> None: """resolve_paths 应正确解析绝对路径。""" init_workspace( workspace_dir, store_dir, questions="benchmarks/Video-MME", skills_version="v1", prompts_version="v1", ) rp = resolve_paths(workspace_dir) # skills_dir/prompts_dir 解析到 workspace（非 store） ws_abs = workspace_dir.resolve() assert rp.skills_dir == ws_abs / "skills" / "v1" assert rp.prompts_dir == ws_abs / "prompts" / "v1" # videos/questions 解析到 store store_abs = store_dir.resolve() assert rp.videos_dir == store_abs / "videos" assert rp.questions_dir == store_abs / "questions" / "benchmarks" / "Video-MME" # db_path, analyses, runs assert rp.db_path == ws_abs / "harness.db" assert rp.analyses_dir == ws_abs / "analyses" assert rp.runs_dir == ws_abs / "runs" # --------------------------------------------------------------------------- # VersionedSkillStore # --------------------------------------------------------------------------- def test_versioned_skill_store_read(store_dir: Path, workspace_dir: Path) -> None: """read_skill 应返回文件全文。""" init_workspace( workspace_dir, store_dir, questions="benchmarks/Video-MME", skills_version="v1", prompts_version="v1", ) skills_dir = workspace_dir / "skills" / "v1" store = VersionedSkillStore(skills_dir) content = store.read_skill("temporal-reasoning.md") assert content == "skill content A" def test_versioned_skill_store_read_missing(store_dir: Path, workspace_dir: Path) -> None: """读取不存在的 skill 文件应 FileNotFoundError。""" init_workspace( workspace_dir, store_dir, questions="benchmarks/Video-MME", skills_version="v1", prompts_version="v1", ) skills_dir = workspace_dir / "skills" / "v1" store = VersionedSkillStore(skills_dir) with pytest.raises(FileNotFoundError): store.read_skill("nonexistent.md") def test_versioned_skill_store_list(store_dir: Path, workspace_dir: Path) -> None: """list_skill_files 应列出所有 skill 文件名（排序）。""" init_workspace( workspace_dir, store_dir, questions="benchmarks/Video-MME", skills_version="v1", prompts_version="v1", ) skills_dir = workspace_dir / "skills" / "v1" store = VersionedSkillStore(skills_dir) files = store.list_skill_files() assert sorted(files) == ["spatial-analysis.md", "temporal-reasoning.md"] # --------------------------------------------------------------------------- # VersionedPromptStore # --------------------------------------------------------------------------- def test_versioned_prompt_store(store_dir: Path, workspace_dir: Path) -> None: """VersionedPromptStore 读写功能验证。""" init_workspace( workspace_dir, store_dir, questions="benchmarks/Video-MME", skills_version="v1", prompts_version="v1", ) prompts_dir = workspace_dir / "prompts" / "v1" store = VersionedPromptStore(prompts_dir) content = store.read_prompt("system.md") assert content == "system prompt v1" files = store.list_prompt_files() assert sorted(files) == ["extract.md", "system.md"] def test_versioned_prompt_store_read_missing(store_dir: Path, workspace_dir: Path) -> None: """读取不存在的 prompt 文件应 FileNotFoundError。""" init_workspace( workspace_dir, store_dir, questions="benchmarks/Video-MME", skills_version="v1", prompts_version="v1", ) prompts_dir = workspace_dir / "prompts" / "v1" store = VersionedPromptStore(prompts_dir) with pytest.raises(FileNotFoundError): store.read_prompt("nonexistent.md") # --------------------------------------------------------------------------- # Protocol 合规 # --------------------------------------------------------------------------- def test_skill_store_protocol_compliance(store_dir: Path, workspace_dir: Path) -> None: """VersionedSkillStore 应满足 core/evolution/protocols.py::SkillStore Protocol。""" init_workspace( workspace_dir, store_dir, questions="benchmarks/Video-MME", skills_version="v1", prompts_version="v1", ) skills_dir = workspace_dir / "skills" / "v1" store = VersionedSkillStore(skills_dir) assert isinstance(store, SkillStore) def test_prompt_store_protocol_compliance(store_dir: Path, workspace_dir: Path) -> None: """VersionedPromptStore 应满足 core/evolution/protocols.py::PromptStore Protocol。""" init_workspace( workspace_dir, store_dir, questions="benchmarks/Video-MME", skills_version="v1", prompts_version="v1", ) prompts_dir = workspace_dir / "prompts" / "v1" store = VersionedPromptStore(prompts_dir) assert isinstance(store, PromptStore)