From dd3462c3cd0a7ff8dbb31c5b4e6d133a04506419 Mon Sep 17 00:00:00 2001 From: Igor Soarez Date: Sat, 24 Oct 2026 14:41:49 +0300 Subject: tests/pi-live.sh: check a real pi release against dingger runs a real pi binary on a tmux server and dingger daemon of its own, with a fake OpenAI-compatible model (tests/fake-model.py, reply delay set by the prompt) and its own pi agent directory: no network, no credentials, the user's pi untouched. plain shells, and a stop if the panes would not run the pi under test (a login shell's profile puts the user's pi first). it checks identification, the extension's state and session reports, a turn (working, then done), /reload, an extension's blocked event, screen detection for a pi without the extension, restore by session file, and exit. pi 1.4.0 and 1.0.3: 23 of 23 each. a dingger whose extension listens for a renamed event fails it. in both, a pi confirm dialog shows as not blocked: pi tells terminals (OSC 7501, which tmux does not forward) but not extensions. --- tests/fake-model.py | 59 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 59 insertions(+) create mode 100755 tests/fake-model.py (limited to 'tests/fake-model.py') diff --git a/tests/fake-model.py b/tests/fake-model.py new file mode 100755 index 0000000..a02b98c --- /dev/null +++ b/tests/fake-model.py @@ -0,0 +1,59 @@ +#!/usr/bin/env python3 +"""A fake OpenAI-compatible model for tests/pi-live.sh: streams a short reply +to /v1/chat/completions, after a delay taken from the last user message +("slow 3" waits 3 seconds). Prints its port, then serves until killed.""" +import json +import re +import sys +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + + +def last_user_text(body): + for message in reversed(body.get("messages", [])): + if message.get("role") == "user": + content = message.get("content") + if isinstance(content, list): + return " ".join(part.get("text", "") for part in content if isinstance(part, dict)) + return content or "" + return "" + + +class Handler(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def do_GET(self): + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(json.dumps({"object": "list", "data": [{"id": "fake-model", "object": "model"}]}).encode()) + + def do_POST(self): + body = json.loads(self.rfile.read(int(self.headers.get("Content-Length", 0))) or b"{}") + text = last_user_text(body) + match = re.search(r"slow (\d+(?:\.\d+)?)", text) + time.sleep(float(match.group(1)) if match else 0.2) + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Cache-Control", "no-cache") + self.end_headers() + + def chunk(delta, finish=None, usage=None): + data = {"id": "fake", "object": "chat.completion.chunk", "created": int(time.time()), "model": "fake-model", + "choices": [{"index": 0, "delta": delta, "finish_reason": finish}]} + if usage: + data["usage"] = usage + self.wfile.write(f"data: {json.dumps(data)}\n\n".encode()) + self.wfile.flush() + + chunk({"role": "assistant", "content": "Done: "}) + chunk({"content": "fake reply."}) + chunk({}, "stop", {"prompt_tokens": 10, "completion_tokens": 3, "total_tokens": 13}) + self.wfile.write(b"data: [DONE]\n\n") + self.wfile.flush() + + +server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) +print(server.server_address[1], flush=True) +server.serve_forever() -- cgit v1.2.3