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