聊天式交互改造（Web UI 升级）
- 前端由分步表单页升级为 DeepSeek 式聊天页（static/chat.html）
- 新增 chat 包：意图识别（intent.py）+ ChatAgent（agent.py）自动驱动
解析→影响→生成→QA 整条工作流，影响确认节点反问、错误分支兜底
- store.py 扩展 pending_intent 字段与 chat_messages 表
- app.py 新增 POST/GET /api/chat/{sid}/messages，GET / 返回聊天页
- README/design.md 同步聊天模式说明；测试 4 文件覆盖
- 全量 pytest 522 passed / 99.03%，覆盖率红线达标
This commit is contained in:
@@ -0,0 +1,454 @@
|
||||
"""C3：聊天 Agent 测试（chat/agent.py）——确认反问/自动推进。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from genesis.chat.agent import ChatAgent
|
||||
from genesis.server.service import GenesisService
|
||||
from genesis.server.store import SessionStore
|
||||
|
||||
_SAMPLE = Path(__file__).resolve().parents[1] / "sample"
|
||||
|
||||
|
||||
class FakeEngine:
|
||||
def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2):
|
||||
from types import SimpleNamespace
|
||||
title = variables.get("title", "x")
|
||||
return SimpleNamespace(
|
||||
data={"title": title,
|
||||
"blocks": [{"type": "paragraph",
|
||||
"text": "本機能はFakeLLMにより生成された十分な説明内容であり、書込規則を満たす。"}]},
|
||||
status="ok",
|
||||
)
|
||||
|
||||
|
||||
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(agent, sid):
|
||||
agent.service.upload_file(sid, "requirements", "requirements_newdev.xlsx",
|
||||
(_SAMPLE / "requirements_newdev.xlsx").read_bytes())
|
||||
agent.service.upload_file(sid, "template", "template_design_ja.docx",
|
||||
(_SAMPLE / "template_design_ja.docx").read_bytes())
|
||||
agent.service.upload_file(sid, "write_instruction", "rules_design_ja.docx",
|
||||
(_SAMPLE / "rules_design_ja.docx").read_bytes())
|
||||
agent.service.upload_file(sid, "rules", "rules_entry_ja.docx",
|
||||
(_SAMPLE / "rules_entry_ja.docx").read_bytes())
|
||||
|
||||
|
||||
def test_generate_full_flow_without_existing_system(tmp_path):
|
||||
"""无既有系统：一条"生成"消息 → 全流程完成（无确认打断）。"""
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
_upload_core(agent, sid)
|
||||
result = agent.handle_message(sid, "请生成概要设计书")
|
||||
assert result["status"] == "done"
|
||||
assert any(p["step"] == "generate" and p["status"] == "ok" for p in result["progress"])
|
||||
assert any(p["step"] == "qa" for p in result["progress"])
|
||||
|
||||
|
||||
def test_generate_with_existing_system_asks_confirmation(tmp_path):
|
||||
"""有既有系统：影响调查完成后反问确认（awaiting_impact_confirm），确认后继续。"""
|
||||
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")
|
||||
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
_upload_core(agent, sid)
|
||||
agent.service.upload_file(sid, "existing_system", "existing.zip", buf.getvalue())
|
||||
|
||||
r1 = agent.handle_message(sid, "请生成概要设计书")
|
||||
assert r1["status"] == "awaiting_impact_confirm"
|
||||
assert any(p["step"] == "impact" for p in r1["progress"])
|
||||
assert "确认" in r1["reply"]
|
||||
|
||||
r2 = agent.handle_message(sid, "确认，继续")
|
||||
assert r2["status"] == "done"
|
||||
steps = [p["step"] for p in r2["progress"]]
|
||||
assert "generate" in steps and "qa" in steps
|
||||
|
||||
|
||||
def test_confirm_without_pending_says_ok(tmp_path):
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
r = agent.handle_message(sid, "确认")
|
||||
assert r["status"] == "uploading"
|
||||
assert "确认" in r["reply"]
|
||||
|
||||
|
||||
def test_status_reports_current_state(tmp_path):
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
r = agent.handle_message(sid, "现在什么状态？")
|
||||
assert r["status"] == "uploading"
|
||||
assert "uploading" in r["reply"]
|
||||
|
||||
|
||||
def test_generate_without_files_hints_upload(tmp_path):
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
r = agent.handle_message(sid, "生成概要设计书")
|
||||
assert r["status"] == "uploading"
|
||||
assert "上传" in r["reply"]
|
||||
|
||||
|
||||
def test_messages_persisted(tmp_path):
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
agent.handle_message(sid, "现在什么状态？")
|
||||
msgs = agent.store.list_messages(sid)
|
||||
assert msgs[0]["role"] == "user"
|
||||
assert msgs[-1]["role"] == "assistant"
|
||||
|
||||
|
||||
def test_reject_impact_returns_to_impact_running(tmp_path):
|
||||
"""影响确认节点：回复「打回」→ 回到 impact_running。"""
|
||||
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")
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
_upload_core(agent, sid)
|
||||
agent.service.upload_file(sid, "existing_system", "e.zip", buf.getvalue())
|
||||
r1 = agent.handle_message(sid, "生成概要设计书")
|
||||
assert r1["status"] == "awaiting_impact_confirm"
|
||||
r2 = agent.handle_message(sid, "打回，重新影响调查")
|
||||
assert r2["status"] == "impact_running"
|
||||
|
||||
|
||||
def test_confirm_at_impact_node_continues_generate(tmp_path):
|
||||
"""影响节点单独确认（无 pending）→ 状态 writing。"""
|
||||
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")
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
_upload_core(agent, sid)
|
||||
agent.service.upload_file(sid, "existing_system", "e.zip", buf.getvalue())
|
||||
r1 = agent.handle_message(sid, "做影响调查")
|
||||
assert r1["status"] == "awaiting_impact_confirm"
|
||||
r2 = agent.handle_message(sid, "确认")
|
||||
assert r2["status"] == "writing"
|
||||
|
||||
|
||||
def test_impact_action_without_zip_hints_generate(tmp_path):
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
r = agent.handle_message(sid, "做影响调查")
|
||||
assert "跳过" in r["reply"]
|
||||
|
||||
|
||||
def test_qa_without_result_hints_generate(tmp_path):
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
r = agent.handle_message(sid, "运行QA校验")
|
||||
assert "尚未生成" in r["reply"]
|
||||
|
||||
|
||||
def test_help_for_unknown_message(tmp_path):
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
r = agent.handle_message(sid, "今天天气怎么样")
|
||||
assert r["status"] == "uploading"
|
||||
assert "生成概要设计书" in r["reply"]
|
||||
|
||||
|
||||
class IntentLLMEngine:
|
||||
"""模拟真实 LLM 意图解析：返回指定 action。"""
|
||||
|
||||
def __init__(self, action, params=None):
|
||||
self.action = action
|
||||
self.params = params or {}
|
||||
self.calls = 0
|
||||
|
||||
def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2):
|
||||
self.calls += 1
|
||||
from types import SimpleNamespace
|
||||
return SimpleNamespace(data={"action": self.action, "params": self.params},
|
||||
status="ok")
|
||||
|
||||
|
||||
def test_llm_intent_mode_generate(tmp_path):
|
||||
"""真实模式（非 fake）：LLM 判定 generate → 自动推进完成。"""
|
||||
eng = IntentLLMEngine("generate")
|
||||
agent = ChatAgent(_svc(tmp_path), fake=False, engine=eng)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
_upload_core(agent, sid)
|
||||
r = agent.handle_message(sid, "帮我生成设计书")
|
||||
assert r["status"] == "done"
|
||||
assert eng.calls >= 1
|
||||
|
||||
|
||||
def test_llm_intent_mode_unknown_falls_back(tmp_path):
|
||||
eng = IntentLLMEngine("unknown")
|
||||
agent = ChatAgent(_svc(tmp_path), fake=False, engine=eng)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
r = agent.handle_message(sid, "随便聊聊")
|
||||
assert r["status"] == "uploading"
|
||||
|
||||
|
||||
def test_generate_failure_reported(tmp_path):
|
||||
"""生成失败 → 回复失败信息（不崩溃）。"""
|
||||
from genesis.server.service import ServiceStepError
|
||||
|
||||
svc = _svc(tmp_path)
|
||||
# 直接注入一个会在 run_generate 抛错的子类
|
||||
class BadGenerateService(type(svc)):
|
||||
def run_generate(self, session_id, output_language="auto"):
|
||||
raise ServiceStepError("生成内部错误")
|
||||
|
||||
agent = ChatAgent(BadGenerateService(svc.store, data_root=str(svc.data_root),
|
||||
engine=FakeEngine()), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
_upload_core(agent, sid)
|
||||
# 先正常解析到 writing
|
||||
agent.service.run_parse(sid)
|
||||
agent.service.confirm_parse(sid)
|
||||
r = agent.handle_message(sid, "生成概要设计书")
|
||||
assert "生成失败" in r["reply"]
|
||||
|
||||
|
||||
def test_explicit_parse_action(tmp_path):
|
||||
"""自然语言「开始解析」→ 解析完成到 awaiting。"""
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
_upload_core(agent, sid)
|
||||
r = agent.handle_message(sid, "开始解析")
|
||||
assert r["status"] in ("awaiting_impact_confirm", "uploading", "writing", "awaiting_parse_confirm")
|
||||
|
||||
|
||||
def test_parse_without_files_hints_upload(tmp_path):
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
r = agent.handle_message(sid, "开始解析")
|
||||
assert "要件定义" in r["reply"] or "上传" in r["reply"]
|
||||
|
||||
|
||||
def test_parse_failure_reported(tmp_path):
|
||||
"""解析失败 → 回复失败信息。"""
|
||||
from genesis.server.service import ServiceStepError
|
||||
svc = _svc(tmp_path)
|
||||
|
||||
class BadParseService(type(svc)):
|
||||
def run_parse(self, session_id):
|
||||
raise ServiceStepError("解析内部错误")
|
||||
|
||||
agent = ChatAgent(BadParseService(svc.store, data_root=str(svc.data_root),
|
||||
engine=FakeEngine()), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
_upload_core(agent, sid)
|
||||
r = agent.handle_message(sid, "生成概要设计书")
|
||||
assert "解析失败" in r["reply"]
|
||||
|
||||
|
||||
def test_impact_confirm_node_unknown_reply(tmp_path):
|
||||
"""影响确认节点：非确认/打回的消息 → 提示再次确认。"""
|
||||
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")
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
_upload_core(agent, sid)
|
||||
agent.service.upload_file(sid, "existing_system", "e.zip", buf.getvalue())
|
||||
agent.handle_message(sid, "生成概要设计书") # -> awaiting_impact_confirm
|
||||
r = agent.handle_message(sid, "这个影响范围对吗")
|
||||
assert "确认" in r["reply"]
|
||||
|
||||
|
||||
def test_qa_runs_after_generate(tmp_path):
|
||||
"""生成后运行 QA → qa 完成。"""
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
_upload_core(agent, sid)
|
||||
agent.handle_message(sid, "生成概要设计书")
|
||||
r = agent.handle_message(sid, "运行QA校验")
|
||||
assert r["status"] == "done"
|
||||
assert "QA 完成" in r["reply"]
|
||||
|
||||
|
||||
def test_qa_failure_reported(tmp_path):
|
||||
"""QA 失败 → warn 提示但不崩溃。"""
|
||||
from genesis.server.service import ServiceStepError
|
||||
svc = _svc(tmp_path)
|
||||
|
||||
class BadQaService(type(svc)):
|
||||
def run_qa(self, session_id):
|
||||
raise ServiceStepError("QA 内部错误")
|
||||
|
||||
agent = ChatAgent(BadQaService(svc.store, data_root=str(svc.data_root),
|
||||
engine=FakeEngine()), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
_upload_core(agent, sid)
|
||||
agent.handle_message(sid, "生成概要设计书")
|
||||
r = agent.handle_message(sid, "运行QA校验")
|
||||
assert "QA 未执行" in r["reply"]
|
||||
|
||||
|
||||
def test_status_reply_done_mentions_download(tmp_path):
|
||||
"""done 状态回复包含下载提示。"""
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
_upload_core(agent, sid)
|
||||
agent.handle_message(sid, "生成概要设计书")
|
||||
r = agent.handle_message(sid, "现在什么状态？")
|
||||
assert "下载" in r["reply"]
|
||||
|
||||
|
||||
def _zip():
|
||||
import io
|
||||
import zipfile
|
||||
buf = io.BytesIO()
|
||||
with zipfile.ZipFile(buf, "w") as zf:
|
||||
zf.writestr("demo/X.java", "package demo; public class X {}")
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
def test_explicit_parse_when_already_done(tmp_path):
|
||||
"""已完成后「开始解析」→ 返回状态回复。"""
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
_upload_core(agent, sid)
|
||||
agent.handle_message(sid, "生成概要设计书")
|
||||
r = agent.handle_message(sid, "开始解析")
|
||||
assert r["status"] in ("writing", "done", "awaiting_parse_confirm")
|
||||
|
||||
|
||||
def test_explicit_parse_failure_reported(tmp_path):
|
||||
"""显式「开始解析」时解析异常 → 回复解析失败。"""
|
||||
from genesis.server.service import ServiceStepError
|
||||
svc = _svc(tmp_path)
|
||||
|
||||
class BadParse(type(svc)):
|
||||
def run_parse(self, session_id):
|
||||
raise ServiceStepError("解析错误")
|
||||
|
||||
agent = ChatAgent(BadParse(svc.store, data_root=str(svc.data_root),
|
||||
engine=FakeEngine()), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
_upload_core(agent, sid)
|
||||
r = agent.handle_message(sid, "开始解析")
|
||||
assert "解析失败" in r["reply"]
|
||||
|
||||
|
||||
def test_explicit_parse_with_zip_asks_confirm(tmp_path):
|
||||
"""解析（带 zip）完成 → 影响确认反问。"""
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
_upload_core(agent, sid)
|
||||
agent.service.upload_file(sid, "existing_system", "e.zip", _zip())
|
||||
r = agent.handle_message(sid, "开始解析")
|
||||
assert r["status"] == "awaiting_impact_confirm"
|
||||
|
||||
|
||||
def test_impact_with_parse_failure(tmp_path):
|
||||
"""「做影响调查」触发解析异常 → 回复解析失败。"""
|
||||
from genesis.server.service import ServiceStepError
|
||||
svc = _svc(tmp_path)
|
||||
|
||||
class BadParse(type(svc)):
|
||||
def run_parse(self, session_id):
|
||||
raise ServiceStepError("解析错误")
|
||||
|
||||
agent = ChatAgent(BadParse(svc.store, data_root=str(svc.data_root),
|
||||
engine=FakeEngine()), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
_upload_core(agent, sid)
|
||||
agent.service.upload_file(sid, "existing_system", "e.zip", _zip())
|
||||
r = agent.handle_message(sid, "做影响调查")
|
||||
assert "解析失败" in r["reply"]
|
||||
|
||||
|
||||
def test_impact_after_awaiting_parse_confirm(tmp_path):
|
||||
"""已解析待确认状态「做影响调查」→ 确认解析后进入影响确认。"""
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
_upload_core(agent, sid)
|
||||
agent.service.upload_file(sid, "existing_system", "e.zip", _zip())
|
||||
agent.handle_message(sid, "开始解析") # -> awaiting_parse_confirm
|
||||
r = agent.handle_message(sid, "做影响调查")
|
||||
assert r["status"] == "awaiting_impact_confirm"
|
||||
|
||||
|
||||
def test_impact_when_already_writing_refused(tmp_path):
|
||||
"""已生成（writing）状态再「做影响调查」→ 提示无法。"""
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
_upload_core(agent, sid)
|
||||
agent.service.upload_file(sid, "existing_system", "e.zip", _zip())
|
||||
# 手动将状态置于 writing（已生成、不再处于影响环节）
|
||||
agent.store.update_status(sid, "writing")
|
||||
r = agent.handle_message(sid, "做影响调查")
|
||||
assert "无法" in r["reply"]
|
||||
|
||||
|
||||
def test_run_impact_from_impact_running_state(tmp_path):
|
||||
"""状态置于 impact_running 时「做影响调查」→ 直接执行影响调查。"""
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
_upload_core(agent, sid)
|
||||
agent.service.upload_file(sid, "existing_system", "e.zip", _zip())
|
||||
# 模拟已进入影响环节但尚未执行
|
||||
agent.store.update_status(sid, "impact_running")
|
||||
r = agent.handle_message(sid, "做影响调查")
|
||||
assert r["status"] == "awaiting_impact_confirm"
|
||||
|
||||
|
||||
def test_confirm_impact_then_generate(tmp_path):
|
||||
"""影响确认节点「确认，继续」→ 自动生成完成。"""
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
_upload_core(agent, sid)
|
||||
agent.service.upload_file(sid, "existing_system", "e.zip", _zip())
|
||||
agent.handle_message(sid, "生成概要设计书") # -> awaiting_impact_confirm
|
||||
r = agent.handle_message(sid, "确认，继续")
|
||||
assert r["status"] == "done"
|
||||
|
||||
|
||||
def test_confirm_impact_failure_reported(tmp_path):
|
||||
"""影响确认异常 → 回复失败信息。"""
|
||||
from genesis.server.service import ServiceStepError
|
||||
svc = _svc(tmp_path)
|
||||
|
||||
class BadImpact(type(svc)):
|
||||
def confirm_impact(self, session_id):
|
||||
raise ServiceStepError("影响确认错误")
|
||||
|
||||
agent = ChatAgent(BadImpact(svc.store, data_root=str(svc.data_root),
|
||||
engine=FakeEngine()), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
_upload_core(agent, sid)
|
||||
agent.service.upload_file(sid, "existing_system", "e.zip", _zip())
|
||||
agent.handle_message(sid, "生成概要设计书") # awaiting_impact_confirm
|
||||
r = agent.handle_message(sid, "确认，继续")
|
||||
assert "影响确认失败" in r["reply"] or "失败" in r["reply"]
|
||||
|
||||
|
||||
def test_status_reply_result_without_impact(tmp_path):
|
||||
"""结果存在但无影响摘要 → 状态回复不崩溃。"""
|
||||
agent = ChatAgent(_svc(tmp_path), fake=True)
|
||||
sid = agent.service.create_session("u1").session_id
|
||||
_upload_core(agent, sid)
|
||||
agent.handle_message(sid, "生成概要设计书")
|
||||
agent.store.update_session(sid, impact_summary=None)
|
||||
r = agent.handle_message(sid, "现在什么状态？")
|
||||
assert "下载" in r["reply"]
|
||||
@@ -0,0 +1,66 @@
|
||||
"""C2：意图解析测试（chat/intent.py）。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from genesis.chat.intent import INTENT_SCHEMA, parse_intent_fake, Intent
|
||||
|
||||
|
||||
def test_schema_defines_actions():
|
||||
enums = INTENT_SCHEMA["properties"]["action"]["enum"]
|
||||
assert {"parse", "impact", "generate", "qa", "status", "confirm", "reject", "unknown"} <= set(enums)
|
||||
|
||||
|
||||
def test_parse_generate_keywords():
|
||||
assert parse_intent_fake("请生成概要设计书").action == "generate"
|
||||
assert parse_intent_fake("开始生成吧").action == "generate"
|
||||
assert parse_intent_fake("生成中文设计书").params.get("output_language") == "zh"
|
||||
|
||||
|
||||
def test_parse_parse_and_impact():
|
||||
assert parse_intent_fake("开始解析").action == "parse"
|
||||
assert parse_intent_fake("做影响调查").action == "impact"
|
||||
|
||||
|
||||
def test_parse_qa_status():
|
||||
assert parse_intent_fake("运行QA校验").action == "qa"
|
||||
assert parse_intent_fake("现在什么状态").action == "status"
|
||||
|
||||
|
||||
def test_parse_confirm_reject():
|
||||
assert parse_intent_fake("确认").action == "confirm"
|
||||
assert parse_intent_fake("可以，继续").action == "confirm"
|
||||
assert parse_intent_fake("打回，重新影响调查").action == "reject"
|
||||
|
||||
|
||||
def test_parse_unknown():
|
||||
it = parse_intent_fake("今天天气怎么样")
|
||||
assert it.action == "unknown"
|
||||
|
||||
|
||||
def test_intent_language_detection_zh_ja():
|
||||
assert parse_intent_fake("用中文生成").params.get("output_language") == "zh"
|
||||
assert parse_intent_fake("日本語で生成して").params.get("output_language") == "ja"
|
||||
|
||||
|
||||
class _Engine:
|
||||
def __init__(self, data, status="ok"):
|
||||
self.data = data
|
||||
self.status = status
|
||||
|
||||
def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2):
|
||||
from types import SimpleNamespace
|
||||
return SimpleNamespace(data=self.data, status=self.status)
|
||||
|
||||
|
||||
def test_parse_intent_llm_ok():
|
||||
from genesis.chat.intent import parse_intent_llm
|
||||
eng = _Engine({"action": "generate", "params": {"output_language": "zh"}})
|
||||
it = parse_intent_llm(eng, "s1", "生成", "context")
|
||||
assert it.action == "generate"
|
||||
assert it.params.get("output_language") == "zh"
|
||||
|
||||
|
||||
def test_parse_intent_llm_failed_falls_to_unknown():
|
||||
from genesis.chat.intent import parse_intent_llm
|
||||
eng = _Engine({}, status="failed")
|
||||
it = parse_intent_llm(eng, "s1", "生成", "context")
|
||||
assert it.action == "unknown"
|
||||
@@ -0,0 +1,97 @@
|
||||
"""C4：聊天 API 端点测试（app.py chat 端点 + GET / 指向聊天页）。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from genesis.server.app import create_app
|
||||
from genesis.server.store import SessionStore
|
||||
|
||||
_SAMPLE = Path(__file__).resolve().parents[1] / "sample"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(tmp_path):
|
||||
app = create_app(
|
||||
store=SessionStore(db_path=str(tmp_path / "s.db")),
|
||||
data_root=str(tmp_path / "data"),
|
||||
engine="fake",
|
||||
)
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
def _upload_core(client, sid):
|
||||
files = [
|
||||
("requirements", "requirements_newdev.xlsx", _SAMPLE / "requirements_newdev.xlsx"),
|
||||
("template", "template_design_ja.docx", _SAMPLE / "template_design_ja.docx"),
|
||||
("write_instruction", "rules_design_ja.docx", _SAMPLE / "rules_design_ja.docx"),
|
||||
("rules", "rules_entry_ja.docx", _SAMPLE / "rules_entry_ja.docx"),
|
||||
]
|
||||
for ft, name, path in files:
|
||||
r = client.post(f"/api/sessions/{sid}/files",
|
||||
data={"file_type": ft},
|
||||
files={"file": (name, path.read_bytes())})
|
||||
assert r.status_code == 200, r.text
|
||||
|
||||
|
||||
def test_root_serves_chat_page(client):
|
||||
r = client.get("/")
|
||||
assert r.status_code == 200
|
||||
assert "chat" in r.text.lower() or "message" in r.text.lower()
|
||||
|
||||
|
||||
def test_chat_message_flow_generate(client):
|
||||
"""聊天消息：上传后「生成概要设计书」→ 全流程完成。"""
|
||||
sid = client.post("/api/sessions", json={"user_id": "u1"}).json()["session_id"]
|
||||
_upload_core(client, sid)
|
||||
r = client.post(f"/api/chat/{sid}/messages", json={"content": "请生成概要设计书"})
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body["status"] == "done"
|
||||
assert any(p["step"] == "generate" for p in body["progress"])
|
||||
|
||||
|
||||
def test_chat_message_asks_confirmation_with_zip(tmp_path):
|
||||
"""有既有系统 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")
|
||||
|
||||
app = create_app(store=SessionStore(db_path=str(tmp_path / "s.db")),
|
||||
data_root=str(tmp_path / "data"), engine="fake")
|
||||
client = TestClient(app)
|
||||
sid = client.post("/api/sessions", json={"user_id": "u1"}).json()["session_id"]
|
||||
_upload_core(client, sid)
|
||||
client.post(f"/api/sessions/{sid}/files", data={"file_type": "existing_system"},
|
||||
files={"file": ("e.zip", buf.getvalue())})
|
||||
|
||||
r1 = client.post(f"/api/chat/{sid}/messages", json={"content": "生成概要设计书"})
|
||||
assert r1.json()["status"] == "awaiting_impact_confirm"
|
||||
|
||||
r2 = client.post(f"/api/chat/{sid}/messages", json={"content": "确认，继续"})
|
||||
assert r2.json()["status"] == "done"
|
||||
|
||||
|
||||
def test_chat_history_endpoint(client):
|
||||
sid = client.post("/api/sessions", json={"user_id": "u1"}).json()["session_id"]
|
||||
client.post(f"/api/chat/{sid}/messages", json={"content": "现在什么状态？"})
|
||||
r = client.get(f"/api/chat/{sid}/messages")
|
||||
assert r.status_code == 200
|
||||
msgs = r.json()
|
||||
assert len(msgs) >= 2
|
||||
assert msgs[0]["role"] == "user"
|
||||
|
||||
|
||||
def test_chat_message_unknown_session_404(client):
|
||||
r = client.post("/api/chat/nope/messages", json={"content": "hi"})
|
||||
assert r.status_code == 404
|
||||
|
||||
|
||||
def test_chat_history_unknown_session_404(client):
|
||||
r = client.get("/api/chat/nope/messages")
|
||||
assert r.status_code == 404
|
||||
@@ -0,0 +1,38 @@
|
||||
"""C1：聊天消息存储 + pending_intent 测试（store.py 扩展）。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from genesis.server.store import SessionStore
|
||||
|
||||
|
||||
def test_add_and_list_messages(tmp_path):
|
||||
store = SessionStore(db_path=str(tmp_path / "s.db"))
|
||||
sid = store.create_session("u1").session_id
|
||||
store.add_message(sid, "user", "生成概要设计书", action="generate")
|
||||
store.add_message(sid, "assistant", "影响调查完成，请确认", action=None)
|
||||
msgs = store.list_messages(sid)
|
||||
assert len(msgs) == 2
|
||||
assert msgs[0]["role"] == "user"
|
||||
assert msgs[0]["content"] == "生成概要设计书"
|
||||
assert msgs[0]["action"] == "generate"
|
||||
assert msgs[1]["role"] == "assistant"
|
||||
|
||||
|
||||
def test_messages_isolated_per_session(tmp_path):
|
||||
store = SessionStore(db_path=str(tmp_path / "s.db"))
|
||||
a = store.create_session("u1").session_id
|
||||
b = store.create_session("u1").session_id
|
||||
store.add_message(a, "user", "msg-a")
|
||||
store.add_message(b, "user", "msg-b")
|
||||
assert [m["content"] for m in store.list_messages(a)] == ["msg-a"]
|
||||
assert [m["content"] for m in store.list_messages(b)] == ["msg-b"]
|
||||
|
||||
|
||||
def test_pending_intent_persisted(tmp_path):
|
||||
store = SessionStore(db_path=str(tmp_path / "s.db"))
|
||||
sid = store.create_session("u1").session_id
|
||||
assert store.get_session(sid).pending_intent == ""
|
||||
store.update_session(sid, pending_intent="generate")
|
||||
assert store.get_session(sid).pending_intent == "generate"
|
||||
# 重开 db 仍在
|
||||
store2 = SessionStore(db_path=str(tmp_path / "s.db"))
|
||||
assert store2.get_session(sid).pending_intent == "generate"
|
||||
Reference in New Issue
Block a user