359 lines
14 KiB
Python
359 lines
14 KiB
Python
"""S4：FastAPI 端点测试（server/app.py，httpx TestClient 全链路）。"""
|
||
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_health(client):
|
||
r = client.get("/api/health")
|
||
assert r.status_code == 200
|
||
assert r.json()["status"] == "ok"
|
||
|
||
|
||
def test_create_and_get_session(client):
|
||
r = client.post("/api/sessions", json={"user_id": "u1"})
|
||
assert r.status_code == 200
|
||
sid = r.json()["session_id"]
|
||
g = client.get(f"/api/sessions/{sid}")
|
||
assert g.status_code == 200
|
||
assert g.json()["status"] == "uploading"
|
||
|
||
|
||
def test_session_list(client):
|
||
client.post("/api/sessions", json={"user_id": "u1"})
|
||
client.post("/api/sessions", json={"user_id": "u1"})
|
||
r = client.get("/api/sessions", params={"user_id": "u1"})
|
||
assert r.status_code == 200
|
||
assert len(r.json()) == 2
|
||
|
||
|
||
def test_full_flow_http(tmp_path):
|
||
"""HTTP 全链路：建会话 → 上传 → 解析 → 确认 → 生成 → QA → 预览/下载。"""
|
||
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)
|
||
|
||
r = client.post(f"/api/sessions/{sid}/start-parse")
|
||
assert r.status_code == 200
|
||
assert client.get(f"/api/sessions/{sid}").json()["status"] == "awaiting_parse_confirm"
|
||
|
||
pr = client.get(f"/api/sessions/{sid}/parse-result")
|
||
assert pr.status_code == 200
|
||
assert "tables" in pr.json()
|
||
|
||
r = client.post(f"/api/sessions/{sid}/confirm-parse")
|
||
assert r.status_code == 200
|
||
assert client.get(f"/api/sessions/{sid}").json()["status"] == "writing" # 无既有系统 → 影响跳过
|
||
|
||
r = client.post(f"/api/sessions/{sid}/generate", json={"output_language": "auto"})
|
||
assert r.status_code == 200
|
||
assert client.get(f"/api/sessions/{sid}").json()["result_path"]
|
||
|
||
r = client.post(f"/api/sessions/{sid}/run-qa")
|
||
assert r.status_code == 200
|
||
got = client.get(f"/api/sessions/{sid}").json()
|
||
assert got["status"] == "done"
|
||
assert "overall_score" in got["qa_summary"]
|
||
|
||
prev = client.get(f"/api/sessions/{sid}/result/preview")
|
||
assert prev.status_code == 200
|
||
assert "html" in prev.json()
|
||
|
||
dl = client.get(f"/api/sessions/{sid}/result/download")
|
||
assert dl.status_code == 200
|
||
assert dl.headers["content-type"].startswith("application/vnd.openxmlformats")
|
||
|
||
|
||
def test_state_transition_invalid_409(client):
|
||
sid = client.post("/api/sessions", json={"user_id": "u1"}).json()["session_id"]
|
||
# uploading 直接 generate → 409
|
||
r = client.post(f"/api/sessions/{sid}/generate", json={})
|
||
assert r.status_code == 409
|
||
|
||
|
||
def test_upload_invalid_type_400(client):
|
||
sid = client.post("/api/sessions", json={"user_id": "u1"}).json()["session_id"]
|
||
r = client.post(f"/api/sessions/{sid}/files",
|
||
data={"file_type": "bogus"},
|
||
files={"file": ("x.txt", b"x")})
|
||
assert r.status_code == 400
|
||
|
||
|
||
def test_root_serves_frontend(client):
|
||
r = client.get("/")
|
||
assert r.status_code == 200
|
||
assert "Genesis" in r.text
|
||
|
||
|
||
def test_delete_session(client):
|
||
sid = client.post("/api/sessions", json={"user_id": "u1"}).json()["session_id"]
|
||
r = client.delete(f"/api/sessions/{sid}")
|
||
assert r.status_code == 200
|
||
assert r.json()["deleted"] is True
|
||
assert client.get(f"/api/sessions/{sid}").status_code == 404
|
||
|
||
|
||
def test_generate_requires_confirm_409(client):
|
||
sid = client.post("/api/sessions", json={"user_id": "u1"}).json()["session_id"]
|
||
_upload_core(client, sid)
|
||
# 未确认解析就生成 → 409
|
||
r = client.post(f"/api/sessions/{sid}/generate", json={})
|
||
assert r.status_code == 409
|
||
|
||
|
||
def test_start_impact_without_existing_system_409(client):
|
||
sid = client.post("/api/sessions", json={"user_id": "u1"}).json()["session_id"]
|
||
_upload_core(client, sid)
|
||
client.post(f"/api/sessions/{sid}/start-parse")
|
||
client.post(f"/api/sessions/{sid}/confirm-parse")
|
||
# 无既有系统 → writing，start-impact 409
|
||
r = client.post(f"/api/sessions/{sid}/start-impact")
|
||
assert r.status_code == 409
|
||
|
||
|
||
def test_qa_requires_generate_409(client):
|
||
sid = client.post("/api/sessions", json={"user_id": "u1"}).json()["session_id"]
|
||
_upload_core(client, sid)
|
||
client.post(f"/api/sessions/{sid}/start-parse")
|
||
client.post(f"/api/sessions/{sid}/confirm-parse")
|
||
r = client.post(f"/api/sessions/{sid}/run-qa")
|
||
assert r.status_code == 409
|
||
|
||
|
||
def test_report_downloads_after_full_flow(tmp_path):
|
||
"""QA/影响报告下载端点（FileResponse 分支）。"""
|
||
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}/start-parse")
|
||
client.post(f"/api/sessions/{sid}/confirm-parse")
|
||
client.post(f"/api/sessions/{sid}/generate", json={})
|
||
client.post(f"/api/sessions/{sid}/run-qa")
|
||
|
||
qa = client.get(f"/api/sessions/{sid}/result/qa-report")
|
||
assert qa.status_code == 200
|
||
assert "application/json" in qa.headers["content-type"]
|
||
qa_body = client.get(f"/api/sessions/{sid}/qa-result")
|
||
assert qa_body.status_code == 200
|
||
assert "overall_score" in qa_body.json()
|
||
|
||
# 无影响报告（未跑影响）→ 404
|
||
assert client.get(f"/api/sessions/{sid}/result/impact-report").status_code == 404
|
||
|
||
|
||
def test_session_not_found_404(client):
|
||
assert client.get("/api/sessions/nope").status_code == 404
|
||
|
||
|
||
def test_upload_to_missing_session_404(client):
|
||
r = client.post("/api/sessions/nope/files",
|
||
data={"file_type": "requirements"},
|
||
files={"file": ("a.xlsx", b"x")})
|
||
assert r.status_code == 404
|
||
|
||
|
||
def test_start_parse_no_files_409(client):
|
||
sid = client.post("/api/sessions", json={"user_id": "u1"}).json()["session_id"]
|
||
assert client.post(f"/api/sessions/{sid}/start-parse").status_code == 409
|
||
|
||
|
||
def test_start_parse_missing_session_404(client):
|
||
assert client.post("/api/sessions/nope/start-parse").status_code == 404
|
||
|
||
|
||
def test_confirm_parse_wrong_state_409(client):
|
||
sid = client.post("/api/sessions", json={"user_id": "u1"}).json()["session_id"]
|
||
# uploading 态直接确认解析 → 409
|
||
assert client.post(f"/api/sessions/{sid}/confirm-parse").status_code == 409
|
||
|
||
|
||
def _upload_zip(client, sid):
|
||
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")
|
||
client.post(f"/api/sessions/{sid}/files",
|
||
data={"file_type": "existing_system"},
|
||
files={"file": ("existing.zip", buf.getvalue())})
|
||
|
||
|
||
def test_impact_flow_http_zip(tmp_path):
|
||
"""HTTP 影响调查全流程 + 报告下载（app 164/168-170/174-178/226）。"""
|
||
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)
|
||
_upload_zip(client, sid)
|
||
client.post(f"/api/sessions/{sid}/start-parse")
|
||
assert client.post(f"/api/sessions/{sid}/confirm-parse").json()["status"] == "impact_running"
|
||
r = client.post(f"/api/sessions/{sid}/start-impact")
|
||
assert r.status_code == 200
|
||
ir = client.get(f"/api/sessions/{sid}/impact-result")
|
||
assert ir.status_code == 200 and "summary" in ir.json()
|
||
assert client.post(f"/api/sessions/{sid}/confirm-impact").json()["status"] == "writing"
|
||
dl = client.get(f"/api/sessions/{sid}/result/impact-report")
|
||
assert dl.status_code == 200 and "application/json" in dl.headers["content-type"]
|
||
|
||
|
||
def test_confirm_impact_wrong_state_409(client):
|
||
sid = client.post("/api/sessions", json={"user_id": "u1"}).json()["session_id"]
|
||
_upload_core(client, sid)
|
||
client.post(f"/api/sessions/{sid}/start-parse")
|
||
client.post(f"/api/sessions/{sid}/confirm-parse")
|
||
# writing（无既有系统）直接确认影响 → 409
|
||
assert client.post(f"/api/sessions/{sid}/confirm-impact").status_code == 409
|
||
|
||
|
||
def test_download_result_not_found_404(client):
|
||
sid = client.post("/api/sessions", json={"user_id": "u1"}).json()["session_id"]
|
||
assert client.get(f"/api/sessions/{sid}/result/download").status_code == 404
|
||
assert client.get(f"/api/sessions/{sid}/result/qa-report").status_code == 404
|
||
|
||
|
||
def test_real_mode_generate_without_key_500(tmp_path, monkeypatch):
|
||
"""engine=None（真实模式）→ generate 按需 build；build 抛错 → 500。"""
|
||
import genesis.server.app as app_mod
|
||
|
||
def boom():
|
||
raise RuntimeError("LLMNotConfiguredError: 缺少 API Key")
|
||
monkeypatch.setattr(app_mod, "build_inference_engine", boom)
|
||
|
||
app = create_app(
|
||
store=SessionStore(db_path=str(tmp_path / "s.db")),
|
||
data_root=str(tmp_path / "data"),
|
||
engine=None,
|
||
)
|
||
client = TestClient(app, raise_server_exceptions=False)
|
||
sid = client.post("/api/sessions", json={"user_id": "u1"}).json()["session_id"]
|
||
_upload_core(client, sid)
|
||
client.post(f"/api/sessions/{sid}/start-parse")
|
||
client.post(f"/api/sessions/{sid}/confirm-parse")
|
||
r = client.post(f"/api/sessions/{sid}/generate", json={})
|
||
assert r.status_code == 500
|
||
|
||
|
||
# ---------- 项目配置（S6） ----------
|
||
|
||
def test_create_and_list_projects(client):
|
||
body = {
|
||
"name": "stock", "display_name": "股票系统",
|
||
"template": str(_SAMPLE / "template_design_ja.docx"),
|
||
"write_instruction": "", "rules": [],
|
||
"existing_system_code_dir": "", "design_docs_dir": "",
|
||
}
|
||
r = client.post("/api/projects", json=body)
|
||
assert r.status_code == 200
|
||
assert r.json()["name"] == "stock"
|
||
lst = client.get("/api/projects").json()
|
||
assert any(p["name"] == "stock" for p in lst)
|
||
one = client.get("/api/projects/stock").json()
|
||
assert one["template"] == str(_SAMPLE / "template_design_ja.docx")
|
||
|
||
|
||
def test_project_invalid_template_400(client):
|
||
r = client.post("/api/projects", json={"name": "p", "template": "/no/such.docx"})
|
||
assert r.status_code == 400
|
||
assert r.json()["detail"]["code"] == "PROJECT_CONFIG_INVALID"
|
||
|
||
|
||
def test_delete_project(client):
|
||
client.post("/api/projects", json={"name": "p", "template": str(_SAMPLE / "template_design_ja.docx")})
|
||
assert client.delete("/api/projects/p").json()["deleted"] is True
|
||
assert client.get("/api/projects/p").status_code == 404
|
||
|
||
|
||
def test_session_accepts_name_and_project(client):
|
||
r = client.post("/api/sessions", json={"user_id": "u1", "name": "我的会话", "project": "projA"})
|
||
assert r.status_code == 200
|
||
sid = r.json()["session_id"]
|
||
assert r.json()["name"] == "我的会话"
|
||
g = client.get(f"/api/sessions/{sid}").json()
|
||
assert g["name"] == "我的会话"
|
||
assert g["project"] == "projA"
|
||
|
||
|
||
def test_session_list_includes_name_project(client):
|
||
client.post("/api/sessions", json={"user_id": "u1", "name": "n1", "project": "pA"})
|
||
r = client.get("/api/sessions", params={"user_id": "u1"}).json()
|
||
assert any(s["name"] == "n1" and s["project"] == "pA" for s in r)
|
||
|
||
|
||
def test_generate_with_project_config_no_template_upload(client):
|
||
"""绑定项目（含模板/规则）后，仅上传要件定义即可解析生成。"""
|
||
client.post("/api/projects", json={
|
||
"name": "projA", "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": "",
|
||
})
|
||
sid = client.post("/api/sessions", json={"user_id": "u1", "project": "projA"}).json()["session_id"]
|
||
# 仅上传要件定义
|
||
r = client.post(f"/api/sessions/{sid}/files", data={"file_type": "requirements"},
|
||
files={"file": ("requirements_newdev.xlsx", (_SAMPLE / "requirements_newdev.xlsx").read_bytes())})
|
||
assert r.status_code == 200
|
||
p = client.post(f"/api/sessions/{sid}/start-parse")
|
||
assert p.status_code == 200, p.text
|
||
assert client.post(f"/api/sessions/{sid}/confirm-parse").status_code == 200
|
||
# 无既有系统 → 直接 writing，可生成
|
||
gen = client.post(f"/api/sessions/{sid}/generate", json={})
|
||
assert gen.status_code == 200, gen.text
|
||
assert client.get(f"/api/sessions/{sid}/result/download").status_code == 200
|
||
|
||
def test_api_list_sessions_filter_by_project(client):
|
||
r0 = client.post("/api/sessions", json={"user_id": "u1", "project": "stock"})
|
||
r1 = client.post("/api/sessions", json={"user_id": "u1", "project": "other"})
|
||
res = client.get("/api/sessions", params={"user_id": "u1", "project": "stock"})
|
||
assert res.status_code == 200
|
||
ids = {x["session_id"] for x in res.json()}
|
||
assert r0.json()["session_id"] in ids
|
||
assert r1.json()["session_id"] not in ids
|