lhl
838a93720e
- 前端由分步表单页升级为 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%，覆盖率红线达标
98 lines
3.5 KiB
Python
98 lines
3.5 KiB
Python
"""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
|