Hermit/scripts/e2e_webui_workspace.py

"""WebUI workspace views: git diff, run settings, saved permission rules.

  python3 scripts/e2e_webui_workspace.py     (CST_AGENT= picks the binary)

Checks: /api/diff lists git status and the working-tree diff (per file too),
an option-looking path is only ever a path, and a directory without git
says so; /api/settings reports protocol, tools, confirmations and the
sandbox, rejects a bad body without applying any of it, applies a good one,
and is a 409 while a turn runs; /api/permissions shows the saved rules and
DELETE clears them. HOME points at a temp dir.
"""
import json, os, shutil, 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"))


class Api(BaseHTTPRequestHandler):
    """A slow plain answer, so a turn is running for a while."""
    def log_message(self, *a):
        pass

    def do_POST(self):
        self.rfile.read(int(self.headers.get("Content-Length", 0)))
        time.sleep(2)
        self.send_response(200)
        self.send_header("Content-Type", "text/event-stream")
        self.end_headers()
        for o in ({"delta": {"role": "assistant", "content": "好"}}, {"delta": {}, "finish_reason": "stop"}):
            chunk = {"id": "x", "object": "chat.completion.chunk", "model": "m", "choices": [dict(index=0, **o)]}
            self.wfile.write(b"data: " + json.dumps(chunk, ensure_ascii=False).encode() + b"\n\n")
        self.wfile.write(b"data: [DONE]\n\n")


fails = 0


def check(ok, msg):
    global fails
    print(("PASS " if ok else "FAIL ") + msg)
    fails += 0 if ok else 1


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=30) as r:
            return r.status, json.loads(r.read() or b"{}")
    except urllib.error.HTTPError as e:
        raw = e.read()
        try:
            return e.code, json.loads(raw or b"{}")
        except ValueError:
            return e.code, {}


def git(*args):
    return subprocess.run(["git", "-C", ws] + list(args), capture_output=True, text=True)


api = ThreadingHTTPServer(("127.0.0.1", 0), Api)
threading.Thread(target=api.serve_forever, daemon=True).start()
ws = tempfile.mkdtemp(prefix="cst-ws-")
plain = tempfile.mkdtemp(prefix="cst-nogit-")
home = tempfile.mkdtemp(prefix="cst-home-")
env = dict(os.environ, HOME=home, XDG_CONFIG_HOME=home + "/.config",
           XDG_DATA_HOME=home + "/.local/share", XDG_CACHE_HOME=home + "/.cache",
           GIT_CONFIG_NOSYSTEM="1")
for k in ("DEEPSEEK_API_KEY", "OPENAI_API_KEY", "CST_API_KEY"):
    env.pop(k, None)
have_git = shutil.which("git") is not None
if have_git:
    git("init", "-q")
    git("config", "user.email", "e2e@example.com")
    git("config", "user.name", "e2e")
    open(ws + "/a.txt", "w").write("one\n")
    git("add", "a.txt")
    git("commit", "-q", "-m", "first")
    open(ws + "/a.txt", "w").write("one\n第二行\n")
    open(ws + "/new.txt", "w").write("new\n")
os.makedirs(ws + "/.agent", exist_ok=True)
json.dump({"tools": ["write_file"], "shell_prefixes": ["make"]}, open(ws + "/.agent/permissions.json", "w"))

log = open(home + "/server.log", "w+")
proc = subprocess.Popen([BIN, "--web", "--web-port", "0", "--workspace", ws, "--new", "--api-key", "sk-x",
                         "--provider", "deepseek", "--model", "deepseek-flash",
                         "--api-base", "http://127.0.0.1:%d/v1" % api.server_address[1],
                         "--protocol", "native-eos", "--approval", "auto"],
                        stdout=log, stderr=subprocess.STDOUT, env=env)


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]


try:
    base = "http://127.0.0.1:%s" % wait(port)

    # 1. diff
    if have_git:
        code, d = call(base, "/api/diff")
        files = {f["path"]: f["status"] for f in d.get("files", [])}
        check(code == 200 and d.get("git") is True, "diff: git repo seen")
        check(files.get("a.txt") == " M" and files.get("new.txt") == "??", "status lists both files: %s" % files)
        check("+第二行" in d.get("diff", "") and "a.txt" in d.get("diff", ""), "the diff has the change")
        code, d = call(base, "/api/diff?path=a.txt")
        check(code == 200 and "+第二行" in d.get("diff", "") and d.get("path") == "a.txt", "diff of one file")
        code, d = call(base, "/api/diff?path=new.txt")
        check(code == 200 and d.get("diff") == "", "an untracked file has no diff")
        evil = ws + "/pwned"
        code, d = call(base, "/api/diff?path=" + urllib.request.quote("--output=" + evil))
        check(code == 200 and not os.path.exists(evil), "an option-looking path stays a path")
    else:
        print("SKIP diff checks: no git")

    # 2. settings: what is there
    code, s = call(base, "/api/settings")
    names = [t["name"] for t in s.get("tools", [])]
    check(code == 200 and s.get("protocol") == "native-eos" and len(s.get("protocols", [])) == 5,
          "settings: protocol %s of %s" % (s.get("protocol"), s.get("protocols")))
    check("read_file" in names and all("perm" in t and "on" in t for t in s["tools"]), "tool list: %d tools" % len(names))
    check(set(s.get("confirm", {})) == {"shell", "write", "net"}, "confirm flags present")
    check(isinstance(s.get("sandbox", {}).get("kind"), str), "sandbox kind: %s" % s.get("sandbox"))
    victim = next((t["name"] for t in s["tools"] if t["perm"] == "net"), names[-1])

    # 3. a bad body changes nothing
    for body, why in (({"protocol": "nope"}, "unknown protocol"), ({"max_steps": 0}, "max_steps 0"),
                      ({"max_steps": "5"}, "max_steps as text"), ({"tool_profile": "huge"}, "unknown profile"),
                      ({"tools": {"nope": False}}, "unknown tool"), ({"tools": {victim: "off"}}, "tool value not bool"),
                      ({"confirm": {"disk": True}}, "unknown confirm key"),
                      ({"max_steps": 9, "protocol": "nope"}, "mixed good and bad")):
        check(call(base, "/api/settings", body)[0] == 400, "400 for %s" % why)
    check(call(base, "/api/settings")[1].get("max_steps") == s.get("max_steps"), "nothing applied from a bad body")

    # 4. a good body applies
    code, d = call(base, "/api/settings", {"protocol": "ctl", "max_steps": 7, "tool_profile": "minimal",
                                            "tools": {victim: False}, "confirm": {"shell": True, "net": False}})
    check(code == 200 and d.get("ok"), "settings accepted (%s %s)" % (code, d))
    code, s2 = call(base, "/api/settings")
    t2 = {t["name"]: t for t in s2.get("tools", [])}
    check(s2.get("protocol") == "ctl" and s2.get("max_steps") == 7 and s2.get("tool_profile") == "minimal",
          "protocol/steps/profile applied")
    check(t2.get(victim, {}).get("off") is True and t2[victim]["on"] is False, "%s switched off" % victim)
    check(s2["confirm"]["shell"] is True and s2["confirm"]["net"] is False, "confirm flags applied")
    st = call(base, "/api/state")[1]
    check(st.get("protocol_tools") == "ctl" and st.get("max_steps") == 7 and st.get("tool_profile") == "minimal",
          "the state shows them")
    call(base, "/api/settings", {"protocol": "native-eos", "tool_profile": "default", "tools": {victim: True}})
    t3 = {t["name"]: t for t in call(base, "/api/settings")[1]["tools"]}
    check(t3[victim]["off"] is False and t3[victim]["on"] is True, "%s back on" % victim)

    # 5. permissions
    code, p = call(base, "/api/permissions")
    check(code == 200 and p.get("tools") == ["write_file"] and p.get("shell_prefixes") == ["make"]
          and "write_file" in p.get("rules", ""), "saved rules: %s %s" % (p.get("tools"), p.get("shell_prefixes")))
    code, _ = call(base, "/api/permissions", method="DELETE")
    check(code == 200, "DELETE /api/permissions")
    p = call(base, "/api/permissions")[1]
    check(p.get("tools") == [] and p.get("shell_prefixes") == [], "the rules are gone")
    saved = json.load(open(ws + "/.agent/permissions.json"))
    check(saved.get("tools") == [] and saved.get("shell_prefixes") == [], "and gone from the file")

    # 6. while a turn runs: settings and the rule reset are 409, reads still work
    code, _ = call(base, "/api/turn", {"text": "慢慢来"})
    check(code == 202, "a slow turn started")
    check(call(base, "/api/settings", {"max_steps": 3})[0] == 409, "settings during a turn: 409")
    check(call(base, "/api/permissions", method="DELETE")[0] == 409, "rule reset during a turn: 409")
    check(call(base, "/api/settings")[0] == 200 and call(base, "/api/permissions")[0] == 200,
          "reads during a turn: 200")
    if have_git:
        check(call(base, "/api/diff")[0] == 200, "diff during a turn: 200")
    wait(lambda: not call(base, "/api/state")[1].get("status_busy"), 20)

    # 7. a directory without git
    if have_git:
        code, _ = call(base, "/api/workspace", {"path": plain})
        code, d = call(base, "/api/diff")
        check(code == 200 and d.get("git") is False and "diff" not in d, "no git repo: git=false")
finally:
    proc.terminate()
    try:
        proc.wait(20)
    except subprocess.TimeoutExpired:
        proc.kill()
        check(False, "server hung on SIGTERM")
    log.close()
    for d in (ws, plain, home):
        shutil.rmtree(d, ignore_errors=True)
    api.shutdown()
print("FAILED:", fails)
sys.exit(1 if fails else 0)