246 lines
9.4 KiB
Python
246 lines
9.4 KiB
Python
"""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) == []
|