Your Name
d9793300f9
- 新增 users 表(user_id 数据隔离,bcrypt 密码) - 认证: 注册/登录(用户名+密码)/会话绑定用户, 首个用户成为管理员并接管旧数据 - 数据隔离: 笔记/分类/标签/回收站/版本/草稿/图谱/FTS 全部按用户隔离 - 前台: 登录/注册弹窗, 登录后★收藏自己的笔记, 游客只读公开笔记 - 后台: 用户名+密码登录, 每人管理自己的工作区, 越权访问返回404 - 冒烟测试重构+新增多租户隔离用例(78/78)
418 lines
18 KiB
Python
418 lines
18 KiB
Python
#!/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})")
|
|
|
|
# ── 多租户数据隔离 ──
|
|
print("\n═══ 12b. 多租户数据隔离 ═══")
|
|
# 记住当前测试用户可见的笔记（应只有本脚本创建的 + 自己空间的）
|
|
st, d = req("GET", "/admin/api/tree")
|
|
own_ids = {x["id"] for x in d.get("data", [])}
|
|
|
|
# 注册第二个用户
|
|
st, d = req("POST", "/api/auth/register", {"username": "iso_" + _uuid.uuid4().hex[:8], "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})")
|
|
|
|
# 两用户笔记 ID 无交集
|
|
report("两用户数据无交集", len(own_ids & other_ids) == 0, f"(own={own_ids} other={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)})")
|
|
|
|
# 游客公开接口能读取公开笔记（跨租户展示，需登出为游客）
|
|
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})")
|
|
|
|
# 重新登录回第一个测试用户（后续清理需要）
|
|
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)
|