CST-agent/scripts/e2e_webui_settings.py

"""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/hermit"
os.makedirs(cfgdir, exist_ok=True)
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)