tamakyi
12c4f3e1be
「传上去」之后还得知道「传得对不对」：同一 slug 传过三次、换过 slug 旧图还在、 重新导出后远端仍是旧版，这些在页面上都看不出来。 - serve.py：s3_request 支持 query；新增 ListObjectsV2（含分页）与 DeleteObject， 端点 /api/s3/list、/api/s3/delete。 - index.html：远端素材清单按对象键名跟本地比 —— 一致 / 旧图（同名但大小不同）/ 多余（当前主题已不引用）/ 缺失。「补齐差异」只传缺的与旧图，「删除多余」批量删， 单行可 ↻ 覆盖或 ✕ 删除；删除走 confirm 二次确认。改过 slug 会提示重新检测。 - 修：list 返回完整键而 upload/delete 收相对键，不剥前缀会删到不存在的键 （S3 回 204，看起来删成功其实没删）。加 relKey() 统一。 - check_sigv4.py 补带 query 的用例（ListObjectsV2 三参数、乱序 token）→ 8/8； selftest_upload.py 补列表/分页/幂等删除 → 20 项全过。
142 lines
6.4 KiB
Python
142 lines
6.4 KiB
Python
"""上传链路自测：起一个假 S3，跑一遍「测试连接 → 上传 → 读回」。
|
||
|
||
覆盖 serve.py 里真正容易错、又不能靠对拍发现的部分：
|
||
· 桶级 HEAD 与对象级 PUT/GET/DELETE 的路径拼接（path style / 虚拟主机）
|
||
· 前缀拼接、Content-Type、immutable 缓存头
|
||
· 4xx 时能否把 S3 的 <Code>/<Message> 还原成人话（含 hint）
|
||
签名数学由 check_sigv4.py 与官方 SDK 对拍保证，这里不重复。
|
||
|
||
用法（项目根）：
|
||
python dev/theme-builder/selftest_upload.py
|
||
"""
|
||
import base64
|
||
import os
|
||
import sys
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, HERE)
|
||
|
||
import serve # noqa: E402
|
||
from mock_s3 import BUCKET, MockS3 # noqa: E402
|
||
|
||
|
||
def spin():
|
||
import mock_s3
|
||
return mock_s3.spin()
|
||
|
||
|
||
FAILS = []
|
||
|
||
|
||
def check(desc, cond, extra=""):
|
||
print(("OK " if cond else "FAIL ") + desc + ((" ← " + extra) if extra and not cond else ""))
|
||
if not cond:
|
||
FAILS.append(desc)
|
||
|
||
|
||
def main():
|
||
srv, endpoint = spin()
|
||
base = dict(endpoint=endpoint, bucket=BUCKET, prefix="themes", region="us-east-1",
|
||
accessKey="AKID", secretKey="SECRET")
|
||
|
||
# 1) 测试连接：4 步应全过，探针最后被删掉
|
||
cfg = serve.S3Config(dict(base))
|
||
res = serve.test_connection(cfg)
|
||
check("测试连接返回 ok", res.get("ok") is True)
|
||
check("测试连接 4 步全过", all(s["ok"] for s in res["steps"]), str(res))
|
||
check("探针已清理", not any(k.startswith("themes/.tb-probe") for k in MockS3.store),
|
||
str(list(MockS3.store)))
|
||
|
||
# 2) 上传：前缀 / 内容 / 头都要对
|
||
f1 = b"\x52\x49\x46\x46fakewebp1"
|
||
f2 = b"\x52\x49\x46\x46fakewebp2"
|
||
up = serve.upload_files(cfg, [
|
||
{"key": "shiroko/asset1.webp", "contentType": "image/webp",
|
||
"data": base64.b64encode(f1).decode()},
|
||
{"key": "shiroko/asset2.webp", "contentType": "image/webp",
|
||
"data": base64.b64encode(f2).decode()},
|
||
])
|
||
check("上传全部成功", up["ok"] and up["total"] == 2 and up["failed"] == 0, str(up))
|
||
check("前缀拼进对象键", "themes/shiroko/asset1.webp" in MockS3.store, str(list(MockS3.store)))
|
||
stored, ctype, cache = MockS3.store.get("themes/shiroko/asset1.webp", (b"", None, None))
|
||
check("内容一致", stored == f1)
|
||
check("Content-Type = image/webp", ctype == "image/webp", str(ctype))
|
||
check("Cache-Control = immutable", "immutable" in (cache or ""), str(cache))
|
||
|
||
# 3) 错误密钥：HEAD 拿不到响应体（规范如此），只能按状态码归因；
|
||
# 真正能读到 S3 错误码的是带 body 的 PUT，所以用上传路径验证还原能力。
|
||
bad = serve.S3Config(dict(base, accessKey="BADKEY"))
|
||
try:
|
||
serve.test_connection(bad)
|
||
check("错误密钥应抛错", False)
|
||
except serve.S3Error as e:
|
||
check("HEAD 失败按状态码归因", e.status == 403 and e.code == "AccessDenied",
|
||
"%s %s" % (e.status, e.code))
|
||
|
||
badUp = serve.upload_files(bad, [{"key": "shiroko/x.webp", "contentType": "image/webp",
|
||
"data": base64.b64encode(b"x").decode()}])
|
||
codes = [f.get("code") for f in badUp["files"]]
|
||
check("PUT 失败能还原 SignatureDoesNotMatch", codes == ["SignatureDoesNotMatch"], str(codes))
|
||
check("该错误有中文提示", "签名不匹配" in serve.S3_HINTS.get("SignatureDoesNotMatch", ""))
|
||
|
||
# 4) 桶名写错：NoSuchBucket
|
||
wrong = serve.S3Config(dict(base, bucket="nope"))
|
||
try:
|
||
serve.test_connection(wrong)
|
||
check("错桶应抛错", False)
|
||
except serve.S3Error as e:
|
||
check("错桶 → NoSuchBucket", e.code == "NoSuchBucket", e.code)
|
||
|
||
# 5) 键里出现 .. 必须被拒（防止越权写到别的前缀）
|
||
try:
|
||
serve.full_key(cfg, "../etc/passwd")
|
||
check(".. 应被拒", False)
|
||
except ValueError:
|
||
check(".. 被拒绝", True)
|
||
|
||
# 6) 前缀留空 = 传桶根
|
||
plain = serve.S3Config(dict(base, prefix=""))
|
||
up2 = serve.upload_files(plain, [{"key": "root.webp", "contentType": "image/webp",
|
||
"data": base64.b64encode(b"x").decode()}])
|
||
check("空前缀传桶根", up2["ok"] and "root.webp" in MockS3.store, str(list(MockS3.store)))
|
||
|
||
# 7) 列远端：前缀过滤、大小、分页
|
||
serve.upload_files(cfg, [{"key": "shiroko/asset3.webp", "contentType": "image/webp",
|
||
"data": base64.b64encode(b"three").decode()}])
|
||
ls = serve.list_objects(cfg, "shiroko/")
|
||
keys = sorted(o["key"] for o in ls["objects"])
|
||
check("按 <前缀>/<slug>/ 列出 3 个对象",
|
||
keys == ["themes/shiroko/asset1.webp", "themes/shiroko/asset2.webp",
|
||
"themes/shiroko/asset3.webp"], str(keys))
|
||
size1 = [o["size"] for o in ls["objects"] if o["key"].endswith("asset1.webp")]
|
||
check("对象大小与上传字节一致", size1 == [len(f1)], str(size1))
|
||
paged = serve.list_objects(cfg, "shiroko/", max_keys=2)
|
||
check("分页把 3 个都取回来了", len(paged["objects"]) == 3, str(len(paged["objects"])))
|
||
everything = serve.list_objects(cfg, "")
|
||
# 空前缀 = 列「路径前缀」之下（跨主题），不扫整个桶：桶里还可能有头像等别的东西。
|
||
check("空前缀列出 <路径前缀>/ 下全部、且不越界",
|
||
all(o["key"].startswith("themes/") for o in everything["objects"])
|
||
and len(everything["objects"]) == 3,
|
||
str([o["key"] for o in everything["objects"]]))
|
||
|
||
# 8) 删除：幂等（不存在的键也当成功），且不影响别的前缀
|
||
del_res = serve.delete_objects(cfg, ["shiroko/asset3.webp", "shiroko/nonexistent.webp"])
|
||
check("删除含不存在的键仍算成功", del_res["ok"] and del_res["total"] == 2, str(del_res))
|
||
after = serve.list_objects(cfg, "shiroko/")
|
||
check("删完只剩 2 个", len(after["objects"]) == 2,
|
||
str([o["key"] for o in after["objects"]]))
|
||
check("桶根对象没被误删", "root.webp" in MockS3.store)
|
||
|
||
# 9) 虚拟主机风格的路径（不带桶名）
|
||
vh = serve.S3Config(dict(base, pathStyle=False))
|
||
check("虚拟主机路径不含桶名", vh.object_path("themes/a.webp") == "/themes/a.webp", vh.object_path("x"))
|
||
check("虚拟主机 host 带桶名", vh.host.startswith(BUCKET + "."), vh.host)
|
||
|
||
srv.shutdown()
|
||
print("\n%d 项失败" % len(FAILS))
|
||
return 1 if FAILS else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|