"""个人设置 e2e：偏好读写校验、改用户名、改密码、历史语言筛选。 用法：python tests/e2e_settings.py """ import json import os import sqlite3 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:5101" _TS = int(time.time()) UA = f"sxa{_TS}" UB = f"sxb{_TS}" NEW_NAME = f"sxc{_TS}" PASS = "set_e2e_123" COOKIE = {"value": ""} def call(method, path, payload=None): 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, json.loads(body or b"{}") except urllib.error.HTTPError as e: try: return e.code, json.loads(e.read() or b"{}") except Exception: return e.code, {} def drop_cookie(): COOKIE["value"] = "" def insert_record(user_id, title, language, created_at): from config import DB_PATH conn = sqlite3.connect(str(DB_PATH)) conn.execute( "INSERT INTO ppt_records (user_id,title,content,scene,language,color_scheme," "page_min,page_max,status,created_at) VALUES (?,?,?,?,?,?,?,?,?,?)", (user_id, title, "内容示例文本，用于语言筛选测试", "report", language, "blue", 10, 15, "done", created_at), ) conn.commit() uid = conn.execute("SELECT id FROM ppt_records WHERE title=?", (title,)).fetchone()[0] conn.close() return uid 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=5101)"], 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": UA, "password": PASS, "confirm_password": PASS}) check("register A", s == 200 and b.get("code") == 0) # ---- GET /api/settings 默认值 ---- s, b = call("GET", "/api/settings") p = (b.get("data") or {}).get("preferences") or {} check("settings defaults", s == 200 and b["data"]["username"] == UA and p.get("default_scene") == "training" and p.get("default_color") == "blue" and p.get("default_language") == "zh" and p.get("default_canvas") == "ppt169" and p.get("default_image_strategy") == "web" and p.get("default_page_min") == 10 and p.get("default_page_max") == 15, str(p)[:80]) # ---- 偏好保存（部分字段）---- s, b = call("PUT", "/api/settings/preferences", {"default_color": "purple", "default_language": "ja", "default_page_min": 5, "default_page_max": 8}) p = ((b.get("data") or {}).get("preferences")) or {} check("save prefs partial", s == 200 and p.get("default_color") == "purple" and p.get("default_language") == "ja" and p.get("default_page_min") == 5 and p.get("default_page_max") == 8 and p.get("default_scene") == "training", str(p)[:80]) s, b = call("PUT", "/api/settings/preferences", {"default_color": "pink"}) check("prefs invalid enum -> 5002", s == 400 and b.get("code") == 5002, str(b)[:60]) s, b = call("PUT", "/api/settings/preferences", {"default_page_min": 20, "default_page_max": 8}) check("prefs min>max -> 2002", s == 400 and b.get("code") == 2002, str(b)[:60]) # ---- 改用户名 ---- s, b = call("PUT", "/api/settings/username", {"username": "ab"}) check("username too short -> 5002", s == 400 and b.get("code") == 5002, str(b)[:50]) s, b = call("POST", "/api/register", {"username": UB, "password": PASS, "confirm_password": PASS}) drop_cookie() call("POST", "/api/login", {"username": UA, "password": PASS}) s, b = call("PUT", "/api/settings/username", {"username": UB}) check("username duplicate -> 1001", s == 409 and b.get("code") == 1001, str(b)[:50]) new_name = NEW_NAME s, b = call("PUT", "/api/settings/username", {"username": new_name}) s2, b2 = call("GET", "/api/me") check("username changed", s == 200 and b2["data"]["username"] == new_name, str(b2)[:60]) # ---- 改密码 ---- s, b = call("PUT", "/api/settings/password", {"old_password": "wrong_old", "new_password": "newpwd456", "confirm_password": "newpwd456"}) check("wrong old pwd -> 5001", s == 400 and b.get("code") == 5001, str(b)[:50]) s, b = call("PUT", "/api/settings/password", {"old_password": PASS, "new_password": "newpwd456", "confirm_password": "different"}) check("confirm mismatch -> 1003", s == 400 and b.get("code") == 1003, str(b)[:50]) s, b = call("PUT", "/api/settings/password", {"old_password": PASS, "new_password": "abc", "confirm_password": "abc"}) check("short new pwd -> 5002", s == 400 and b.get("code") == 5002, str(b)[:50]) s, b = call("PUT", "/api/settings/password", {"old_password": PASS, "new_password": "newpwd456", "confirm_password": "newpwd456"}) check("password changed", s == 200 and b.get("code") == 0) drop_cookie() s, _ = call("POST", "/api/login", {"username": new_name, "password": PASS}) check("old password rejected -> 401", s == 401) s, b = call("POST", "/api/login", {"username": new_name, "password": "newpwd456"}) check("login with new password", s == 200 and b.get("code") == 0) # ---- 历史语言筛选 ---- uid = None from src.agent import memory as mem u = mem.get_user_by_name(new_name) uid = u["id"] insert_record(uid, "筛选测试中文稿", "zh", "2026-08-25 10:00:00") insert_record(uid, "フィルタ日本語資料", "ja", "2026-08-25 10:01:00") s, b = call("GET", "/api/records?language=ja") titles_ja = [r["title"] for r in b["data"]["records"]] s, b = call("GET", "/api/records?language=zh") titles_zh = [r["title"] for r in b["data"]["records"]] s, b = call("GET", "/api/records") titles_all = [r["title"] for r in b["data"]["records"]] check("filter ja only", "フィルタ日本語資料" in titles_ja and "筛选测试中文稿" not in titles_ja, str(titles_ja)[:70]) check("filter zh only", "筛选测试中文稿" in titles_zh and "フィルタ日本語資料" not in titles_zh, str(titles_zh)[:70]) check("no filter shows both", "筛选测试中文稿" in titles_all and "フィルタ日本語資料" in titles_all, f"n={len(titles_all)}") # 清理测试插入的记录 conn = sqlite3.connect(str(ROOT / "src" / "web" / "data" / "ppt.db")) conn.execute("DELETE FROM ppt_records WHERE title IN (?,?)", ("筛选测试中文稿", "フィルタ日本語資料")) conn.commit() conn.close() finally: proc.terminate() print("\n" + ("SETTINGS E2E ALL PASSED" if not failures else "FAILURES: " + "; ".join(failures))) sys.exit(1 if failures else 0) if __name__ == "__main__": main()