lhl
215c605650
参赛成果物 03「交互界面 + 数据存储」落地：
- src/genesis/server/: store.py（SQLite 会话持久化）、service.py（会话化服务层：
上传→解析→确认→影响→确认→生成→QA）、app.py（api-design §2 核心端点 9 组）、
static/index.html（内嵌单页，零构建无 node_modules 依赖）
- scripts/serve.py 启动入口（--fake 离线引擎 / 默认真实 LLM）
- pyproject 加 fastapi/uvicorn/python-multipart
- 修复 qa_loop._build meta={} 导致真实模板 {{doc_title}} 等占位符残留 DocxInjectError
- README Web 服务说明 + service_url 登记指引；design.md §12.5 记录（含同步执行/
zip 既有系统/无 WebSocket 的诚实偏差标注）
- 测试：test_server_store/service/api 共 34 用例（TestClient 全链路 + zip 影响流程 + 错误分支）
全量 pytest 473 passed / 99.20%
82 lines
2.5 KiB
Python
82 lines
2.5 KiB
Python
"""S2：SQLite 会话存储测试（server/store.py）。
|
||
|
||
覆盖：创建会话 / 列表 / 状态更新 / 文件登记 / 结果路径 / 持久化重建。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from genesis.server.store import SessionStore, SessionRecord, SessionNotFoundError
|
||
|
||
|
||
@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"
|