"""Store 版本操作 + Seed 管理的单元测试。""" from __future__ import annotations import json import sqlite3 import pytest from app.harness.store import ( _parse_version, _write_meta, advance_version, extract_run_db, init_seed, init_store, list_seeds, list_versions, next_version, promote_to_seed, read_seed, ) # --------------------------------------------------------------------------- # _parse_version # --------------------------------------------------------------------------- class TestParseVersion: """_parse_version 解析 v\\d+ 格式版本号。""" def test_parse_version_normal(self) -> None: assert _parse_version("v1") == 1 assert _parse_version("v10") == 10 assert _parse_version("v999") == 999 def test_parse_version_invalid(self) -> None: with pytest.raises(ValueError, match="无效版本号"): _parse_version("abc") with pytest.raises(ValueError, match="无效版本号"): _parse_version("v") with pytest.raises(ValueError, match="无效版本号"): _parse_version("v1.0") # --------------------------------------------------------------------------- # list_versions — 数字排序 # --------------------------------------------------------------------------- class TestListVersions: """list_versions 按数字排序，v10 排在 v2 后。""" def test_list_versions_numeric_sort(self, tmp_path: "Path") -> None: """v10 必须排在 v2 后面（非字典序）。""" store = tmp_path / "store" resource = store / "skills" resource.mkdir(parents=True) for v in ("v1", "v10", "v2", "v20", "v3"): (resource / v).mkdir() result = list_versions(store, "skills") assert result == ["v1", "v2", "v3", "v10", "v20"] def test_list_versions_empty(self, tmp_path: "Path") -> None: store = tmp_path / "store" assert list_versions(store, "skills") == [] def test_list_versions_ignores_non_version_dirs(self, tmp_path: "Path") -> None: """非 v\\d+ 格式的目录被忽略。""" store = tmp_path / "store" resource = store / "skills" resource.mkdir(parents=True) (resource / "v1").mkdir() (resource / "backup").mkdir() (resource / ".hidden").mkdir() assert list_versions(store, "skills") == ["v1"] # --------------------------------------------------------------------------- # next_version # --------------------------------------------------------------------------- class TestNextVersion: """next_version 返回下一个可用版本号。""" def test_next_version_empty(self, tmp_path: "Path") -> None: store = tmp_path / "store" assert next_version(store, "skills") == "v1" def test_next_version_after_existing(self, tmp_path: "Path") -> None: store = tmp_path / "store" resource = store / "skills" resource.mkdir(parents=True) (resource / "v1").mkdir() (resource / "v2").mkdir() assert next_version(store, "skills") == "v3" def test_next_version_with_gap(self, tmp_path: "Path") -> None: """v1 和 v10 之间有 gap，next 应为 v11。""" store = tmp_path / "store" resource = store / "skills" resource.mkdir(parents=True) (resource / "v1").mkdir() (resource / "v10").mkdir() assert next_version(store, "skills") == "v11" # --------------------------------------------------------------------------- # advance_version # --------------------------------------------------------------------------- class TestAdvanceVersion: """advance_version copytree + _write_meta。""" def test_advance_version(self, tmp_path: "Path") -> None: store = tmp_path / "store" resource = store / "skills" resource.mkdir(parents=True) (resource / "v1").mkdir() source = tmp_path / "new_skills" source.mkdir() (source / "skill_a.md").write_text("内容A") version = advance_version( store, "skills", source, {"source": "evolution", "description": "进化产出"}, ) assert version == "v2" assert (resource / "v2" / "skill_a.md").read_text() == "内容A" meta = json.loads((resource / "v2" / "meta.json").read_text()) assert meta["version"] == "v2" assert meta["source"] == "evolution" assert meta["description"] == "进化产出" assert "created_at" in meta # --------------------------------------------------------------------------- # init_store # --------------------------------------------------------------------------- class TestInitStore: """init_store 初始化 Store 目录结构。""" def test_init_store(self, tmp_path: "Path") -> None: videos = tmp_path / "videos_src" videos.mkdir() (videos / "v001").mkdir() (videos / "v001" / "tree.json").write_text("{}") skills = tmp_path / "skills_src" skills.mkdir() (skills / "search.md").write_text("skill") prompts = tmp_path / "prompts_src" prompts.mkdir() (prompts / "system.md").write_text("prompt") store = tmp_path / "store" init_store(store, videos, skills, prompts) assert (store / "videos" / "v001" / "tree.json").exists() assert (store / "questions" / "benchmarks").is_dir() assert (store / "questions" / "generated").is_dir() assert (store / "skills" / "v1" / "search.md").read_text() == "skill" assert (store / "prompts" / "v1" / "system.md").read_text() == "prompt" skills_meta = json.loads( (store / "skills" / "v1" / "meta.json").read_text() ) assert skills_meta["version"] == "v1" assert skills_meta["source"] == "manual" def test_init_store_exists_raises(self, tmp_path: "Path") -> None: store = tmp_path / "store" store.mkdir() with pytest.raises(FileExistsError, match="Store 已存在"): init_store(store, tmp_path, tmp_path, tmp_path) # --------------------------------------------------------------------------- # Seed 相关 # --------------------------------------------------------------------------- def _make_seed_fixtures(tmp_path): """创建 seed 测试所需的公共 fixture。""" store = tmp_path / "store" store.mkdir(parents=True) skills_dir = tmp_path / "sk" skills_dir.mkdir() (skills_dir / "search.md").write_text("skill") prompts_dir = tmp_path / "pr" prompts_dir.mkdir() (prompts_dir / "system.md").write_text("prompt") baseline_db = tmp_path / "base.db" conn = sqlite3.connect(baseline_db) conn.execute( "CREATE TABLE _runs (run_id TEXT PRIMARY KEY, status TEXT)" ) conn.execute("INSERT INTO _runs VALUES ('r1', 'done')") conn.execute( "CREATE TABLE predictions (run_id TEXT, question_id TEXT, answer TEXT)" ) conn.execute("INSERT INTO predictions VALUES ('r1', 'q1', 'A')") conn.commit() conn.close() return store, skills_dir, prompts_dir, baseline_db class TestInitSeed: """init_seed 创建种子目录。""" def test_init_seed(self, tmp_path: "Path") -> None: store, skills_dir, prompts_dir, baseline_db = _make_seed_fixtures(tmp_path) seed_dir = init_seed( store, "initial", skills_dir, prompts_dir, baseline_db, "r1", None, "初始种子", ) assert seed_dir == store / "seeds" / "initial" assert (seed_dir / "skills" / "search.md").exists() assert (seed_dir / "prompts" / "system.md").exists() assert (seed_dir / "baseline.db").exists() meta = json.loads((seed_dir / "seed.json").read_text()) assert meta["baseline_run_id"] == "r1" assert meta["parent"] is None assert meta["description"] == "初始种子" assert "created_at" in meta def test_init_seed_exists_raises(self, tmp_path: "Path") -> None: store, skills_dir, prompts_dir, baseline_db = _make_seed_fixtures(tmp_path) init_seed(store, "dup", skills_dir, prompts_dir, baseline_db, "r1", None, "first") with pytest.raises(FileExistsError, match="种子已存在"): init_seed( store, "dup", skills_dir, prompts_dir, baseline_db, "r1", None, "second" ) class TestListSeeds: """list_seeds 列出所有种子。""" def test_list_seeds(self, tmp_path: "Path") -> None: store, skills_dir, prompts_dir, baseline_db = _make_seed_fixtures(tmp_path) init_seed(store, "beta", skills_dir, prompts_dir, baseline_db, "r1", None, "b") init_seed(store, "alpha", skills_dir, prompts_dir, baseline_db, "r1", None, "a") assert list_seeds(store) == ["alpha", "beta"] def test_list_seeds_empty(self, tmp_path: "Path") -> None: store = tmp_path / "store" assert list_seeds(store) == [] class TestReadSeed: """read_seed 读取 seed.json。""" def test_read_seed(self, tmp_path: "Path") -> None: store, skills_dir, prompts_dir, baseline_db = _make_seed_fixtures(tmp_path) init_seed(store, "s1", skills_dir, prompts_dir, baseline_db, "r1", None, "desc") meta = read_seed(store, "s1") assert meta["baseline_run_id"] == "r1" assert meta["description"] == "desc" def test_read_seed_not_found(self, tmp_path: "Path") -> None: store = tmp_path / "store" store.mkdir() with pytest.raises(FileNotFoundError, match="种子不存在"): read_seed(store, "no_such") # --------------------------------------------------------------------------- # extract_run_db # --------------------------------------------------------------------------- class TestExtractRunDb: """extract_run_db 抽取指定 run 的行并保留 PK。""" def _make_src_db(self, path): """创建带 _runs + predictions 表的源 db。""" conn = sqlite3.connect(path) conn.execute( "CREATE TABLE _runs (run_id TEXT PRIMARY KEY, status TEXT)" ) conn.execute("INSERT INTO _runs VALUES ('r1', 'done')") conn.execute("INSERT INTO _runs VALUES ('r2', 'done')") conn.execute( "CREATE TABLE predictions " "(run_id TEXT, question_id TEXT, answer TEXT)" ) conn.execute("INSERT INTO predictions VALUES ('r1', 'q1', 'A')") conn.execute("INSERT INTO predictions VALUES ('r1', 'q2', 'B')") conn.execute("INSERT INTO predictions VALUES ('r2', 'q1', 'C')") conn.commit() conn.close() def test_extract_run_db_preserves_pk(self, tmp_path: "Path") -> None: """原始 CREATE 保留主键约束。""" src = tmp_path / "src.db" dst = tmp_path / "dst.db" self._make_src_db(src) extract_run_db(src, dst, "r1") conn = sqlite3.connect(dst) # 验证 _runs 表有 PK create_sql = conn.execute( "SELECT sql FROM sqlite_master WHERE type='table' AND name='_runs'" ).fetchone()[0] assert "PRIMARY KEY" in create_sql # 验证只有 r1 的行 runs = conn.execute("SELECT * FROM _runs").fetchall() assert len(runs) == 1 assert runs[0][0] == "r1" preds = conn.execute("SELECT * FROM predictions").fetchall() assert len(preds) == 2 conn.close() def test_extract_run_db_missing_table(self, tmp_path: "Path") -> None: """源 db 无目标表时报错。""" src = tmp_path / "src.db" dst = tmp_path / "dst.db" conn = sqlite3.connect(src) conn.execute("CREATE TABLE other (id TEXT)") conn.commit() conn.close() with pytest.raises(RuntimeError, match="源 db 无表"): extract_run_db(src, dst, "r1") def test_extract_run_db_no_rows(self, tmp_path: "Path") -> None: """目标 run_id 不存在时报错。""" src = tmp_path / "src.db" dst = tmp_path / "dst.db" self._make_src_db(src) with pytest.raises(RuntimeError, match="无 run_id="): extract_run_db(src, dst, "nonexistent") # --------------------------------------------------------------------------- # promote_to_seed # --------------------------------------------------------------------------- def _make_promote_fixtures(tmp_path): """创建 promote_to_seed 测试所需的 workspace + store。""" ws = tmp_path / "ws" ws.mkdir() store = tmp_path / "store" store.mkdir() # workspace 内的 skills/prompts 版本目录 (ws / "skills" / "v2").mkdir(parents=True) (ws / "skills" / "v2" / "skill.md").write_text("evolved") (ws / "prompts" / "v2").mkdir(parents=True) (ws / "prompts" / "v2" / "system.md").write_text("prompt v2") # workspace harness.db db_path = ws / "harness.db" conn = sqlite3.connect(db_path) conn.execute(""" CREATE TABLE _runs ( run_id TEXT PRIMARY KEY, skills_version TEXT, prompts_version TEXT ) """) conn.execute( "INSERT INTO _runs VALUES ('eval_001', 'v2', 'v2')" ) conn.execute(""" CREATE TABLE predictions ( run_id TEXT, question_id TEXT, answer TEXT ) """) conn.execute("INSERT INTO predictions VALUES ('eval_001', 'q1', 'A')") conn.commit() conn.close() return ws, store class TestPromoteToSeed: """promote_to_seed 固化 workspace 版本为种子。""" def test_promote_to_seed_success(self, tmp_path: "Path") -> None: ws, store = _make_promote_fixtures(tmp_path) seed_dir = promote_to_seed(ws, store, "v2", "eval_001", "evolved-seed", "good") assert seed_dir == store / "seeds" / "evolved-seed" assert (seed_dir / "skills" / "skill.md").exists() assert (seed_dir / "prompts" / "system.md").exists() assert (seed_dir / "baseline.db").exists() meta = json.loads((seed_dir / "seed.json").read_text()) assert meta["baseline_run_id"] == "eval_001" assert meta["parent"] == "ws:v2" def test_promote_to_seed_version_mismatch(self, tmp_path: "Path") -> None: """eval run 的 skills_version 与 --version 不符时报错。""" ws, store = _make_promote_fixtures(tmp_path) with pytest.raises(ValueError, match="版本.*不符"): promote_to_seed(ws, store, "v3", "eval_001", "bad", "mismatch") def test_promote_to_seed_null_version(self, tmp_path: "Path") -> None: """eval run 的版本为 NULL 时报错。""" ws = tmp_path / "ws2" ws.mkdir() store = tmp_path / "store2" store.mkdir() db_path = ws / "harness.db" conn = sqlite3.connect(db_path) conn.execute(""" CREATE TABLE _runs ( run_id TEXT PRIMARY KEY, skills_version TEXT, prompts_version TEXT ) """) conn.execute("INSERT INTO _runs VALUES ('eval_null', NULL, NULL)") conn.commit() conn.close() with pytest.raises(ValueError, match="NULL"): promote_to_seed(ws, store, "v1", "eval_null", "bad", "null ver") def test_promote_to_seed_run_not_found(self, tmp_path: "Path") -> None: """eval run 不存在时报错。""" ws, store = _make_promote_fixtures(tmp_path) with pytest.raises(ValueError, match="eval run 不存在"): promote_to_seed(ws, store, "v1", "nonexistent", "bad", "no run") def test_promote_cleanup_tmp_db(self, tmp_path: "Path") -> None: """finally 清理临时 db 文件。""" ws, store = _make_promote_fixtures(tmp_path) promote_to_seed(ws, store, "v2", "eval_001", "clean-test", "cleanup") # 临时 db 应已清理 assert not (ws / "_promote_tmp.db").exists() def test_promote_cleanup_tmp_db_on_error(self, tmp_path: "Path") -> None: """即使 init_seed 失败（同名种子），临时 db 也应被清理。""" ws, store = _make_promote_fixtures(tmp_path) promote_to_seed(ws, store, "v2", "eval_001", "first", "first time") with pytest.raises(FileExistsError): promote_to_seed(ws, store, "v2", "eval_001", "first", "second time") assert not (ws / "_promote_tmp.db").exists()