lhl
916c5beed7
- 会话支持 name/project 字段，上传要件定义后自动命名；前端侧边栏会话历史 + localStorage 恢复，顶部只显示会话名 - 新增 ProjectsStore（SQLite）与 /api/projects CRUD；绑定项目后 _rebuild_source 合并模板/规则/代码库/设计文档，上传区仅要件定义 - StructuredSource.design_docs 与 ImpactReport.design_references；影响调查新增既有设计文档确定性交叉引用（无 LLM） - 同步更新 docs/design.md §12.7、README、_AI_USAGE_LOG.md；全量测试 558 通过，覆盖率 99.10%
536 lines
22 KiB
Python
536 lines
22 KiB
Python
"""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_status_reply_with_project_design_docs(tmp_path):
|
|
from genesis.server.store import ProjectsStore
|
|
store = SessionStore(db_path=str(tmp_path / "s.db"))
|
|
dd = tmp_path / "design"
|
|
dd.mkdir()
|
|
(dd / "d.docx").write_bytes(b"PK\x03\x04")
|
|
projects = ProjectsStore(db_path=str(tmp_path / "p.db"))
|
|
projects.upsert(name="projA", display_name="P", template="", write_instruction="",
|
|
rules=[], existing_system_code_dir="", design_docs_dir=str(dd))
|
|
agent = ChatAgent(GenesisService(store=store, data_root=str(tmp_path / "data"),
|
|
engine=FakeEngine(), projects=projects), fake=True)
|
|
sid = agent.service.create_session("u1", project="projA").session_id
|
|
r = agent.handle_message(sid, "现在什么状态？")
|
|
assert "design_docs" in r["reply"]
|
|
|
|
|
|
def test_status_reply_lists_existing_system_and_design_docs(tmp_path):
|
|
from genesis.server.store import ProjectsStore
|
|
store = SessionStore(db_path=str(tmp_path / "s.db"))
|
|
dd = tmp_path / "design"; dd.mkdir(); (dd / "d.docx").write_bytes(b"PK\x03\x04")
|
|
code = tmp_path / "code"; code.mkdir()
|
|
projects = ProjectsStore(db_path=str(tmp_path / "p.db"))
|
|
projects.upsert(name="projA", display_name="P", template="", write_instruction="",
|
|
rules=[], existing_system_code_dir=str(code), design_docs_dir=str(dd))
|
|
agent = ChatAgent(GenesisService(store=store, data_root=str(tmp_path / "data"),
|
|
engine=FakeEngine(), projects=projects), fake=True)
|
|
sid = agent.service.create_session("u1", project="projA").session_id
|
|
reply = agent._status_reply(agent.service.get_session(sid), [])
|
|
assert "existing_system" in reply["reply"] and "design_docs" in reply["reply"]
|
|
|
|
|
|
def test_auto_generate_unknown_status_falls_to_status_reply(tmp_path):
|
|
from genesis.server.store import ProjectsStore
|
|
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="")
|
|
agent = ChatAgent(GenesisService(store=store, data_root=str(tmp_path / "data"),
|
|
engine=FakeEngine(), projects=projects), fake=True)
|
|
sid = agent.service.create_session("u1", project="projA").session_id
|
|
# 上传要件定义，使 has_file 通过；随后将状态置为 done 触发 _status_reply 分支
|
|
agent.service.upload_file(sid, "requirements", "requirements_newdev.xlsx",
|
|
(_SAMPLE / "requirements_newdev.xlsx").read_bytes())
|
|
agent.service.store.update_status(sid, "done")
|
|
r = agent._auto_generate(sid, "auto", [])
|
|
assert "done" in r["reply"]
|
|
|
|
|
|
def test_run_generate_from_awaiting_impact_confirm_confirms_first(tmp_path):
|
|
agent = ChatAgent(_svc(tmp_path), fake=True)
|
|
sid = agent.service.create_session("u1").session_id
|
|
agent.service.store.update_status(sid, "awaiting_impact_confirm")
|
|
rec = agent.service.get_session(sid)
|
|
r = agent._run_generate(sid, rec, [], output_language="auto")
|
|
# confirm_impact 先执行（line 194），随后 generate 因缺文件抛错被捕获
|
|
assert r["status"] in ("writing", "awaiting_impact_confirm")
|
|
|
|
|
|
def test_run_impact_from_awaiting_parse_confirm(tmp_path):
|
|
from genesis.server.store import ProjectsStore
|
|
store = SessionStore(db_path=str(tmp_path / "s.db"))
|
|
code = tmp_path / "code"; code.mkdir(); (code / "A.java").write_text("class A {}")
|
|
projects = ProjectsStore(db_path=str(tmp_path / "p.db"))
|
|
projects.upsert(name="projA", display_name="P", template="", write_instruction="",
|
|
rules=[], existing_system_code_dir=str(code), design_docs_dir="")
|
|
agent = ChatAgent(GenesisService(store=store, data_root=str(tmp_path / "data"),
|
|
engine=FakeEngine(), projects=projects), fake=True)
|
|
sid = agent.service.create_session("u1", project="projA").session_id
|
|
agent.service.store.update_status(sid, "awaiting_parse_confirm")
|
|
r = agent._run_impact(sid, [])
|
|
assert r["status"] == "awaiting_impact_confirm"
|
|
|
|
|
|
def test_impact_brief_malformed_returns_default(tmp_path):
|
|
agent = ChatAgent(_svc(tmp_path), fake=True)
|
|
rec = agent.service.create_session("u1")
|
|
rec.impact_summary = "not-json"
|
|
assert agent._impact_brief(rec) == "已完成"
|
|
|
|
|
|
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"]
|