198 lines
7.2 KiB
Python
198 lines
7.2 KiB
Python
"""Web 全链路端到端测试：初始化DB → 启动Flask → 注册 → 生成 → 轮询 → 下载。
|
|
|
|
用法：
|
|
python tests/e2e_web.py
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
import urllib.error
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
BASE = "http://127.0.0.1:5099"
|
|
USER = f"e2e_user_{int(time.time())}"
|
|
PASS = "e2e_pass_123"
|
|
SAMPLE = """# 项目背景
|
|
为提升内部知识周转效率，启动知识库平台建设
|
|
已完成技术选型与原型验证
|
|
# 本期成果
|
|
部署完成基础服务，导入文档120份
|
|
混合检索准确率显著优于基线方案
|
|
# 下一步
|
|
扩大试点范围至三个部门
|
|
建立内容运营与质量巡检机制
|
|
"""
|
|
|
|
|
|
def call(method: str, path: str, payload=None, raw=False):
|
|
url = BASE + path
|
|
data = None if payload is None else json.dumps(payload).encode("utf-8")
|
|
req = urllib.request.Request(url, data=data, method=method)
|
|
if payload is not None:
|
|
req.add_header("Content-Type", "application/json")
|
|
# cookie 会话
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
body = resp.read()
|
|
set_cookie = resp.headers.get("Set-Cookie")
|
|
return resp.status, (body if raw else json.loads(body or b"{}")), set_cookie
|
|
except urllib.error.HTTPError as e:
|
|
body = e.read()
|
|
try:
|
|
return e.code, json.loads(body or b"{}"), e.headers.get("Set-Cookie")
|
|
except Exception:
|
|
return e.code, {"raw": body.decode("utf-8", "ignore")}, e.headers.get("Set-Cookie")
|
|
|
|
|
|
COOKIE = {"value": ""}
|
|
|
|
|
|
def callx(method, path, payload=None, raw=False):
|
|
status, body, sc = call(method, path, payload, raw)
|
|
if sc and COOKIE["value"] == "":
|
|
COOKIE["value"] = sc.split(";")[0]
|
|
return status, body
|
|
|
|
|
|
# 注入 cookie
|
|
_orig_call = call
|
|
|
|
|
|
def call(method, path, payload=None, raw=False): # noqa: F811
|
|
url = BASE + path
|
|
data = None if payload is None else json.dumps(payload).encode("utf-8")
|
|
req = urllib.request.Request(url, data=data, method=method)
|
|
if payload is not None:
|
|
req.add_header("Content-Type", "application/json")
|
|
if COOKIE["value"]:
|
|
req.add_header("Cookie", COOKIE["value"])
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=60) as resp:
|
|
body = resp.read()
|
|
sc = resp.headers.get("Set-Cookie")
|
|
if sc and not COOKIE["value"]:
|
|
COOKIE["value"] = sc.split(";")[0]
|
|
return resp.status, (body if raw else json.loads(body or b"{}"))
|
|
except urllib.error.HTTPError as e:
|
|
body = e.read()
|
|
try:
|
|
parsed = json.loads(body or b"{}")
|
|
except Exception:
|
|
parsed = {"raw": body.decode("utf-8", "ignore")}
|
|
sc = e.headers.get("Set-Cookie")
|
|
if sc and not COOKIE["value"]:
|
|
COOKIE["value"] = sc.split(";")[0]
|
|
return e.code, parsed
|
|
|
|
|
|
def main():
|
|
failures = []
|
|
|
|
def check(name, cond, extra=""):
|
|
print(f" [{'OK' if cond else 'NG'}] {name} {extra}")
|
|
if not cond:
|
|
failures.append(name)
|
|
|
|
print("== 初始化数据库 ==")
|
|
import subprocess
|
|
r = subprocess.run([sys.executable, "-m", "src.db.init"], cwd=str(ROOT),
|
|
capture_output=True, text=True, encoding="utf-8", errors="replace",
|
|
env={**__import__("os").environ, "PYTHONIOENCODING": "utf-8"})
|
|
check("db init", r.returncode == 0, r.stdout.strip()[-40:] + r.stderr.strip()[-80:])
|
|
|
|
print("== 启动 Flask ==")
|
|
proc = subprocess.Popen(
|
|
[sys.executable, "-c",
|
|
f"import sys; sys.path.insert(0, r'{ROOT}');"
|
|
"from src.web.app import app; app.run(host='127.0.0.1', port=5099)"],
|
|
cwd=str(ROOT), env={**__import__("os").environ, "PYTHONIOENCODING": "utf-8"},
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
)
|
|
try:
|
|
up = False
|
|
for _ in range(30):
|
|
time.sleep(0.5)
|
|
try:
|
|
s, _b = call("GET", "/api/me")
|
|
up = True
|
|
break
|
|
except Exception:
|
|
continue
|
|
check("server up", up)
|
|
|
|
print("== 认证链路 ==")
|
|
s, b = call("POST", "/api/register",
|
|
{"username": USER, "password": PASS, "confirm_password": PASS})
|
|
check("register", s == 200 and b.get("code") == 0, str(b)[:60])
|
|
|
|
s, b = call("GET", "/api/me")
|
|
check("me with session", s == 200 and b.get("data", {}).get("username") == USER)
|
|
prefs = b.get("data", {}).get("preferences") or {}
|
|
|
|
print("== 校验拦截 ==")
|
|
s, b = call("POST", "/api/generate",
|
|
{**{"title": "t", "content": "太短", "scene": "report"},
|
|
**{k: prefs.get("default_" + k[0:-4], v) for k, v in []}})
|
|
check("short content rejected (2001)", b.get("code") == 2001, str(b)[:60])
|
|
|
|
print("== 真实生成 ==")
|
|
gen_payload = {
|
|
"title": "E2E冒烟报告",
|
|
"content": SAMPLE,
|
|
"scene": "report", "language": "zh", "canvas": "ppt169",
|
|
"image_strategy": "off", "color_scheme": "blue",
|
|
"page_min": 5, "page_max": 8,
|
|
}
|
|
s, b = call("POST", "/api/generate", gen_payload)
|
|
check("generate accepted", s == 200 and b.get("code") == 0, str(b)[:80])
|
|
task_id = (b.get("data") or {}).get("task_id")
|
|
|
|
final = None
|
|
for _ in range(120):
|
|
time.sleep(1)
|
|
s, b = call("GET", f"/api/generate/{task_id}/status")
|
|
d = b.get("data") or {}
|
|
if d.get("status") in ("done", "failed"):
|
|
final = d
|
|
break
|
|
check("generation done", final and final.get("status") == "done", str(final)[:120])
|
|
|
|
s, b = call("GET", f"/api/generate/{task_id}/result")
|
|
record = (b.get("data") or {}).get("record") or {}
|
|
rid = record.get("id")
|
|
check("result has record", bool(rid), f"id={rid} pages={record.get('page_count')} size={record.get('file_size')}")
|
|
check("page_count sane", 5 <= (record.get("page_count") or 0) <= 8,
|
|
str(record.get("page_count")))
|
|
|
|
print("== 历史与下载 ==")
|
|
s, b = call("GET", "/api/records?keyword=E2E")
|
|
recs = ((b.get("data") or {}).get("records")) or []
|
|
check("records list contains new", any(x["id"] == rid for x in recs))
|
|
|
|
s, body_bytes = call("GET", f"/api/files/{rid}/download", raw=True)
|
|
ok_pptx = body_bytes[:2] == b"P" + b"K" # PK zip header
|
|
check("download pptx bytes", s == 200 and ok_pptx, f"{len(body_bytes)} bytes head={body_bytes[:2]}")
|
|
|
|
pptx_path = ROOT / "tests" / "_e2e_download.pptx"
|
|
pptx_path.write_bytes(body_bytes)
|
|
from pptx import Presentation
|
|
prs = Presentation(str(pptx_path))
|
|
texts = "\n".join(sh.text_frame.text for sl in prs.slides
|
|
for sh in sl.shapes if sh.has_text_frame)
|
|
check("pptx contains title", "E2E冒烟报告" in texts)
|
|
check("pptx contains content keyword", "混合检索" in texts)
|
|
finally:
|
|
proc.terminate()
|
|
|
|
print("\n" + ("E2E ALL PASSED" if not failures else "E2E FAILURES: " + "; ".join(failures)))
|
|
sys.exit(1 if failures else 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|