337 lines
13 KiB
Python
337 lines
13 KiB
Python
"""S3：会话化服务层测试（server/service.py）。
|
||
|
||
用真实样本文件 + FakeEngine 验证闭环：上传 → 解析 → 确认 → 影响 → 确认 → 生成 → QA。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
|
||
import asyncio
|
||
import pytest
|
||
|
||
from genesis.server.store import SessionStore, ProjectsStore
|
||
from genesis.server.service import GenesisService, ServiceStepError, FileTypeError
|
||
|
||
_SAMPLE = Path(__file__).resolve().parents[1] / "sample"
|
||
|
||
|
||
class FakeEngine:
|
||
def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2):
|
||
title = variables.get("title", "x")
|
||
# 日文样本（章节标题含假名）→ 返回日文正文，满足语言一致性强制
|
||
return SimpleNamespace(
|
||
data={
|
||
"title": title,
|
||
"blocks": [
|
||
{"type": "paragraph",
|
||
"text": "本機能はFakeLLMにより生成された十分な説明内容であり、書込規則を満たす。"},
|
||
],
|
||
},
|
||
status="ok",
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def svc(tmp_path):
|
||
return GenesisService(
|
||
store=SessionStore(db_path=str(tmp_path / "s.db")),
|
||
data_root=str(tmp_path / "data"),
|
||
engine=FakeEngine(),
|
||
)
|
||
|
||
|
||
def _upload_core(svc, session_id):
|
||
svc.upload_file(session_id, "requirements", "requirements_newdev.xlsx",
|
||
(_SAMPLE / "requirements_newdev.xlsx").read_bytes())
|
||
svc.upload_file(session_id, "template", "template_design_ja.docx",
|
||
(_SAMPLE / "template_design_ja.docx").read_bytes())
|
||
svc.upload_file(session_id, "write_instruction", "rules_design_ja.docx",
|
||
(_SAMPLE / "rules_design_ja.docx").read_bytes())
|
||
svc.upload_file(session_id, "rules", "rules_entry_ja.docx",
|
||
(_SAMPLE / "rules_entry_ja.docx").read_bytes())
|
||
|
||
|
||
def test_create_and_upload_files(svc):
|
||
s = svc.create_session("u1")
|
||
_upload_core(svc, s.session_id)
|
||
got = svc.get_session(s.session_id)
|
||
assert set(got.files.keys()) == {"requirements", "template", "write_instruction", "rules"}
|
||
assert got.files["requirements"]["name"] == "requirements_newdev.xlsx"
|
||
|
||
|
||
def test_create_session_name_and_project(tmp_path):
|
||
svc = GenesisService(
|
||
store=SessionStore(db_path=str(tmp_path / "s.db")),
|
||
data_root=str(tmp_path / "data"),
|
||
engine=FakeEngine(),
|
||
)
|
||
s = svc.create_session("u1", name="我的会话", project="projA")
|
||
got = svc.get_session(s.session_id)
|
||
assert got.name == "我的会话"
|
||
assert got.project == "projA"
|
||
|
||
|
||
def test_upload_requirements_auto_names_session(tmp_path):
|
||
svc = GenesisService(
|
||
store=SessionStore(db_path=str(tmp_path / "s.db")),
|
||
data_root=str(tmp_path / "data"),
|
||
engine=FakeEngine(),
|
||
)
|
||
s = svc.create_session("u1") # 默认「新会话」
|
||
svc.upload_file(s.session_id, "requirements", "要件定义_v2.xlsx", b"PK\x03\x04")
|
||
assert svc.get_session(s.session_id).name == "要件定义_v2"
|
||
|
||
|
||
def test_has_file_falls_back_to_project_config(tmp_path):
|
||
store = SessionStore(db_path=str(tmp_path / "s.db"))
|
||
projects = ProjectsStore(db_path=str(tmp_path / "p.db"))
|
||
projects.upsert(name="projA", display_name="P", template=str(_SAMPLE / "template_design_ja.docx"),
|
||
write_instruction="", rules=[], existing_system_code_dir="", design_docs_dir="")
|
||
svc = GenesisService(
|
||
store=store, data_root=str(tmp_path / "data"), engine=FakeEngine(), projects=projects,
|
||
)
|
||
s = svc.create_session("u1", project="projA")
|
||
svc.upload_file(s.session_id, "requirements", "requirements_newdev.xlsx",
|
||
(_SAMPLE / "requirements_newdev.xlsx").read_bytes())
|
||
rec = svc.get_session(s.session_id)
|
||
# requirements 仅用户上传；template 来自项目配置
|
||
assert svc.has_file(rec, "requirements") is True
|
||
assert svc.has_file(rec, "template") is True
|
||
assert svc.has_file(rec, "write_instruction") is False
|
||
assert svc._eff_path(rec, "template") == str(_SAMPLE / "template_design_ja.docx")
|
||
|
||
|
||
def test_rebuild_source_merges_project_config(tmp_path):
|
||
store = SessionStore(db_path=str(tmp_path / "s.db"))
|
||
projects = ProjectsStore(db_path=str(tmp_path / "p.db"))
|
||
projects.upsert(name="projA", display_name="P", template=str(_SAMPLE / "template_design_ja.docx"),
|
||
write_instruction=str(_SAMPLE / "rules_design_ja.docx"),
|
||
rules=[str(_SAMPLE / "rules_entry_ja.docx")],
|
||
existing_system_code_dir="", design_docs_dir="")
|
||
svc = GenesisService(
|
||
store=store, data_root=str(tmp_path / "data"), engine=FakeEngine(), projects=projects,
|
||
)
|
||
s = svc.create_session("u1", project="projA")
|
||
svc.upload_file(s.session_id, "requirements", "requirements_newdev.xlsx",
|
||
(_SAMPLE / "requirements_newdev.xlsx").read_bytes())
|
||
ss = svc._rebuild_source(svc.get_session(s.session_id))
|
||
assert ss.template is not None
|
||
assert len(ss.rule_docs) >= 2 # write_instruction + rules 合并
|
||
|
||
|
||
def test_rebuild_source_enumerates_design_docs_dir(tmp_path):
|
||
from docx import Document
|
||
store = SessionStore(db_path=str(tmp_path / "s.db"))
|
||
projects = ProjectsStore(db_path=str(tmp_path / "p.db"))
|
||
dd = tmp_path / "design"
|
||
dd.mkdir()
|
||
d = Document()
|
||
d.add_paragraph("设计文档：订单管理整体方案")
|
||
d.save(dd / "d1.docx")
|
||
projects.upsert(name="projA", display_name="P", template="", write_instruction="",
|
||
rules=[], existing_system_code_dir="", design_docs_dir=str(dd))
|
||
svc = GenesisService(
|
||
store=store, data_root=str(tmp_path / "data"), engine=FakeEngine(), projects=projects,
|
||
)
|
||
s = svc.create_session("u1", project="projA")
|
||
svc.upload_file(s.session_id, "requirements", "requirements_newdev.xlsx",
|
||
(_SAMPLE / "requirements_newdev.xlsx").read_bytes())
|
||
ss = svc._rebuild_source(svc.get_session(s.session_id))
|
||
assert len(ss.design_docs) == 1
|
||
|
||
|
||
def test_has_file_without_projects_returns_false(svc):
|
||
s = svc.create_session("u1")
|
||
rec = svc.get_session(s.session_id)
|
||
assert svc.has_file(rec, "template") is False
|
||
assert svc._eff_path(rec, "template") is None
|
||
|
||
|
||
def test_upload_requirements_does_not_override_named_session(tmp_path):
|
||
svc = GenesisService(
|
||
store=SessionStore(db_path=str(tmp_path / "s.db")),
|
||
data_root=str(tmp_path / "data"), engine=FakeEngine(),
|
||
)
|
||
s = svc.create_session("u1", name="已命名会话")
|
||
svc.upload_file(s.session_id, "requirements", "other_name.xlsx", b"PK\x03\x04")
|
||
assert svc.get_session(s.session_id).name == "已命名会话"
|
||
|
||
|
||
def test_upload_invalid_type_rejected(svc):
|
||
s = svc.create_session("u1")
|
||
with pytest.raises(FileTypeError):
|
||
svc.upload_file(s.session_id, "bogus", "x.txt", b"x")
|
||
|
||
|
||
def test_parse_requires_files(svc):
|
||
s = svc.create_session("u1")
|
||
with pytest.raises(ServiceStepError):
|
||
svc.run_parse(s.session_id)
|
||
|
||
|
||
def test_full_flow_fake(tmp_path):
|
||
"""完整闭环：解析 → 确认 → 影响 → 确认 → 生成 → QA → 结果路径就绪。"""
|
||
svc = GenesisService(
|
||
store=SessionStore(db_path=str(tmp_path / "s.db")),
|
||
data_root=str(tmp_path / "data"),
|
||
engine=FakeEngine(),
|
||
)
|
||
s = svc.create_session("u1")
|
||
_upload_core(svc, s.session_id)
|
||
|
||
svc.run_parse(s.session_id)
|
||
assert svc.get_session(s.session_id).status == "awaiting_parse_confirm"
|
||
assert "機能" in svc.get_session(s.session_id).structured_summary or "tables" in svc.get_session(s.session_id).structured_summary
|
||
|
||
svc.confirm_parse(s.session_id)
|
||
assert svc.get_session(s.session_id).status == "writing" # 无既有系统 → 影响跳过 → writing
|
||
|
||
svc.run_generate(s.session_id, output_language="auto")
|
||
got = svc.get_session(s.session_id)
|
||
assert got.status == "writing"
|
||
assert got.result_path and Path(got.result_path).is_file()
|
||
|
||
# 预览可读
|
||
html = svc.result_preview(s.session_id)
|
||
assert "<html" in html or "概要" in html or len(html) > 0
|
||
|
||
|
||
def test_generate_requires_confirmed_parse(svc):
|
||
s = svc.create_session("u1")
|
||
_upload_core(svc, s.session_id)
|
||
with pytest.raises(ServiceStepError):
|
||
svc.run_generate(s.session_id) # 尚未解析确认
|
||
|
||
|
||
def test_run_qa_after_generate(tmp_path):
|
||
svc = GenesisService(
|
||
store=SessionStore(db_path=str(tmp_path / "s.db")),
|
||
data_root=str(tmp_path / "data"),
|
||
engine=FakeEngine(),
|
||
)
|
||
s = svc.create_session("u1")
|
||
_upload_core(svc, s.session_id)
|
||
svc.run_parse(s.session_id)
|
||
svc.confirm_parse(s.session_id)
|
||
svc.run_generate(s.session_id)
|
||
svc.run_qa(s.session_id)
|
||
got = svc.get_session(s.session_id)
|
||
assert got.qa_summary # QA 报告 JSON 已持久化
|
||
assert "language" in got.qa_summary or "overall_score" in got.qa_summary
|
||
|
||
|
||
def _make_existing_zip(tmp_path) -> bytes:
|
||
"""构造含 1 个 Java 文件的 zip（既有系统样本）。"""
|
||
import io
|
||
import zipfile
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w") as zf:
|
||
zf.writestr("demo/OrderController.java",
|
||
"package demo;\n"
|
||
"@RestController public class OrderController {\n"
|
||
" @GetMapping public String list() { return \"ok\"; }\n}\n")
|
||
return buf.getvalue()
|
||
|
||
|
||
def test_existing_system_zip_triggers_impact_flow(tmp_path):
|
||
"""既有系统 zip 上传 → 影响调查全流程（run_impact/confirm_impact）。"""
|
||
svc = GenesisService(
|
||
store=SessionStore(db_path=str(tmp_path / "s.db")),
|
||
data_root=str(tmp_path / "data"),
|
||
engine=FakeEngine(),
|
||
)
|
||
s = svc.create_session("u1")
|
||
_upload_core(svc, s.session_id)
|
||
svc.upload_file(s.session_id, "existing_system", "existing.zip", _make_existing_zip(tmp_path))
|
||
|
||
svc.run_parse(s.session_id)
|
||
svc.confirm_parse(s.session_id)
|
||
assert svc.get_session(s.session_id).status == "impact_running"
|
||
|
||
asyncio.run(svc.run_impact(s.session_id))
|
||
got = svc.get_session(s.session_id)
|
||
assert got.status == "awaiting_impact_confirm"
|
||
assert got.impact_summary
|
||
assert got.impact_report_path and Path(got.impact_report_path).exists()
|
||
|
||
svc.confirm_impact(s.session_id)
|
||
assert svc.get_session(s.session_id).status == "writing"
|
||
|
||
# 从 awaiting_impact_confirm 也能生成（兼容）
|
||
svc.run_generate(s.session_id)
|
||
assert svc.get_session(s.session_id).result_path
|
||
|
||
|
||
def test_impact_state_guard(svc):
|
||
s = svc.create_session("u1")
|
||
_upload_core(svc, s.session_id)
|
||
svc.run_parse(s.session_id)
|
||
svc.confirm_parse(s.session_id)
|
||
# 无既有系统 → writing，start-impact 应报错
|
||
with pytest.raises(ServiceStepError):
|
||
asyncio.run(svc.run_impact(s.session_id))
|
||
|
||
|
||
def test_qa_requires_generated(svc):
|
||
s = svc.create_session("u1")
|
||
_upload_core(svc, s.session_id)
|
||
svc.run_parse(s.session_id)
|
||
svc.confirm_parse(s.session_id)
|
||
with pytest.raises(ServiceStepError):
|
||
svc.run_qa(s.session_id) # 尚未生成
|
||
|
||
|
||
def test_preview_missing_result_raises(svc):
|
||
s = svc.create_session("u1")
|
||
with pytest.raises(ServiceStepError):
|
||
svc.result_preview(s.session_id)
|
||
|
||
|
||
def test_preview_with_table(tmp_path):
|
||
"""预览含表格的 docx → 覆盖表格渲染分支（service 241-244）。"""
|
||
from docx import Document
|
||
from docx.table import Table
|
||
from genesis.server.store import SessionStore
|
||
svc = GenesisService(
|
||
store=SessionStore(db_path=str(tmp_path / "s.db")),
|
||
data_root=str(tmp_path / "data"),
|
||
engine=FakeEngine(),
|
||
)
|
||
s = svc.create_session("u1")
|
||
doc_path = tmp_path / "data" / s.session_id / "with_table.docx"
|
||
doc_path.parent.mkdir(parents=True, exist_ok=True)
|
||
doc = Document()
|
||
doc.add_paragraph("概要段落")
|
||
t = doc.add_table(rows=1, cols=2)
|
||
t.rows[0].cells[0].text = "列A"
|
||
t.rows[0].cells[1].text = "列B"
|
||
doc.save(str(doc_path))
|
||
svc.store.update_session(s.session_id, result_path=str(doc_path))
|
||
html_out = svc.result_preview(s.session_id)
|
||
assert "<table" in html_out
|
||
assert "列A" in html_out
|
||
|
||
|
||
def test_confirm_parse_wrong_state_raises(svc):
|
||
s = svc.create_session("u1")
|
||
_upload_core(svc, s.session_id)
|
||
# uploading 态直接确认解析 → 报错（service 139）
|
||
with pytest.raises(ServiceStepError):
|
||
svc.confirm_parse(s.session_id)
|
||
|
||
|
||
def test_confirm_impact_wrong_state_raises(tmp_path):
|
||
svc = GenesisService(
|
||
store=SessionStore(db_path=str(tmp_path / "s.db")),
|
||
data_root=str(tmp_path / "data"),
|
||
engine=FakeEngine(),
|
||
)
|
||
s = svc.create_session("u1")
|
||
_upload_core(svc, s.session_id)
|
||
svc.run_parse(s.session_id)
|
||
svc.confirm_parse(s.session_id)
|
||
# writing 态直接确认影响 → 报错（service 167）
|
||
with pytest.raises(ServiceStepError):
|
||
svc.confirm_impact(s.session_id)
|