Your Name
261 lines
10 KiB
Python
261 lines
10 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", "password=wrongpass")
|
|
report("错误密码登录返回401", st == 401, f"(got {st})")
|
|
|
|
# 正确登录
|
|
st, d = req("POST", "/admin/login", "password=admin123")
|
|
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})")
|
|
|
|
# 旧固定 cookie 不再有效
|
|
class TmpOpener:
|
|
pass
|
|
# 手动构造请求带旧的 fixed cookie
|
|
import urllib.request as u
|
|
reqf = urllib.request.Request(BASE + "/admin/api/tree", headers={"Cookie": "admin_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 访问 -> 应被拒
|
|
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})")
|
|
|
|
# 公开无密码笔记
|
|
public_ok = False
|
|
st, d = req("GET", f"/api/notes/{note_id}")
|
|
report("公开接口读取无密码笔记成功", st == 200 and d.get("code") == 0, f"(got {st})")
|
|
|
|
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("\n═══ 9. 清理测试数据 ═══")
|
|
test_ids = [x for x in (note_id, protected_id, tmp_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)
|