diary-news/scripts/smoke_ingest.py

"""端到端 smoke:用真实 ingest token 推一条短新闻,验证完整链路。

用法:
  python scripts/smoke_ingest.py --token <RAW_TOKEN> [--host <host>] [--port <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())