tamabox/dev/theme-builder/selftest_upload.py

"""上传链路自测：起一个假 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))

    # 2b) 逐文件覆盖缓存头：README/清单这类「键不变、内容会变」的文件不能 immutable
    serve.upload_files(cfg, [
        {"key": "shiroko/INSTALL.txt", "contentType": "text/plain; charset=utf-8",
         "cacheControl": "public, max-age=60",
         "data": base64.b64encode(b"readme").decode()},
    ])
    _, _, cc = MockS3.store.get("themes/shiroko/INSTALL.txt", (b"", None, None))
    check("可覆盖 Cache-Control", cc == "public, max-age=60", str(cc))
    # 这一步只为验头，别让它污染后面的对象计数
    MockS3.store.pop("themes/shiroko/INSTALL.txt", None)

    # 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())