chaos
55f56d27c9
Prefer Cloudflare Worker KV for verification codes with optional short IMAP fallback disabled for private deploy; track config.json/.env for private repo. Also add worker project, protocol mint backoff, and fail-fast Turnstile/mail timeouts from batch runs.
83 lines
2.8 KiB
Python
Executable File
83 lines
2.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Test Cloudflare Email Worker code API (and optional local hybrid path)."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from urllib.parse import urlencode
|
|
from urllib.request import Request, urlopen
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def load_cfg():
|
|
p = ROOT / "config.json"
|
|
if p.is_file():
|
|
return json.loads(p.read_text(encoding="utf-8"))
|
|
return {}
|
|
|
|
|
|
def main():
|
|
cfg = load_cfg()
|
|
ap = argparse.ArgumentParser(description="Test gmail Email Worker /code API")
|
|
ap.add_argument("--url", default=cfg.get("gmail_worker_url") or os.getenv("GMAIL_WORKER_URL") or "")
|
|
ap.add_argument("--key", default=cfg.get("gmail_worker_api_key") or os.getenv("GMAIL_WORKER_API_KEY") or "")
|
|
ap.add_argument("--to", required=True, help="alias@domain to query")
|
|
ap.add_argument("--timeout", type=float, default=20)
|
|
ap.add_argument("--interval", type=float, default=0.5)
|
|
ap.add_argument("--ingest-demo", action="store_true", help="POST a fake code then GET it")
|
|
ap.add_argument("--no-consume", action="store_true")
|
|
args = ap.parse_args()
|
|
base = (args.url or "").rstrip("/")
|
|
key = args.key or ""
|
|
if not base or not key:
|
|
print("需要 --url/--key 或 config.json 的 gmail_worker_url / gmail_worker_api_key")
|
|
return 1
|
|
|
|
def get(path, params=None):
|
|
q = dict(params or {})
|
|
q["key"] = key
|
|
url = f"{base}{path}?{urlencode(q)}"
|
|
req = Request(url, headers={"x-api-key": key})
|
|
with urlopen(req, timeout=10) as r:
|
|
return r.status, json.loads(r.read().decode())
|
|
|
|
def post(path, body):
|
|
url = f"{base}{path}?{urlencode({'key': key})}"
|
|
data = json.dumps(body).encode()
|
|
req = Request(url, data=data, headers={"content-type": "application/json", "x-api-key": key}, method="POST")
|
|
with urlopen(req, timeout=10) as r:
|
|
return r.status, json.loads(r.read().decode())
|
|
|
|
st, health = get("/health")
|
|
print("health", st, health)
|
|
|
|
if args.ingest_demo:
|
|
code = "AB1-CD2"
|
|
st, body = post("/ingest", {"to": args.to, "code": code, "subject": f"{code} xAI confirmation code"})
|
|
print("ingest", st, body)
|
|
|
|
deadline = time.time() + args.timeout
|
|
while time.time() < deadline:
|
|
try:
|
|
st, body = get(
|
|
"/code",
|
|
{"to": args.to.lower(), "consume": "0" if args.no_consume else "1"},
|
|
)
|
|
print("code", st, body)
|
|
if st == 200 and body.get("code"):
|
|
return 0
|
|
except Exception as e:
|
|
print("wait", e)
|
|
time.sleep(args.interval)
|
|
print("timeout — 若是真实邮件，检查 Email Routing 是否绑到 Worker")
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|