2026Technology-Competition/tests/e2e_chat.py

"""对话修改闭环 e2e：生成 → 换色 → 改标题 → 改页内容 → 两版本/下载验证。

用法：python tests/e2e_chat.py
"""

import json
import os
import subprocess
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:5098"
USER = f"chat_e2e_{int(time.time())}"
PASS = "chat_e2e_123"
SAMPLE = "# 项目背景\n为提升内部知识周转效率，启动知识库平台建设\n已完成技术选型与原型验证\n# 本期成果\n部署完成基础服务\n导入文档120份\n"

COOKIE = {"value": ""}


def call(method, path, 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")
    if COOKIE["value"]:
        req.add_header("Cookie", COOKIE["value"])
    try:
        with urllib.request.urlopen(req, timeout=120) 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)

    env = {**os.environ, "PYTHONIOENCODING": "utf-8"}
    subprocess.run([sys.executable, "-m", "src.db.init"], cwd=str(ROOT),
                   capture_output=True, env=env)
    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=5098)"],
        cwd=str(ROOT), env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
    )
    try:
        up = False
        for _ in range(30):
            time.sleep(0.5)
            try:
                call("GET", "/api/me")
                up = True
                break
            except Exception:
                continue
        check("server up", up)

        s, b = call("POST", "/api/register",
                    {"username": USER, "password": PASS, "confirm_password": PASS})
        check("register", s == 200 and b.get("code") == 0)

        s, b = call("POST", "/api/generate", {
            "title": "对话修改测试", "content": SAMPLE,
            "scene": "report", "language": "zh", "canvas": "ppt169",
            "image_strategy": "off", "color_scheme": "blue",
            "page_min": 5, "page_max": 8,
        })
        check("generate accepted", b.get("code") == 0, str(b)[:60])
        task_id = b["data"]["task_id"]
        for _ in range(120):
            time.sleep(1)
            s, b = call("GET", f"/api/generate/{task_id}/status")
            if b["data"]["status"] in ("done", "failed"):
                break
        check("generation done", b["data"]["status"] == "done", str(b)[:80])
        s, b = call("GET", f"/api/generate/{task_id}/result")
        rid = b["data"]["record"]["id"]
        check("record ready", bool(rid), f"rid={rid}")

        def chat(msg):
            return call("POST", "/api/chat", {"record_id": rid, "message": msg})

        s, b = chat("换成绿色")
        d = b.get("data") or {}
        check("change color", d.get("changed") is True, str(d.get("reply", ""))[:60])
        check("record color updated", (d.get("record") or {}).get("color_scheme") == "green")

        s, b = chat('把标题改成"知识平台周报"')
        d = b.get("data") or {}
        check("change title", d.get("changed") is True
              and (d.get("record") or {}).get("title") == "知识平台周报",
              str(d.get("reply", ""))[:60])

        s, b = chat('把"知识周转"改成"信息流转"')
        d = b.get("data") or {}
        check("edit page replace", d.get("changed") is True, str(d.get("reply", ""))[:60])

        s, b = chat("今天天气怎么样")
        d = b.get("data") or {}
        check("unknown intent no change", d.get("changed") is False
              and "不支持" in d.get("reply", ""), str(d.get("reply", ""))[:50])

        s, b = call("GET", f"/api/records/{rid}")
        versions = b["data"]["versions"]
        check("two versions kept", len(versions) == 2 and versions[0]["version_no"] == 1,
              f"n={len(versions)}")

        s, b = call("GET", f"/api/chat/{rid}/messages")
        msgs = b["data"]["messages"]
        check("messages logged", len(msgs) == 8, f"n={len(msgs)}")

        s, raw = call("GET", f"/api/files/{rid}/download", raw=True)
        check("download ok", s == 200 and raw[:2] == b"PK", f"{len(raw)}B")
        p = ROOT / "tests" / "_chat_download.pptx"
        p.write_bytes(raw)
        from pptx import Presentation
        prs = Presentation(str(p))
        texts = "\n".join(sh.text_frame.text for sl in prs.slides
                          for sh in sl.shapes if sh.has_text_frame)
        check("new title in pptx", "知识平台周报" in texts)
        check("edited text in pptx", "信息流转" in texts and "知识周转" not in texts)
        p.unlink()
    finally:
        proc.terminate()

    print("\n" + ("CHAT E2E ALL PASSED" if not failures
                  else "FAILURES: " + "; ".join(failures)))
    sys.exit(1 if failures else 0)


if __name__ == "__main__":
    main()