Files
note-manager/smoke_test.py
T
Your Name 13c53fea0a feat: 管理员跨租户管理 + 注册开关
- 管理员可管理任意用户笔记(读取/修改/删除/回收站/标签/图谱/FTS 全平台)
- 普通用户仍数据隔离, 越权返回404
- 新增注册开关: 管理员后台⚙设置可开/关, 支持REGISTRATION_ENABLED环境变量
- 注册关闭时前台/登录页隐藏注册入口, 注册接口返回400
- 冒烟测试扩展到91用例全过
2026-08-11 13:07:56 +08:00

479 lines
22 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""note-manager 新功能端到端冒烟测试"""
import json
import re
import sys
import urllib.request
import urllib.parse
import http.cookiejar
import io
import zipfile
BASE = "http://localhost:8080"
cj = http.cookiejar.CookieJar()
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj))
PASSED = 0
FAILED = 0
def report(name, ok, detail=""):
global PASSED, FAILED
if ok:
PASSED += 1
print(f"{name}")
else:
FAILED += 1
print(f"{name} {detail}")
def req(method, path, data=None, raw=False, headers=None):
url = BASE + path
body = None
hdrs = dict(headers or {})
if data is not None:
if isinstance(data, (dict, list)):
body = json.dumps(data).encode()
hdrs["Content-Type"] = "application/json"
elif isinstance(data, str):
body = data.encode()
hdrs["Content-Type"] = "application/x-www-form-urlencoded"
r = urllib.request.Request(url, data=body, method=method, headers=hdrs)
try:
resp = opener.open(r, timeout=10)
content = resp.read()
if raw:
return resp.status, content, dict(resp.headers)
return resp.status, json.loads(content.decode() if content else "{}")
except urllib.error.HTTPError as e:
content = e.read()
if raw:
return e.code, content, dict(e.headers)
try:
return e.code, json.loads(content.decode() if content else "{}")
except Exception:
return e.code, {}
except Exception as ex:
return -1, {"error": str(ex)}
print("═══ 1. 认证安全(多租户用户名+密码) ═══")
# 未登录访问管理 API -> 401
st, d = req("GET", "/admin/api/tree")
report("未登录访问管理API返回401", st == 401, f"(got {st})")
st, d = req("POST", "/admin/api/notes")
report("未登录创建笔记返回401", st == 401, f"(got {st})")
st, d = req("POST", "/admin/notes")
report("未登录(错误路径)不泄露", st in (401, 404), f"(got {st})")
# 错误密码登录
st, d = req("POST", "/admin/login", "username=smoketest&password=wrongpass")
report("错误密码登录返回401", st == 401, f"(got {st})")
# 用随机用户名注册一个专属测试管理员(保证可重复运行)
import uuid as _uuid
TUSER = "smoke_" + _uuid.uuid4().hex[:8]
st, d = req("POST", "/api/auth/register", {"username": TUSER, "password": "smoke123", "display_name": "冒烟测试"})
report("注册测试用户成功", st == 200 and d.get("code") == 0, f"(got {st} {d})")
# 注册后自动登录 -> 可访问管理 API
st, d = req("GET", "/admin/api/tree")
report("注册后自动登录(管理API可用)", st == 200 and d.get("code") == 0, f"(got {st})")
# 退出登录
st, d = req("POST", "/api/auth/logout")
report("登出成功", st == 200 and d.get("code") == 0, f"(got {st})")
# 用户名+密码正确登录
st, d = req("POST", "/api/auth/login", {"username": TUSER, "password": "smoke123"})
report("用户名密码登录成功", st == 200 and d.get("code") == 0, f"(got {st} {d})")
# 登录后可访问
st, d = req("GET", "/admin/api/tree")
report("登录后访问管理API成功", st == 200 and d.get("code") == 0, f"(got {st})")
# 重复注册同一用户名 -> 失败
st, d = req("POST", "/api/auth/register", {"username": TUSER, "password": "smoke123"})
report("重复注册同名用户被拒", st == 400, f"(got {st} {d})")
# 旧固定 cookie 不再有效
import urllib.request as u
reqf = urllib.request.Request(BASE + "/admin/api/tree", headers={"Cookie": "note_token=authenticated"})
try:
r = u.urlopen(reqf, timeout=10)
st_old = r.status
except urllib.error.HTTPError as e:
st_old = e.code
report("固定字符串cookie已失效(返回401)", st_old == 401, f"(got {st_old})")
print("═══ 2. 笔记 CRUD + 版本历史 ═══")
st, d = req("POST", "/admin/api/notes", {
"title": "测试笔记A", "content": "第一版内容", "category": "测试", "tags": '["go","测试"]', "is_public": True
})
note_id = d.get("data", {}).get("id")
report("创建笔记成功", st == 200 and note_id, f"(got {st} {d})")
st, d = req("PUT", f"/admin/api/notes/{note_id}", {"content": "第二版内容更新了"})
report("更新笔记成功", st == 200 and d.get("code") == 0, f"(got {st})")
st, d = req("GET", f"/admin/api/notes/{note_id}/versions")
vers = d.get("data", [])
report("版本历史记录>=2", st == 200 and len(vers) >= 2, f"(got {len(vers)})")
# 回滚到第一版
old_ver = vers[-1]
st, d = req("POST", f"/admin/api/notes/{note_id}/restore-version", "version_id=%d" % old_ver["id"])
report("恢复历史版本成功", st == 200 and d.get("code") == 0, f"(got {st})")
st, d = req("GET", f"/admin/api/notes/{note_id}")
report("恢复后内容为第一版", d.get("data", {}).get("content") == "第一版内容", f"(got {d.get('data',{}).get('content')})")
print("═══ 3. 安全:公开接口不泄露带密码笔记 ═══")
# 创建一个带密码的公开笔记
st, d = req("POST", "/admin/api/notes", {
"title": "受保护笔记", "content": "秘密内容", "is_public": True, "password": "secret123"
})
protected_id = d.get("data", {}).get("id")
report("创建带密码笔记成功", st == 200 and protected_id, f"(got {st})")
# 登出后(游客身份)通过公开接口 GET /api/notes/:id 访问 -> 应被拒
req("POST", "/api/auth/logout")
st, d = req("GET", f"/api/notes/{protected_id}")
report("游客公开接口拒绝带密码笔记", st in (401, 403), f"(got {st})")
# 密码验证接口(公开,游客)应能正确返回
st, d = req("POST", f"/api/notes/{protected_id}/access", {"password": "secret123"})
report("密码验证访问成功", st == 200 and "秘密内容" in d.get("data", {}).get("content", ""), f"(got {st})")
st, d = req("POST", f"/api/notes/{protected_id}/access", {"password": "wrong"})
report("错误密码访问被拒", st == 401, f"(got {st})")
# 公开无密码笔记(游客)
st, d = req("GET", f"/api/notes/{note_id}")
report("游客公开接口读取无密码笔记成功", st == 200 and d.get("code") == 0, f"(got {st})")
# 重新登录回测试用户
req("POST", "/api/auth/login", {"username": TUSER, "password": "smoke123"})
print("═══ 4. 回收站(软删除/恢复/清空) ═══")
# 删除受保护笔记 -> 进入回收站
st, d = req("DELETE", f"/admin/api/notes/{protected_id}")
report("软删除笔记成功", st == 200 and d.get("code") == 0, f"(got {st})")
# 前台公开接口应查不到已删除笔记
st, d = req("GET", f"/api/notes/{protected_id}")
report("已删除笔记公开接口404", st == 404, f"(got {st})")
st, d = req("GET", "/admin/api/trash")
trash = d.get("data", [])
report("回收站包含已删笔记", any(t.get("id") == protected_id for t in trash), f"(got {[t.get('id') for t in trash]})")
# 恢复
st, d = req("POST", f"/admin/api/restore/{protected_id}")
report("从回收站恢复成功", st == 200 and d.get("code") == 0, f"(got {st})")
st, d = req("GET", "/admin/api/trash")
trash = d.get("data", [])
report("恢复后回收站不再包含该笔记", not any(t.get("id") == protected_id for t in trash))
st, d = req("GET", f"/api/notes/{protected_id}")
report("恢复后可再次访问(仍带密码被拒)", st in (401, 403, 200), f"(got {st})")
# 彻底删除测试
st, d = req("POST", "/admin/api/notes", {"title": "待彻底删除", "content": "x", "is_public": True})
tmp_id = d.get("data", {}).get("id")
req("DELETE", f"/admin/api/notes/{tmp_id}")
st, d = req("POST", f"/admin/api/purge/{tmp_id}")
report("彻底删除成功", st == 200 and d.get("code") == 0, f"(got {st})")
st, d = req("GET", "/admin/api/trash")
report("彻底删除后回收站无该记录", not any(t.get("id") == tmp_id for t in d.get("data", [])))
print("═══ 5. 分享链接 ═══")
# 给测试笔记A创建分享
st, d = req("POST", f"/admin/api/notes/{note_id}/share", "expire_hours=24")
share_token = d.get("data", {}).get("share_token")
report("创建分享链接成功", st == 200 and share_token, f"(got {st} {d})")
# 通过分享链接访问
st, d = req("GET", f"/api/share/{share_token}")
report("分享链接可公开访问", st == 200 and d.get("data", {}).get("id") == note_id, f"(got {st})")
# 分享阅读页
st, raw, _ = req("GET", f"/share/{share_token}", raw=True)
report("分享阅读页返回HTML", st == 200 and b"<html" in raw.lower(), f"(got {st})")
# 撤销分享
st, d = req("POST", f"/admin/api/notes/{note_id}/revoke-share")
report("撤销分享成功", st == 200 and d.get("code") == 0, f"(got {st})")
st, d = req("GET", f"/api/share/{share_token}")
report("撤销后分享链接失效", st == 404, f"(got {st})")
print("═══ 6. 批量导出 zip ═══")
st, raw, hdrs = req("GET", "/admin/api/export-all", raw=True)
is_zip = False
if st == 200:
try:
zf = zipfile.ZipFile(io.BytesIO(raw))
names = zf.namelist()
is_zip = True
report("批量导出zip包含测试笔记", any("测试笔记" in n for n in names), f"(names={names})")
report("批量导出zip含front matter", any("---" in zf.read(n).decode()[:200] for n in names if n.endswith(".md")))
except Exception as ex:
report("批量导出zip解析", False, f"(err {ex})")
report("批量导出返回zip", st == 200 and raw[:2] == b"PK" and is_zip, f"(got {st})")
print("═══ 7. 单笔记导出 + 图片上传嗅探 ═══")
st, d = req("GET", f"/admin/api/notes/{note_id}")
report("后台读取笔记详情成功", st == 200 and d.get("data", {}).get("id") == note_id)
st, raw, _ = req("GET", f"/admin/api/export/{note_id}", raw=True)
report("单笔记导出Markdown", st == 200 and "测试笔记" in raw.decode('utf-8', 'ignore'), f"(got {st})")
# 上传伪造文件(伪装成 png 的文本)
import uuid
fake_bytes = b"<script>alert(1)</script>"
# 用 multipart 上传
boundary = "----WebKitFormBoundary" + uuid.uuid4().hex
body = (
f"--{boundary}\r\n"
f'Content-Disposition: form-data; name="image"; filename="evil.png"\r\n'
f"Content-Type: image/png\r\n\r\n"
).encode() + fake_bytes + f"\r\n--{boundary}--\r\n".encode()
st, d = req("POST", "/admin/api/upload", data=body, headers={"Content-Type": f"multipart/form-data; boundary={boundary}"}, raw=False)
report("伪造图片(非真实图片)被拒", st in (400, 500), f"(got {st} {d})")
print("═══ 8. 列表/分类/标签/树 ═══")
st, d = req("GET", "/api/notes")
report("公开列表接口正常", st == 200 and d.get("code") == 0)
st, d = req("GET", "/api/categories")
report("分类列表正常", st == 200 and "测试" in d.get("data", []))
st, d = req("GET", "/api/tags")
report("标签列表正常", st == 200)
st, d = req("GET", "/api/tree")
report("公开树正常", st == 200)
print("═══ 9. FTS5 全文搜索 ═══")
# 创建一篇含中文与英文的笔记用于搜索测试
st, d = req("POST", "/admin/api/notes", {
"title": "搜索引擎优化笔记", "content": "这是一篇关于数据库索引与搜索引擎的笔记 涉及中文分词 ASP.NET",
"category": "技术", "tags": '["全文","优化"]', "is_public": True
})
fts_id = d.get("data", {}).get("id")
report("创建搜索测试笔记", st == 200 and fts_id, f"(got {st})")
# 中文短词(1~2字)搜索
q_enc = urllib.parse.urlencode({"q": "数据库"})
st, d = req("GET", f"/api/notes/search?{q_enc}")
hits = [x.get("title") for x in (d.get("data") or [])]
report("中文短词搜索命中", st == 200 and "搜索引擎优化笔记" in hits, f"(got {hits})")
# 英文搜索
q_enc = urllib.parse.urlencode({"q": "ASP.NET"})
st, d = req("GET", f"/api/notes/search?{q_enc}")
hits = [x.get("title") for x in (d.get("data") or [])]
report("英文搜索命中", st == 200 and "搜索引擎优化笔记" in hits, f"(got {hits})")
# 无结果关键词
q_enc = urllib.parse.urlencode({"q": "不存在的关键词XYZ"})
st, d = req("GET", f"/api/notes/search?{q_enc}")
report("无匹配关键词返回空", st == 200 and d.get("total", 0) == 0, f"(got {d.get('total')})")
print("═══ 10. 双向链接 / 知识图谱 ═══")
# 创建两篇互相/单向链接的笔记
st, da = req("POST", "/admin/api/notes", {
"title": "链接源A", "content": f"参考 [[搜索引擎优化笔记]] 和 [[链接目标B]]",
"is_public": True
})
linkA_id = da.get("data", {}).get("id")
st, db_ = req("POST", "/admin/api/notes", {
"title": "链接目标B", "content": "内容B", "is_public": True
})
linkB_id = db_.get("data", {}).get("id")
report("创建链接笔记成功", st == 200 and linkA_id and linkB_id, f"(got {st})")
# 反向链接:谁链接到了搜索引擎优化笔记(应为 链接源A)
st, d = req("GET", f"/admin/api/notes/{fts_id}/backlinks")
backlinks = [x.get("title") for x in (d.get("data") or [])]
report("反向链接查询", st == 200 and "链接源A" in backlinks, f"(got {backlinks})")
# 知识图谱:应包含节点,且存在 edges
st, d = req("GET", "/admin/api/graph")
g = d.get("data") or {}
gnodes = g.get("nodes") or []
gedges = g.get("edges") or []
report("知识图谱返回节点", st == 200 and any(n.get("title") == "链接源A" for n in gnodes), f"(got {st})")
report("知识图谱含链接边", st == 200 and len(gedges) >= 1, f"(got {len(gedges)})")
print("═══ 11. 自动保存草稿 ═══")
st, d = req("POST", f"/admin/api/notes/{fts_id}/draft", {"content": "这是未保存的草稿"})
report("保存草稿成功", st == 200 and d.get("code") == 0, f"(got {st})")
st, d = req("GET", f"/admin/api/notes/{fts_id}")
report("读取到草稿字段", d.get("data", {}).get("draft_content") == "这是未保存的草稿", f"(got {d.get('data',{}).get('draft_content')})")
st, d = req("DELETE", f"/admin/api/notes/{fts_id}/draft")
report("清除草稿成功", st == 200 and d.get("code") == 0, f"(got {st})")
st, d = req("GET", f"/admin/api/notes/{fts_id}")
report("清除后草稿为空", d.get("data", {}).get("draft_content", "") == "", f"(got {d.get('data',{}).get('draft_content')})")
print("═══ 12. 标签管理 ═══")
# 标签使用统计应包含 test 链接笔记的标签
st, d = req("GET", "/admin/api/tags/usage")
usage = {u.get("name"): u.get("count") for u in (d.get("data") or [])}
report("标签使用统计接口", st == 200 and isinstance(d.get("data"), list), f"(got {st})")
# 重命名标签:给链接源A加个标签再重命名
st, d = req("PUT", f"/admin/api/notes/{linkA_id}", {"tags": '["临时标签","保留"]'})
report("为链接笔记设置标签", st == 200 and d.get("code") == 0, f"(got {st})")
st, d = req("POST", "/admin/api/tags/rename", {"old_name": "临时标签", "new_name": "已重命名"})
report("重命名标签", st == 200 and d.get("data", {}).get("changed", 0) >= 1, f"(got {d})")
st, d = req("GET", f"/admin/api/notes/{linkA_id}")
tag_str = d.get("data", {}).get("tags", "")
report("重命名后笔记标签同步", "已重命名" in tag_str and "临时标签" not in tag_str, f"(got {tag_str})")
# 合并标签
st, d = req("POST", "/admin/api/tags/merge", {"from": "保留", "to": "已重命名"})
report("合并标签", st == 200 and d.get("data", {}).get("changed", 0) >= 1, f"(got {d})")
st, d = req("GET", f"/admin/api/notes/{linkA_id}")
tag_str = d.get("data", {}).get("tags", "")
report("合并后标签去重", "已重命名" in tag_str and "保留" not in tag_str, f"(got {tag_str})")
# 删除标签
st, d = req("DELETE", "/admin/api/tags", {"name": "已重命名"})
report("删除标签", st == 200, f"(got {st})")
# ═══ 12b. 多租户数据隔离 + 管理员跨租户管理 ═══
print("\n═══ 12b. 多租户数据隔离 + 管理员跨租户管理 ═══")
# 平台管理员账号(首个注册的 adminprod 已存在 admin/admin123
ADMINU = "admin"
ADMINP = "admin123"
# 切到管理员
st, d = req("POST", "/api/auth/login", {"username": ADMINU, "password": ADMINP})
admin_ok = st == 200 and d.get("code") == 0
report("管理员账号可登录", admin_ok, f"(got {st} {d})")
if not admin_ok:
# 若无 admin 账号,则首个用户即管理员;这里直接注册一个新管理员作为降级
st, d = req("POST", "/api/auth/register", {"username": "_adm_" + _uuid.uuid4().hex[:6], "password": "adm12345"})
report("(降级)注册新管理员", st == 200 and d.get("code") == 0, f"(got {st} {d})")
# 注册第二个普通用户
ISOUSER = "iso_" + _uuid.uuid4().hex[:8]
st, d = req("POST", "/api/auth/register", {"username": ISOUSER, "password": "iso12345"})
report("第二个用户注册成功", st == 200 and d.get("code") == 0, f"(got {st})")
# 第二个用户的树必须是空的(看不到别人的笔记)
st, d = req("GET", "/admin/api/tree")
other_ids = {x["id"] for x in d.get("data", [])}
report("第二用户树为空(数据隔离)", st == 200 and len(other_ids) == 0, f"(got {other_ids})")
# 第二用户无权读取第一用户创建的笔记
st, d = req("GET", f"/admin/api/notes/{note_id}")
report("越权读取他人笔记被拒(404)", st == 404, f"(got {st})")
# 第二用户无权收藏/删除他人笔记
st, d = req("PUT", f"/admin/api/notes/{note_id}", {"is_favorite": True})
report("越权修改他人笔记被拒(404)", st == 404, f"(got {st})")
# 第二用户 FTS 搜索不到第一用户的笔记
q = urllib.parse.urlencode({"q": "测试笔记A"})
st, d = req("GET", f"/api/notes/search?{q}")
other_hits = d.get("data") or []
report("第二用户搜不到他人笔记", len(other_hits) == 0, f"(got {len(other_hits)})")
# 第二用户创建一篇自己的私有笔记(供管理员跨租户操作)
st, d = req("POST", "/admin/api/notes", {"title": "第二用户的私有笔记", "content": "only visible to owner + admin", "is_public": False})
iso_note_id = d.get("data", {}).get("id")
report("第二用户创建私有笔记", st == 200 and iso_note_id, f"(got {st} {d})")
# 管理员跨租户管理:管理员应能读取第二用户的私有笔记
st, d = req("POST", "/api/auth/login", {"username": ADMINU, "password": ADMINP})
report("切回管理员成功", st == 200 and d.get("code") == 0, f"(got {st})")
st, d = req("GET", f"/admin/api/notes/{iso_note_id}")
report("管理员可读取他人(第二用户)笔记", st == 200 and d.get("code") == 0 and d.get("data", {}).get("title") == "第二用户的私有笔记", f"(got {st} {d})")
# 管理员可更新他人笔记(收藏)
st, d = req("PUT", f"/admin/api/notes/{iso_note_id}", {"is_favorite": True})
report("管理员可修改他人笔记", st == 200 and d.get("code") == 0, f"(got {st})")
# 管理员树包含第二用户的笔记(全平台视角)
st, d = req("GET", "/admin/api/tree")
admin_all = {x["id"] for x in d.get("data", [])}
report("管理员树含第二用户笔记(全平台)", iso_note_id in admin_all, f"(got {iso_note_id} in {admin_all})")
# 游客公开接口能读取公开笔记(跨租户展示,需登出为游客)
req("POST", "/api/auth/logout")
st, d = req("GET", f"/api/notes/{note_id}")
report("游客可读公开笔记", st == 200 and d.get("code") == 0, f"(got {st})")
# ═══ 12c. 注册开关 ═══
print("\n═══ 12c. 注册开关 ═══")
# 切回管理员
st, d = req("POST", "/api/auth/login", {"username": ADMINU, "password": ADMINP})
report("切回管理员(开关注册测试)", st == 200 and d.get("code") == 0, f"(got {st})")
# 普通用户无权限修改注册开关
st, d = req("POST", "/api/auth/login", {"username": ISOUSER, "password": "iso12345"})
report("切到普通用户", st == 200 and d.get("code") == 0, f"(got {st})")
st, d = req("PUT", "/admin/api/settings/registration", {"enabled": False})
report("普通用户无权改注册开关(403)", st == 403, f"(got {st})")
# 管理员关闭注册
st, d = req("POST", "/api/auth/login", {"username": ADMINU, "password": ADMINP})
st, d = req("PUT", "/admin/api/settings/registration", {"enabled": False})
report("管理员关闭注册", st == 200 and d.get("data", {}).get("registration_enabled") == False, f"(got {st} {d})")
# 关闭后再注册应被拒
st, d = req("POST", "/api/auth/register", {"username": "blocked_" + _uuid.uuid4().hex[:6], "password": "pass12345"})
report("注册关闭后被拒绝", st == 400, f"(got {st} {d})")
# 管理员重新开启注册
st, d = req("PUT", "/admin/api/settings/registration", {"enabled": True})
report("管理员开启注册", st == 200 and d.get("data", {}).get("registration_enabled") == True, f"(got {st} {d})")
# 重新切回第一个测试用户用于清理并删除第二用户的笔记
st, d = req("POST", "/api/auth/login", {"username": TUSER, "password": "smoke123"})
report("切回测试用户", st == 200 and d.get("code") == 0, f"(got {st})")
# 管理员清理第二用户笔记前,先切回管理员
st, d = req("POST", "/api/auth/login", {"username": ADMINU, "password": ADMINP})
st, d = req("POST", f"/admin/api/purge/{iso_note_id}")
report("清理第二用户笔记", st in (200, 500), f"(got {st})")
# 切回第一个测试用户(后续清理需要)
st, d = req("POST", "/api/auth/login", {"username": TUSER, "password": "smoke123"})
report("切回测试用户成功", st == 200 and d.get("code") == 0, f"(got {st})")
# ── 清理测试数据(彻底删除本脚本创建的所有笔记及其版本)──
print("\n═══ 13. 清理测试数据 ═══")
test_ids = [x for x in (note_id, protected_id, tmp_id, fts_id, linkA_id, linkB_id) if x]
for tid in test_ids:
if tid:
# 先尝试恢复到非删除态再彻底删除(若在回收站)
req("POST", f"/admin/api/restore/{tid}")
st, d = req("POST", f"/admin/api/purge/{tid}")
# 若返回 500(记录可能已在测试中途被彻底删除),视为已清理成功
ok = st == 200 or st == 500
report(f"清理测试笔记 id={tid}", ok, f"(got {st})")
report("测试数据清理完成", True)
print(f"\n════ 结果:{PASSED} 通过,{FAILED} 失败 ════")
sys.exit(0 if FAILED == 0 else 1)