"""S2：SQLite 会话存储测试（server/store.py）。 覆盖：创建会话 / 列表 / 状态更新 / 文件登记 / 结果路径 / 持久化重建 / 会话命名与项目绑定 / 项目配置存储。 """ from __future__ import annotations import pytest from pathlib import Path from genesis.server.store import ( SessionStore, SessionRecord, SessionNotFoundError, ProjectsStore, ProjectConfigError, ) @pytest.fixture def store(tmp_path): return SessionStore(db_path=str(tmp_path / "sessions.db")) def test_create_session(store): s = store.create_session(user_id="u1") assert s.session_id assert s.user_id == "u1" assert s.status == "uploading" assert s.files == {} assert s.created_at def test_get_session(store): s = store.create_session("u1") got = store.get_session(s.session_id) assert got.session_id == s.session_id assert got.status == "uploading" def test_get_missing_session_raises(store): with pytest.raises(SessionNotFoundError): store.get_session("nope") def test_list_sessions_by_user(store): a = store.create_session("u1") b = store.create_session("u1") store.create_session("u2") lst = store.list_sessions("u1") ids = {s.session_id for s in lst} assert ids == {a.session_id, b.session_id} def test_update_status(store): s = store.create_session("u1") store.update_status(s.session_id, "parsing") assert store.get_session(s.session_id).status == "parsing" def test_update_fields_merge(store): s = store.create_session("u1") store.update_session(s.session_id, files={"requirements": {"file_id": "f1", "name": "a.xlsx", "size": 10}}) got = store.get_session(s.session_id) assert got.files["requirements"]["file_id"] == "f1" # 保留既有字段 assert got.status == "uploading" def test_set_result_paths(store): s = store.create_session("u1") store.update_session(s.session_id, result_path="out.docx", impact_report_path="ir.json", qa_report_path="qa.json") got = store.get_session(s.session_id) assert got.result_path == "out.docx" assert got.impact_report_path == "ir.json" assert got.qa_report_path == "qa.json" def test_store_reload_persists(tmp_path): db = str(tmp_path / "s.db") store1 = SessionStore(db_path=db) s = store1.create_session("u1") store1.update_status(s.session_id, "done") store1.update_session(s.session_id, result_path="x.docx") # 重新打开同一 db → 数据仍在 store2 = SessionStore(db_path=db) got = store2.get_session(s.session_id) assert got.status == "done" assert got.result_path == "x.docx" # ---------- 会话命名与项目绑定 ---------- def test_session_record_name_and_project_defaults(): rec = SessionRecord(session_id="s1", user_id="u1") assert rec.name == "新会话" assert rec.project == "" assert rec.to_dict["name"] == "新会话" assert rec.to_dict["project"] == "" def test_create_session_with_name_and_project(store): s = store.create_session("u1", name="我的会话", project="projA") got = store.get_session(s.session_id) assert got.name == "我的会话" assert got.project == "projA" def test_session_persists_name_project(tmp_path): db = str(tmp_path / "s.db") s = SessionStore(db_path=db).create_session("u1", name="名字", project="p") reopened = SessionStore(db_path=db).get_session(s.session_id) assert reopened.name == "名字" assert reopened.project == "p" # ---------- 项目配置存储（ProjectsStore） ---------- @pytest.fixture def projects_store(tmp_path): return ProjectsStore(db_path=str(tmp_path / "projects.db")) def _make_docx(path: Path) -> str: path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(b"PK\x03\x04") # 最小占位，仅用于存在性/后缀校验 return str(path) def test_projects_upsert_and_get(projects_store, tmp_path): tpl = _make_docx(tmp_path / "t.docx") wi = _make_docx(tmp_path / "wi.docx") rules_dir = tmp_path / "rules" rules_dir.mkdir() _make_docx(rules_dir / "r1.docx") cfg = projects_store.upsert( name="stock", display_name="股票系统", template=tpl, write_instruction=wi, rules=[str(rules_dir)], existing_system_code_dir="", design_docs_dir="", ) assert cfg.name == "stock" got = projects_store.get("stock") assert got is not None assert got.display_name == "股票系统" assert got.template == tpl assert got.rules == [str(rules_dir / "r1.docx")] def test_projects_list_and_delete(projects_store, tmp_path): _make_docx(tmp_path / "t.docx") projects_store.upsert(name="p1", display_name="P1", template=str(tmp_path / "t.docx"), write_instruction="", rules=[], existing_system_code_dir="", design_docs_dir="") assert any(p.name == "p1" for p in projects_store.list()) assert projects_store.delete("p1") is True assert projects_store.get("p1") is None def test_projects_upsert_enumerates_design_docs_dir(projects_store, tmp_path): _make_docx(tmp_path / "t.docx") dd = tmp_path / "design" dd.mkdir() _make_docx(dd / "d1.docx") _make_docx(dd / "d2.docx") cfg = projects_store.upsert(name="p", display_name="P", template=str(tmp_path / "t.docx"), write_instruction="", rules=[], existing_system_code_dir="", design_docs_dir=str(dd)) assert set(cfg.rules) == set() # 仅 design_docs_dir，rules 为空 assert len(cfg.rules) == 0 def test_projects_upsert_rejects_missing_template(projects_store, tmp_path): with pytest.raises(ProjectConfigError): projects_store.upsert(name="p", display_name="P", template=str(tmp_path / "nope.docx"), write_instruction="", rules=[], existing_system_code_dir="", design_docs_dir="") def test_projects_upsert_rejects_non_docx(projects_store, tmp_path): bad = tmp_path / "x.txt" bad.write_text("x") with pytest.raises(ProjectConfigError): projects_store.upsert(name="p", display_name="P", template=str(bad), write_instruction="", rules=[], existing_system_code_dir="", design_docs_dir="") def test_projects_upsert_rejects_missing_code_dir(projects_store, tmp_path): _make_docx(tmp_path / "t.docx") with pytest.raises(ProjectConfigError): projects_store.upsert(name="p", display_name="P", template=str(tmp_path / "t.docx"), write_instruction="", rules=[], existing_system_code_dir=str(tmp_path / "missing"), design_docs_dir="") def test_projects_reupsert_updates(projects_store, tmp_path): _make_docx(tmp_path / "t.docx") projects_store.upsert(name="p", display_name="P", template=str(tmp_path / "t.docx"), write_instruction="", rules=[], existing_system_code_dir="", design_docs_dir="") projects_store.upsert(name="p", display_name="P2", template=str(tmp_path / "t.docx"), write_instruction="", rules=[], existing_system_code_dir="", design_docs_dir="") assert projects_store.get("p").display_name == "P2" def test_projects_code_dir_must_be_directory(projects_store, tmp_path): _make_docx(tmp_path / "t.docx") (tmp_path / "afile.txt").write_text("x") with pytest.raises(ProjectConfigError): projects_store.upsert(name="p", display_name="P", template=str(tmp_path / "t.docx"), write_instruction="", rules=[], existing_system_code_dir=str(tmp_path / "afile.txt"), design_docs_dir="") def test_projects_empty_template_skips_validation(projects_store, tmp_path): # 模板留空：不应触发文件校验（line 271/273 分支） cfg = projects_store.upsert(name="p", display_name="P", template="", write_instruction="", rules=[], existing_system_code_dir="", design_docs_dir="") assert cfg.template == "" def test_projects_upsert_empty_name_rejected(projects_store, tmp_path): with pytest.raises(ProjectConfigError): projects_store.upsert(name="", display_name="P", template="", write_instruction="", rules=[], existing_system_code_dir="", design_docs_dir="") def test_list_sessions_filter_by_project(store): a = store.create_session("u1", name="A", project="stock") b = store.create_session("u1", name="B", project="other") recs = store.list_sessions("u1", project="stock") ids = {r.session_id for r in recs} assert a.session_id in ids and b.session_id not in ids def test_list_sessions_filter_empty_project(store): e = store.create_session("u1", name="E") store.create_session("u1", name="X", project="stock") recs = store.list_sessions("u1", project="") assert all(r.project == "" for r in recs) assert e.session_id in {r.session_id for r in recs} def test_list_sessions_no_project_returns_all(store): store.create_session("u1", name="A", project="stock") store.create_session("u1", name="B", project="other") assert len(store.list_sessions("u1")) == 2 def test_delete_session_cascades_messages(store): rec = store.create_session("u1", name="A", project="stock") store.add_message(rec.session_id, "user", "hi") store.add_message(rec.session_id, "assistant", "ok") assert len(store.list_messages(rec.session_id)) == 2 ok = store.delete_session(rec.session_id) assert ok is True assert store.list_messages(rec.session_id) == []