From aa1114338e25f313a0dedfa30f3ed30dfee92f7b Mon Sep 17 00:00:00 2001 From: WorkBuddy Date: Tue, 22 Sep 2026 12:26:37 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E5=AF=B9=E8=B1=A1=E5=AD=98?= =?UTF-8?q?=E5=82=A8=E7=9C=9F=E5=AE=9E=E9=93=BE=E8=B7=AF=E6=8E=A2=E9=92=88?= =?UTF-8?q?=EF=BC=88=E5=88=A4=E6=8D=AE=EF=BC=8C=E9=9D=9E=20pytest=20?= =?UTF-8?q?=E7=94=A8=E4=BE=8B=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 背景：E2E 全套 65 例对对象存储零覆盖，而 generatePresignedUrl 只是本地 签名、不访问存储，因此"签名接口返回 200"无法证明对象真能存进去。 实证（2026-09-22）：MinIO 的 /data 下只有 .minio.sys，无任何 bucket； mc cp 到 private 报 `Bucket 'private' does not exist`。全仓库零建桶代码、 零初始化脚本 —— 真实上传链路在当前配置下必然失败，而所有测试仍全绿。 本探针覆盖 PhotoService 的真实路径并要求字节级一致：服务端直传、 presigned PUT（客户端直传）、presigned GET，并同时验证 path / virtual-host 两种 addressing style。支持 --no-create-bucket 负向模式，用于证明判据 有判别力而非恒绿。 不在 CI 运行：依赖 boto3 + requests，不进 tests/e2e/requirements.txt。 --- tests/e2e/storage_link_probe.py | 148 ++++++++++++++++++++++++++++++++ 1 file changed, 148 insertions(+) create mode 100644 tests/e2e/storage_link_probe.py diff --git a/tests/e2e/storage_link_probe.py b/tests/e2e/storage_link_probe.py new file mode 100644 index 0000000..94045a8 --- /dev/null +++ b/tests/e2e/storage_link_probe.py @@ -0,0 +1,148 @@ +"""对象存储真实链路探针（手工/CI 外验证工具，非 pytest 用例）。 + +为什么需要它 +------------ +本项目 E2E 全套对对象存储**零覆盖**（2026-09-22 核实：65 例中无任何一例 +触及 photo / upload / presigned）。而 `generatePresignedUrl` 只是本地签名、 +不访问存储，因此"签名接口返回 200"完全不能证明对象真的能存进去。 + +这直接导致一类缺陷长期隐身：bucket 从未被任何代码或脚本创建过 +（2026-09-22 实证：MinIO /data 下只有 .minio.sys，mc cp 报 +`Bucket 'private' does not exist`），于是 `putObject` 必然失败，但所有 +测试仍然全绿。 + +本探针覆盖 PhotoService 的真实调用路径，并要求**字节级一致**： + A. 服务端直传 <- amazonS3.putObject + B. presigned PUT <- getUploadToken 下发的 uploadUrl（客户端直传） + C. presigned GET <- getSignedUrl 下发的展示 URL + D. addressing style <- path / virtual-host（后端 AWS SDK v1 未显式设置） + +用法（需 boto3 + requests，不在 CI 依赖里，故不在 CI 运行） +------------------------------------------------------- + # 正向：验证链路可用 + python tests/e2e/storage_link_probe.py --endpoint http://localhost:9003 \ + --ak --sk + + # 负向：不预建桶，应当失败 —— 用于证明本探针确实有判别力， + # 而不是恒绿。改动存储配置后务必先跑这一遍。 + python tests/e2e/storage_link_probe.py --endpoint http://localhost:9003 \ + --ak --sk --no-create-bucket +""" +import argparse +import sys + +import boto3 +import requests +from botocore.config import Config +from botocore.exceptions import ClientError + +PAYLOAD = b"danyang-storage-probe-" + b"x" * 512 +KEY_DIRECT = "probe/direct.bin" +KEY_SIGNED = "probe/signed.bin" + + +def build(endpoint, ak, sk, style): + return boto3.client( + "s3", + endpoint_url=endpoint, + aws_access_key_id=ak, + aws_secret_access_key=sk, + region_name="us-east-1", + config=Config(signature_version="s3v4", s3={"addressing_style": style}), + ) + + +def bucket_exists(s3, bucket): + try: + s3.head_bucket(Bucket=bucket) + return True, "exists" + except ClientError as e: + return False, e.response.get("Error", {}).get("Code", "?") + except Exception as e: # noqa: BLE001 - 探针需容忍任意后端异常 + return False, type(e).__name__ + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--endpoint", required=True) + ap.add_argument("--ak", required=True) + ap.add_argument("--sk", required=True) + ap.add_argument("--bucket", default="private") + ap.add_argument("--no-create-bucket", action="store_true", + help="不预建桶，用于负向验证（判据应当失败）") + a = ap.parse_args() + + failures = [] + print(f"===== 对象存储链路探针 @ {a.endpoint} =====") + + for style in ("path", "virtual"): + print(f"\n--- addressing_style={style} ---") + s3 = build(a.endpoint, a.ak, a.sk, style) + ok, detail = bucket_exists(s3, a.bucket) + + if not ok: + if a.no_create_bucket: + print(f" bucket: MISSING ({detail}) -> 负向模式，跳过写入") + failures.append(f"[{style}] bucket missing") + continue + try: + s3.create_bucket(Bucket=a.bucket) + print(f" bucket: 已创建 (was {detail})") + except Exception as e: # noqa: BLE001 + print(f" bucket: 创建失败 {type(e).__name__}: {str(e)[:120]}") + failures.append(f"[{style}] create_bucket") + continue + else: + print(f" bucket: {detail}") + + # A 服务端直传 + try: + s3.put_object(Bucket=a.bucket, Key=KEY_DIRECT, Body=PAYLOAD) + print(" A put_object : OK") + except Exception as e: # noqa: BLE001 + print(f" A put_object : FAIL {type(e).__name__}: {str(e)[:110]}") + failures.append(f"[{style}] put_object") + continue + + # B presigned PUT（客户端直传） + try: + url = s3.generate_presigned_url( + "put_object", Params={"Bucket": a.bucket, "Key": KEY_SIGNED}, ExpiresIn=600 + ) + r = requests.put(url, data=PAYLOAD, timeout=30) + if r.status_code not in (200, 201): + print(f" B presigned PUT : FAIL http={r.status_code} {r.text[:110]}") + failures.append(f"[{style}] presigned PUT") + continue + print(f" B presigned PUT : OK ({r.status_code})") + except Exception as e: # noqa: BLE001 + print(f" B presigned PUT : FAIL {type(e).__name__}: {str(e)[:110]}") + failures.append(f"[{style}] presigned PUT") + continue + + # C presigned GET（字节级比对） + try: + url = s3.generate_presigned_url( + "get_object", Params={"Bucket": a.bucket, "Key": KEY_SIGNED}, ExpiresIn=600 + ) + r = requests.get(url, timeout=30) + if r.status_code == 200 and r.content == PAYLOAD: + print(f" C presigned GET : OK ({len(r.content)} bytes, 字节一致)") + else: + print(f" C presigned GET : FAIL http={r.status_code} bytes={len(r.content)}") + failures.append(f"[{style}] presigned GET") + except Exception as e: # noqa: BLE001 + print(f" C presigned GET : FAIL {type(e).__name__}: {str(e)[:110]}") + failures.append(f"[{style}] presigned GET") + + print("\n===== 汇总 =====") + print(f"failures: {len(failures)}") + for f in failures: + print(" -", f) + if failures and a.no_create_bucket: + print("（负向模式：有 failures 才是正确结果，说明判据有判别力）") + return 0 if not failures else 1 + + +if __name__ == "__main__": + sys.exit(main())