"""WebUI settings and streaming against fake OpenAI-compatible APIs. python3 scripts/e2e_webui_settings.py (CST_AGENT= picks the binary) Checks: Chinese reasoning streams as whole characters in few events; the provider list, the model list with capabilities, config validation and persistence; a wrong key is rejected, a good one switches the provider (requests then go to the new base URL) and is saved; and all of it survives a restart. HOME and XDG_CONFIG_HOME point at a temp dir, so no real credentials or settings are read or written. """ import json, os, shutil, stat, subprocess, sys, tempfile, threading, time, urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer BIN = os.environ.get("CST_AGENT", os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "cst-agent")) THINK = ["好的，我先看看工作区里", "有哪些文件，然后用中文", "回答这个问题。😀 结束"] ANSWER = "这是中文回答。" GOOD_KEYS = {"sk-good", "sk-good2"} B_REQS = [] def sse_reply(h, model, think, answer): h.send_response(200) h.send_header("Content-Type", "text/event-stream") h.end_headers() base = {"id": "x", "object": "chat.completion.chunk", "model": model} def ev(o): h.wfile.write(b"data: " + json.dumps({**base, **o}, ensure_ascii=False).encode() + b"\n\n") h.wfile.flush() for part in think: ev({"choices": [{"index": 0, "delta": {"role": "assistant", "reasoning_content": part}}]}) ev({"choices": [{"index": 0, "delta": {"content": answer}}]}) ev({"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]}) h.wfile.write(b"data: [DONE]\n\n") h.wfile.flush() def send_json(h, code, obj): body = json.dumps(obj).encode() h.send_response(code) h.send_header("Content-Type", "application/json") h.send_header("Content-Length", str(len(body))) h.end_headers() h.wfile.write(body) class A(BaseHTTPRequestHandler): """The starting provider: any key, one model, Chinese reasoning.""" def log_message(self, *a): pass def do_GET(self): send_json(self, 200, {"data": [{"id": "deepseek-flash"}]}) def do_POST(self): self.rfile.read(int(self.headers.get("Content-Length", 0))) sse_reply(self, "deepseek-flash", THINK, ANSWER) class B(BaseHTTPRequestHandler): """The provider switched to: /models checks the key.""" def log_message(self, *a): pass def key_ok(self): auth = self.headers.get("Authorization", "") return auth.startswith("Bearer ") and auth[7:] in GOOD_KEYS def do_GET(self): if not self.key_ok(): return send_json(self, 401, {"error": {"message": "invalid api key"}}) send_json(self, 200, {"data": [{"id": "fake-vis"}, {"id": "fake-text"}]}) def do_POST(self): body = json.loads(self.rfile.read(int(self.headers.get("Content-Length", 0))) or b"{}") B_REQS.append({"auth": self.headers.get("Authorization", ""), "model": body.get("model")}) if not self.key_ok(): return send_json(self, 401, {"error": {"message": "invalid api key"}}) sse_reply(self, body.get("model") or "fake-vis", [], "B 的回答") fails = 0 def check(ok, msg): global fails print(("PASS " if ok else "FAIL ") + msg) fails += 0 if ok else 1 class Sse: def __init__(self, base): self.lines = [] self.r = urllib.request.urlopen(base + "/api/events", timeout=60) threading.Thread(target=self.pump, daemon=True).start() def pump(self): try: for raw in self.r: self.lines.append(raw) except Exception: pass def events(self, name): out = [] for i, l in enumerate(self.lines): if l.strip() == b"event: " + name.encode() and i + 1 < len(self.lines): out.append(self.lines[i + 1][6:]) return out def close(self): try: self.r.close() except Exception: pass def wait(fn, t=15): end = time.time() + t while time.time() < end: v = fn() if v: return v time.sleep(0.1) return None def call(base, path, obj=None, method=None): data = None if obj is None else json.dumps(obj).encode() req = urllib.request.Request(base + path, data=data, method=method or ("POST" if data else "GET"), headers={"Content-Type": "application/json"}) try: with urllib.request.urlopen(req, timeout=20) as r: raw = r.read() return r.status, json.loads(raw or b"{}"), raw except urllib.error.HTTPError as e: raw = e.read() try: return e.code, json.loads(raw or b"{}"), raw except ValueError: return e.code, {}, raw def serve(handler): s = ThreadingHTTPServer(("127.0.0.1", 0), handler) threading.Thread(target=s.serve_forever, daemon=True).start() return s api_a, api_b = serve(A), serve(B) ws = tempfile.mkdtemp(prefix="cst-set-") home = tempfile.mkdtemp(prefix="cst-home-") cfgdir = home + "/.config/cst-agent" os.makedirs(cfgdir) with open(cfgdir + "/models.json", "w") as f: json.dump({"providers": {"fakeb": {"api_base": "http://127.0.0.1:%d/v1" % api_b.server_address[1], "key_env": "FAKEB_API_KEY", "default_model": ""}}, "models": [{"match": "fake-vis", "vision": True, "context": 200000}, {"match": "fake-text", "vision": False}]}, f) env = dict(os.environ, HOME=home, XDG_CONFIG_HOME=home + "/.config", XDG_DATA_HOME=home + "/.local/share", XDG_CACHE_HOME=home + "/.cache") for k in ("DEEPSEEK_API_KEY", "FAKEB_API_KEY", "OPENAI_API_KEY", "CST_API_KEY"): env.pop(k, None) procs = [] def start(args, tag): log = open("%s/%s.log" % (ws, tag), "w+") p = subprocess.Popen([BIN, "--web", "--web-port", "0", "--workspace", ws] + args, stdout=log, stderr=subprocess.STDOUT, env=env) procs.append((p, log)) def port(): log.seek(0) for l in log.read().splitlines(): if l.startswith("webui: http://127.0.0.1:"): return l.split(":")[3].split("/")[0] pt = wait(port) return p, "http://127.0.0.1:%s" % pt def stop(p): p.terminate() try: p.wait(20) except subprocess.TimeoutExpired: p.kill() check(False, "server hung on SIGTERM") check(p.returncode is not None and 0 <= p.returncode < 128, "clean exit (%s)" % p.returncode) def leaked(raw): return any(k.encode() in raw for k in GOOD_KEYS | {"sk-bad", "sk-x"}) try: p, base = start(["--new", "--api-key", "sk-x", "--provider", "deepseek", "--model", "deepseek-flash", "--api-base", "http://127.0.0.1:%d/v1" % api_a.server_address[1], "--protocol", "native-eos", "--approval", "auto"], "first") a = Sse(base) time.sleep(0.3) # 1. Chinese reasoning: whole characters, a handful of events code, _, _ = call(base, "/api/turn", {"text": "用中文想一想"}) check(code == 202, "turn accepted") end = wait(lambda: a.events("turn_end"), 20) check(bool(end) and json.loads(end[0]).get("rc") == 0, "turn ended rc=0") raw = a.events("think_delta") ok = True for r in raw: try: r.decode("utf-8") except UnicodeDecodeError: ok = False check(ok, "every think_delta frame is valid UTF-8") text = "".join(json.loads(r.decode("utf-8", "replace"))["text"] for r in raw) want = "".join(THINK) check(text == want and "\ufffd" not in text, "the thinking text round trips: %r" % text[:60]) check(0 < len(raw) <= 2 * len(THINK) + 2, "think_delta events: %d for %d bytes" % (len(raw), len(want.encode()))) ans = "".join(json.loads(r)["text"] for r in a.events("assistant_delta")) check(ans == ANSWER, "the answer round trips: %r" % ans) # 2. providers and models code, d, rawp = call(base, "/api/providers") names = {x["name"]: x for x in d.get("providers", [])} check(code == 200 and "deepseek" in names and "fakeb" in names, "provider list: %s" % sorted(names)) check(names.get("deepseek", {}).get("current") and names["deepseek"].get("has_key") and names["deepseek"].get("key_from") == "session", "deepseek: current, key for this run") check(names.get("fakeb", {}).get("has_key") is False and names["fakeb"].get("source") == "user", "fakeb: no key yet, from models.json") check(not leaked(rawp), "no key in /api/providers") code, d, _ = call(base, "/api/models") ids = [m["id"] for m in d.get("models", [])] first = (d.get("models") or [{}])[0] check(code == 200 and "deepseek-flash" in ids and first.get("api") is True and "context" in first and "label" in first, "model list: %s" % ids[:5]) # 3. config: validation and persistence check(call(base, "/api/config", {"effort": "bogus"})[0] == 400, "a bad effort is a 400") check(call(base, "/api/config", {"provider": "fakeb"})[0] == 400, "provider via /api/config is a 400") check(call(base, "/api/config", {"model": ""})[0] == 400, "an empty model is a 400") code, d, _ = call(base, "/api/config", {"model": "deepseek-v4-pro", "effort": "high"}) check(code == 200 and d.get("ok"), "model + effort accepted") code, s, _ = call(base, "/api/state") check(s.get("model") == "deepseek-v4-pro" and s.get("effort") == "high", "the state has them") check(s.get("effort_applies") is True and s.get("key_set") is True and "model_info" in s and s.get("perm_timeout", 0) > 0 and "turn_usage" in s, "the state has the new fields") st = open(cfgdir + "/settings.json").read() if os.path.exists(cfgdir + "/settings.json") else "" check("deepseek-v4-pro" in st and "high" in st, "settings.json remembers model and effort") # 4. switching provider: no key, a wrong key, a good key code, d, _ = call(base, "/api/provider", {"name": "fakeb"}) check(code == 400 and d.get("need_key"), "fakeb without a key: need_key (%s)" % code) code, d, rawp = call(base, "/api/provider", {"name": "fakeb", "key": "sk-bad"}) check(code == 401 and "拒绝" in d.get("error", ""), "a wrong key is rejected (%s %s)" % (code, d.get("error"))) check(not leaked(rawp), "no key in the rejection") check(call(base, "/api/state")[1].get("provider") == "deepseek", "still on deepseek") check(call(base, "/api/provider", {"name": "nope"})[0] == 404, "an unknown provider is a 404") code, d, rawp = call(base, "/api/provider", {"name": "fakeb", "key": "sk-good"}) check(code == 200 and d.get("provider") == "fakeb" and d.get("model") == "fake-vis", "switched to fakeb, first listed model picked (%s %s)" % (code, d)) check(not leaked(rawp), "no key in the switch reply") creds = cfgdir + "/credentials" cr = open(creds).read() if os.path.exists(creds) else "" check("fakeb.api_key=sk-good" in cr, "the key was saved to credentials") if os.name == "posix" and os.path.exists(creds): check(stat.S_IMODE(os.stat(creds).st_mode) == 0o600, "credentials is 0600") code, d, _ = call(base, "/api/models") ms = {m["id"]: m for m in d.get("models", [])} check("fake-vis" in ms and ms["fake-vis"].get("vision") is True and ms["fake-vis"].get("context") == 200000 and ms.get("fake-text", {}).get("vision") is False, "fakeb models with capabilities") code, d, _ = call(base, "/api/config", {"effort": "low"}) check(code == 200 and d.get("warning"), "effort on a model without it: saved with a warning") # 5. a turn now goes to the new base URL with the new key n0 = len(B_REQS) code, _, _ = call(base, "/api/turn", {"text": "hi B"}) check(code == 202, "turn on fakeb accepted") wait(lambda: len(a.events("turn_end")) >= 2, 20) check(len(B_REQS) > n0 and B_REQS[-1]["auth"] == "Bearer sk-good" and B_REQS[-1]["model"] == "fake-vis", "the request reached fakeb: %s" % (B_REQS[-1:] or "none")) # 6. replacing the key code, d, _ = call(base, "/api/key", {"key": "sk-bad"}) check(code == 401, "a wrong replacement key is rejected (%s)" % code) code, d, _ = call(base, "/api/key", {"key": "sk-good2"}) check(code == 200 and d.get("ok"), "the replacement key is accepted") check("fakeb.api_key=sk-good2" in open(creds).read(), "the replacement key was saved") check(call(base, "/api/key", {})[0] == 400, "an empty key body is a 400") a.close() stop(p) # 7. the next start remembers provider, model and key p, base = start(["--new"], "second") code, s, raws = call(base, "/api/state") check(s.get("provider") == "fakeb" and s.get("model") == "fake-vis" and s.get("key_set") is True and s.get("key_from") == "saved", "restart: %s/%s key %s" % (s.get("provider"), s.get("model"), s.get("key_from"))) check(not leaked(raws), "no key in /api/state") stop(p) logs = b"".join(open(lg.name, "rb").read() for _, lg in procs) check(not leaked(logs.replace(b"sk-x", b"")) , "no key in the server logs") finally: for p, log in procs: if p.poll() is None: p.kill() log.close() shutil.rmtree(ws, ignore_errors=True) shutil.rmtree(home, ignore_errors=True) print("FAILED:", fails) sys.exit(1 if fails else 0)