201 lines
8.4 KiB
Python
201 lines
8.4 KiB
Python
"""个人设置 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()
|