From f5fcde11532efe357340ec52021210d46c9d81aa Mon Sep 17 00:00:00 2001 From: xiaji Date: Sun, 14 Jun 2026 20:30:35 +0800 Subject: [PATCH] =?UTF-8?q?test(ingest):=20=E7=AB=AF=E5=88=B0=E7=AB=AF=20s?= =?UTF-8?q?moke=20=E8=84=9A=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit scripts/smoke_ingest.py — 验证 /api/v1/ingest 完整链路,6 步: 1) 新建 → 201 created 2) external_id 重复 → 200 duplicate reason=external_id_match 3) 不带 external_id 新建 → 201 created 4) 内容指纹重复 → 200 duplicate reason=content_hash_match 5) 错误 token → 401 6) body 超 5000 字 → 400 / 422 用法:python scripts/smoke_ingest.py --token 默认 host=207.57.129.228 port=19717(直连服务器), 生产走 caddy 时:--scheme https --host your-domain --port 443 需要 owner 在 web 端生成 ingest token(API Push 源的 🔑 Token 按钮)。 smoke 数据会留在 Feed 里(source_ref='smoke', tags 含 'smoke'), 如果想清理:DELETE FROM articles WHERE source_ref='smoke'; 或用 source_ref 过滤后手动删。 --- scripts/smoke_ingest.py | 132 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 132 insertions(+) create mode 100644 scripts/smoke_ingest.py diff --git a/scripts/smoke_ingest.py b/scripts/smoke_ingest.py new file mode 100644 index 0000000..e46ca41 --- /dev/null +++ b/scripts/smoke_ingest.py @@ -0,0 +1,132 @@ +"""端到端 smoke:用真实 ingest token 推一条短新闻,验证完整链路。 + +用法: + python scripts/smoke_ingest.py --token [--host ] [--port ] + +行为: + 1) POST /api/v1/ingest 推第一条 → 期望 201 created + 2) 立即再推同样内容 → 期望 200 duplicate reason=external_id_match + 3) 不带 external_id 推不同内容 → 期望 201 created + 4) 不带 external_id 重复推同样内容 → 期望 200 duplicate reason=content_hash_match + 5) 不带 token → 期望 401 + 6) body 超 5000 字 → 期望 400 / 422 + +输出 PASS / FAIL,任何一步失败非零退出。 +""" +from __future__ import annotations + +import argparse +import json +import sys +import time +from urllib.parse import urljoin + +import httpx + + +def _post(client: httpx.Client, url: str, headers: dict, body: dict, expected: tuple[int, ...]): + """发起 POST 并断言状态码 + body 关键字段。返回响应对象。""" + r = client.post(url, headers=headers, json=body, timeout=15.0) + status = r.status_code + try: + data = r.json() + except Exception: + data = {"raw": r.text[:200]} + assert status in expected, f"期望 {expected} 实际 {status}: {json.dumps(data, ensure_ascii=False)[:300]}" + return status, data + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--token", required=True, help="X-Ingest-Token 的 raw 值") + ap.add_argument("--host", default="207.57.129.228") + ap.add_argument("--port", type=int, default=19717) + ap.add_argument("--scheme", default="http", help="http / https(走 caddy 时用 https + 域名)") + args = ap.parse_args() + + base = f"{args.scheme}://{args.host}:{args.port}" if args.port not in (80, 443) else f"{args.scheme}://{args.host}" + url = urljoin(base + "/", "api/v1/ingest") + auth = {"X-Ingest-Token": args.token, "Content-Type": "application/json"} + + ext_id = f"smoke-{int(time.time())}" + payload1 = { + "external_id": ext_id, + "title": f"smoke test {ext_id}", + "body": "smoke test body — 端到端链路验证", + "source_ref": "smoke", + "tags": ["测试", "smoke"], + } + + fail = 0 + with httpx.Client() as client: + # 1) 新建 + try: + s, d = _post(client, url, auth, payload1, (201,)) + assert d.get("status") == "created", f"期望 status=created, 实际 {d}" + print(f"[1/6] PASS created article_id={d['article_id']} hash={d['content_hash'][:12]}") + except AssertionError as e: + print(f"[1/6] FAIL created {e}") + fail += 1 + + # 2) external_id 重复 + try: + s, d = _post(client, url, auth, payload1, (200,)) + assert d.get("status") == "duplicate" and d.get("reason") == "external_id_match", \ + f"期望 external_id_match duplicate, 实际 {d}" + print(f"[2/6] PASS external_id_match article_id={d['article_id']} ext={d.get('matched_external_id')}") + except AssertionError as e: + print(f"[2/6] FAIL external_id_match {e}") + fail += 1 + + # 3) 不带 external_id 新建 + payload3 = { + "title": f"smoke no-ext {int(time.time())}", + "body": f"unique body {time.time_ns()}", + "source_ref": "smoke", + } + try: + s, d = _post(client, url, auth, payload3, (201,)) + assert d.get("status") == "created" + print(f"[3/6] PASS no-ext created article_id={d['article_id']}") + except AssertionError as e: + print(f"[3/6] FAIL no-ext created {e}") + fail += 1 + + # 4) 内容重复(同 payload3) + try: + s, d = _post(client, url, auth, payload3, (200,)) + assert d.get("status") == "duplicate" and d.get("reason") == "content_hash_match", \ + f"期望 content_hash_match duplicate, 实际 {d}" + print(f"[4/6] PASS content_hash_match article_id={d['article_id']}") + except AssertionError as e: + print(f"[4/6] FAIL content_hash_match {e}") + fail += 1 + + # 5) 错误 token + try: + s, d = _post(client, url, {"X-Ingest-Token": "definitely-wrong", "Content-Type": "application/json"}, + payload1, (401,)) + print(f"[5/6] PASS bad-token 401") + except AssertionError as e: + print(f"[5/6] FAIL bad-token {e}") + fail += 1 + + # 6) body 超长 + try: + bad = {"title": "x", "body": "x" * 5001} + s, d = _post(client, url, auth, bad, (400, 422)) + print(f"[6/6] PASS body-too-long status={s}") + except AssertionError as e: + print(f"[6/6] FAIL body-too-long {e}") + fail += 1 + + print(f"\n{'-' * 40}") + if fail: + print(f"FAILED ({fail}/6)") + return 1 + print("ALL PASS (6/6)") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file