init

scripts/longrun_test.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import argparse
+import json
+import ssl
+import time
+import urllib.request
+
+
+def _json_request(url: str, method: str = "GET", payload: dict | None = None, timeout: int = 30) -> dict:
+    body = None if payload is None else json.dumps(payload, ensure_ascii=False).encode("utf-8")
+    req = urllib.request.Request(url=url, method=method, data=body)
+    if payload is not None:
+        req.add_header("Content-Type", "application/json")
+    ctx = ssl._create_unverified_context() if url.startswith("https://") else None
+    with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:
+        return json.loads(resp.read().decode("utf-8"))
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Long-run stability test")
+    parser.add_argument("--base-url", default="https://127.0.0.1:8080")
+    parser.add_argument("--minutes", type=int, default=60)
+    parser.add_argument("--interval-sec", type=int, default=30)
+    args = parser.parse_args()
+
+    end_ts = time.time() + args.minutes * 60
+    rounds = 0
+    while time.time() < end_ts:
+        rounds += 1
+        data = _json_request(
+            f"{args.base_url}/chat/text",
+            method="POST",
+            payload={"text": f"长稳测试第{rounds}轮，请简短回复。"},
+            timeout=60,
+        )
+        health = _json_request(f"{args.base_url}/health")
+        print(
+            f"round={rounds} ok={data.get('ok')} "
+            f"last_latency={health.get('last_latency_ms')} "
+            f"llm={health.get('last_llm_latency_ms')} "
+            f"tts={health.get('last_tts_latency_ms')}"
+        )
+        time.sleep(args.interval_sec)
+
+    print("longrun done, rounds:", rounds)
+
+
+if __name__ == "__main__":
+    main()