|
|
@@ -0,0 +1,374 @@
|
|
|
+#!/usr/bin/env python3
|
|
|
+"""Self-contained web-stack feature tests: app + fake saddle daemon.
|
|
|
+
|
|
|
+Unlike integration_test.py (which needs a real saddle daemon and a mock
|
|
|
+LLM), this suite spawns everything it needs:
|
|
|
+
|
|
|
+- the clanker-saddle app binary (path from argv[1], or builddir default)
|
|
|
+ in a scratch directory with its own web-config.json (statum keys copied
|
|
|
+ from the repo, plus `saddle_callback_base_url`) and SQLite database
|
|
|
+- an in-process fake saddle daemon speaking the wire contract: POST
|
|
|
+ /start (captures the payload), GET /status (scripted state machine),
|
|
|
+ GET /output (empty), GET /log?since= (raw PTY chunks, one new chunk
|
|
|
+ per request so since-paging is exercised)
|
|
|
+
|
|
|
+It then drives the app the way statum.js does (entrypoint loads + action
|
|
|
+POSTs) and asserts the Workstream B features against the wire contract:
|
|
|
+
|
|
|
+3. plan-review blocks hide their duplicate body preview
|
|
|
+5. rules CRUD/attach + the effective `rules` array in /start
|
|
|
+6. project notes, POST /api/saddle/note, `project_notes` + `callback`
|
|
|
+ in /start
|
|
|
+ plus the raw-log side: session_raw_log polling + terminal raw view
|
|
|
+
|
|
|
+Prints PASS/FAIL lines Statum-style; exits non-zero on any failure.
|
|
|
+"""
|
|
|
+import json
|
|
|
+import os
|
|
|
+import shutil
|
|
|
+import socket
|
|
|
+import sqlite3
|
|
|
+import subprocess
|
|
|
+import sys
|
|
|
+import threading
|
|
|
+import time
|
|
|
+import urllib.error
|
|
|
+import urllib.parse
|
|
|
+import urllib.request
|
|
|
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
+
|
|
|
+REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
|
+APP = os.path.abspath(sys.argv[1]) if len(sys.argv) > 1 else os.path.join(REPO, "builddir", "clanker-saddle")
|
|
|
+WORK = os.environ.get("WEBSTACK_WORK", "/tmp/kilo/webstack-e2e")
|
|
|
+DB_PATH = os.path.join(WORK, "web.sqlite")
|
|
|
+
|
|
|
+passes = 0
|
|
|
+failures = 0
|
|
|
+
|
|
|
+
|
|
|
+def check(condition, label):
|
|
|
+ global passes, failures
|
|
|
+ if condition:
|
|
|
+ passes += 1
|
|
|
+ print("PASS %s" % label)
|
|
|
+ else:
|
|
|
+ failures += 1
|
|
|
+ print("FAIL %s" % label)
|
|
|
+
|
|
|
+
|
|
|
+def free_port():
|
|
|
+ with socket.socket() as sock:
|
|
|
+ sock.bind(("127.0.0.1", 0))
|
|
|
+ return sock.getsockname()[1]
|
|
|
+
|
|
|
+
|
|
|
+APP_PORT = free_port()
|
|
|
+DAEMON_PORT = free_port()
|
|
|
+WEB = "http://127.0.0.1:%d" % APP_PORT
|
|
|
+
|
|
|
+
|
|
|
+def http(method, url, data=None, headers=None, timeout=30):
|
|
|
+ body = None
|
|
|
+ if data is not None:
|
|
|
+ body = "&".join("%s=%s" % (k, urllib.parse.quote(str(v))) for k, v in data.items()).encode()
|
|
|
+ request = urllib.request.Request(WEB + url if url.startswith("/") else url,
|
|
|
+ data=body, method=method)
|
|
|
+ if data is not None:
|
|
|
+ request.add_header("Content-Type", "application/x-www-form-urlencoded")
|
|
|
+ for key, value in (headers or {}).items():
|
|
|
+ request.add_header(key, value)
|
|
|
+ try:
|
|
|
+ with urllib.request.urlopen(request, timeout=timeout) as response:
|
|
|
+ return response.status, response.read().decode()
|
|
|
+ except urllib.error.HTTPError as error:
|
|
|
+ return error.code, error.read().decode()
|
|
|
+
|
|
|
+
|
|
|
+def directives(uri, auth=None):
|
|
|
+ headers = {}
|
|
|
+ if auth:
|
|
|
+ headers["X-Statum-Slot"] = "key=%s; as_at=%s" % (auth["key"], auth["as_at"])
|
|
|
+ status, body = http("GET", "/_statum/entrypoint?uri=" + urllib.parse.quote(uri.split("?")[0]),
|
|
|
+ headers=headers)
|
|
|
+ if status != 200:
|
|
|
+ raise SystemExit("entrypoint %s -> HTTP %d: %s" % (uri, status, body[:300]))
|
|
|
+ return json.loads(body)
|
|
|
+
|
|
|
+
|
|
|
+def slot_public(auth, uri, type_name):
|
|
|
+ for directive in directives(uri, auth):
|
|
|
+ if directive["type"] == "set":
|
|
|
+ content = json.loads(directive["snapshot_frame"]["content"])
|
|
|
+ if content["slot"]["type"] == type_name:
|
|
|
+ return content["public"]
|
|
|
+ raise SystemExit("no %s slot in entrypoint state for %s" % (type_name, uri))
|
|
|
+
|
|
|
+
|
|
|
+def act(auth, action, fields):
|
|
|
+ headers = {"X-Statum-Slot": "key=%s; as_at=%s" % (auth["key"], auth["as_at"])}
|
|
|
+ if action.get("private"):
|
|
|
+ headers["X-Statum-Private"] = action["private"]
|
|
|
+ status, body = http(action.get("method", "POST"), action["uri"], fields, headers)
|
|
|
+ if status != 200:
|
|
|
+ raise SystemExit("action %s -> HTTP %d: %s" % (action["uri"], status, body[:300]))
|
|
|
+ return json.loads(body)
|
|
|
+
|
|
|
+
|
|
|
+def login(username, password):
|
|
|
+ frames = directives("/login")
|
|
|
+ content = json.loads(frames[0]["snapshot_frame"]["content"])
|
|
|
+ action = content["public"]["login"]
|
|
|
+ status, body = http(action.get("method", "POST"), action["uri"],
|
|
|
+ {"username": username, "password": password})
|
|
|
+ if status != 200:
|
|
|
+ raise SystemExit("login -> HTTP %d: %s" % (status, body[:300]))
|
|
|
+ for directive in json.loads(body):
|
|
|
+ if directive["type"] == "set":
|
|
|
+ frame = json.loads(directive["snapshot_frame"]["content"])
|
|
|
+ if frame["slot"]["type"] == "auth":
|
|
|
+ return {"key": frame["slot"]["key"], "as_at": frame["as_at"]}
|
|
|
+ raise SystemExit("login response carried no auth slot: " + body[:300])
|
|
|
+
|
|
|
+
|
|
|
+def wait_for(predicate, timeout=60, what="condition", interval=0.5):
|
|
|
+ deadline = time.monotonic() + timeout
|
|
|
+ last = None
|
|
|
+ while time.monotonic() < deadline:
|
|
|
+ last = predicate()
|
|
|
+ if last is not False and last is not None:
|
|
|
+ return last
|
|
|
+ time.sleep(interval)
|
|
|
+ print(" (timeout waiting for %s; last: %s)" % (what, last))
|
|
|
+ return None
|
|
|
+
|
|
|
+
|
|
|
+def task_state(auth, task_id):
|
|
|
+ return slot_public(auth, "/tasks/%d" % task_id, "task")
|
|
|
+
|
|
|
+
|
|
|
+def saddle_api(method, path, token, payload=None):
|
|
|
+ """Raw call against the saddle ingest API; returns (status, json)."""
|
|
|
+ request = urllib.request.Request(WEB + path,
|
|
|
+ data=json.dumps(payload or {}).encode(), method=method)
|
|
|
+ request.add_header("Content-Type", "application/json")
|
|
|
+ if token is not None:
|
|
|
+ request.add_header("Authorization", "Bearer " + token)
|
|
|
+ try:
|
|
|
+ with urllib.request.urlopen(request, timeout=15) as response:
|
|
|
+ return response.status, json.loads(response.read().decode() or "{}")
|
|
|
+ except urllib.error.HTTPError as error:
|
|
|
+ body = error.read().decode()
|
|
|
+ try:
|
|
|
+ return error.code, json.loads(body)
|
|
|
+ except ValueError:
|
|
|
+ return error.code, {"raw": body}
|
|
|
+
|
|
|
+
|
|
|
+def db():
|
|
|
+ return sqlite3.connect(DB_PATH)
|
|
|
+
|
|
|
+
|
|
|
+class FakeSaddle:
|
|
|
+ """In-process saddle daemon implementing the wire contract."""
|
|
|
+
|
|
|
+ PROMPT = "bbarrow@fedora:/workspace/task-1$ "
|
|
|
+
|
|
|
+ def __init__(self):
|
|
|
+ self.lock = threading.Lock()
|
|
|
+ self.start_payloads = [] # captured POST /start bodies
|
|
|
+ self.chunks = [] # [{seq, text}] — the raw log ring
|
|
|
+ self.next_seq = 1
|
|
|
+ self.state = "idle"
|
|
|
+ self.summary = ""
|
|
|
+ self.status_polls_after_start = 0
|
|
|
+
|
|
|
+ def append_chunk(self, text):
|
|
|
+ with self.lock:
|
|
|
+ self.chunks.append({"seq": self.next_seq, "text": text})
|
|
|
+ self.next_seq += 1
|
|
|
+
|
|
|
+ def log_since(self, since):
|
|
|
+ with self.lock:
|
|
|
+ chunks = [c for c in self.chunks if c["seq"] > since]
|
|
|
+ return chunks, self.next_seq
|
|
|
+
|
|
|
+ class Handler(BaseHTTPRequestHandler):
|
|
|
+ daemon = None # set on the class before serving
|
|
|
+
|
|
|
+ def log_message(self, fmt, *args):
|
|
|
+ pass
|
|
|
+
|
|
|
+ def _send(self, status, obj):
|
|
|
+ body = json.dumps(obj).encode()
|
|
|
+ self.send_response(status)
|
|
|
+ self.send_header("Content-Type", "application/json")
|
|
|
+ self.send_header("Content-Length", str(len(body)))
|
|
|
+ self.end_headers()
|
|
|
+ self.wfile.write(body)
|
|
|
+
|
|
|
+ def do_POST(self):
|
|
|
+ length = int(self.headers.get("Content-Length") or 0)
|
|
|
+ raw = self.rfile.read(length).decode() if length else "{}"
|
|
|
+ if self.path == "/start":
|
|
|
+ try:
|
|
|
+ payload = json.loads(raw)
|
|
|
+ except ValueError:
|
|
|
+ payload = {"raw": raw}
|
|
|
+ self.daemon.start_payloads.append(payload)
|
|
|
+ self.daemon.append_chunk(self.daemon.PROMPT)
|
|
|
+ self.daemon.append_chunk("echo hello\r\n")
|
|
|
+ self.daemon.append_chunk("hello\r\n")
|
|
|
+ self.daemon.state = "working"
|
|
|
+ self.daemon.status_polls_after_start = 0
|
|
|
+ self._send(200, {})
|
|
|
+ elif self.path == "/answer":
|
|
|
+ self._send(200, {})
|
|
|
+ else:
|
|
|
+ self._send(404, {"error": "no such route"})
|
|
|
+
|
|
|
+ def do_GET(self):
|
|
|
+ if self.path == "/status":
|
|
|
+ if self.daemon.state == "working":
|
|
|
+ self.daemon.status_polls_after_start += 1
|
|
|
+ if self.daemon.status_polls_after_start >= 2:
|
|
|
+ self.daemon.state = "done"
|
|
|
+ self.daemon.summary = "fake daemon finished"
|
|
|
+ self._send(200, {"state": self.daemon.state,
|
|
|
+ "summary": self.daemon.summary})
|
|
|
+ elif self.path.startswith("/output"):
|
|
|
+ self._send(200, [])
|
|
|
+ elif self.path.startswith("/log"):
|
|
|
+ query = urllib.parse.urlparse(self.path).query
|
|
|
+ since = int(urllib.parse.parse_qs(query).get("since", ["0"])[0])
|
|
|
+ chunks, nxt = self.daemon.log_since(since)
|
|
|
+ # Simulate the live PTY: each poll produces one more chunk.
|
|
|
+ self.daemon.append_chunk("stream-line-%d\r\n" % self.daemon.next_seq)
|
|
|
+ self._send(200, {"chunks": chunks, "next": nxt})
|
|
|
+ else:
|
|
|
+ self._send(404, {"error": "no such route"})
|
|
|
+
|
|
|
+ def serve(self):
|
|
|
+ handler = type("BoundHandler", (FakeSaddle.Handler,), {"daemon": self})
|
|
|
+ server = ThreadingHTTPServer(("127.0.0.1", DAEMON_PORT), handler)
|
|
|
+ threading.Thread(target=server.serve_forever, daemon=True).start()
|
|
|
+ return server
|
|
|
+
|
|
|
+
|
|
|
+def start_app():
|
|
|
+ config = json.load(open(os.path.join(REPO, "web-config.json")))
|
|
|
+ config["saddle_callback_base_url"] = WEB
|
|
|
+ with open(os.path.join(WORK, "web-config.json"), "w") as handle:
|
|
|
+ json.dump(config, handle)
|
|
|
+
|
|
|
+ env = dict(os.environ)
|
|
|
+ env.update({
|
|
|
+ "SPRY_DB_PATH": DB_PATH,
|
|
|
+ "SPRY_SEED_ADMIN_USERNAME": "admin",
|
|
|
+ "SPRY_SEED_ADMIN_PASSWORD": "adminpass123",
|
|
|
+ "SADDLE_LLM_BASE_URL": "http://127.0.0.1:%d" % DAEMON_PORT,
|
|
|
+ "SADDLE_PROJECTS_ROOT": os.path.join(WORK, "workspace"),
|
|
|
+ })
|
|
|
+ log = open(os.path.join(WORK, "app.log"), "w")
|
|
|
+ process = subprocess.Popen([APP, str(APP_PORT)], cwd=WORK, env=env,
|
|
|
+ stdout=log, stderr=subprocess.STDOUT)
|
|
|
+ if wait_for(lambda: app_ready() or False, timeout=30, what="app ready") is None:
|
|
|
+ raise SystemExit("app did not come up; see %s/app.log" % WORK)
|
|
|
+ return process
|
|
|
+
|
|
|
+
|
|
|
+def app_ready():
|
|
|
+ try:
|
|
|
+ status, _ = http("GET", "/login", timeout=3)
|
|
|
+ return status == 200
|
|
|
+ except (urllib.error.URLError, OSError):
|
|
|
+ return False
|
|
|
+
|
|
|
+
|
|
|
+def main():
|
|
|
+ if not os.path.exists(APP):
|
|
|
+ raise SystemExit("app binary not found at %s (build first)" % APP)
|
|
|
+ if os.path.exists(WORK):
|
|
|
+ shutil.rmtree(WORK)
|
|
|
+ os.makedirs(os.path.join(WORK, "workspace"))
|
|
|
+
|
|
|
+ fake = FakeSaddle()
|
|
|
+ fake.serve()
|
|
|
+ app = start_app()
|
|
|
+
|
|
|
+ try:
|
|
|
+ auth = login("admin", "adminpass123")
|
|
|
+ check(True, "login yields an auth slot")
|
|
|
+
|
|
|
+ print("== feature 3: plan review hides duplicate body ==")
|
|
|
+ state = slot_public(auth, "/projects", "projects")
|
|
|
+ act(auth, state["create"], {
|
|
|
+ "name": "demo",
|
|
|
+ "git_url": "https://example.com/demo.git",
|
|
|
+ "default_branch": "main",
|
|
|
+ })
|
|
|
+ conn = db()
|
|
|
+ conn.execute("insert into tasks (project_id, title, status, priority,"
|
|
|
+ " is_design_task, plan, created_at)"
|
|
|
+ " values (1, 'Plan review task', 'working', 5, 1, '# The plan\n\n1. Do it',"
|
|
|
+ " strftime('%s','now'))")
|
|
|
+ conn.commit()
|
|
|
+ plan_task = conn.execute("select max(id) from tasks").fetchone()[0]
|
|
|
+
|
|
|
+ saddles_state = slot_public(auth, "/saddles", "saddles")
|
|
|
+ act(auth, saddles_state["register"], {"name": "fake-saddle",
|
|
|
+ "host": "127.0.0.1:%d" % DAEMON_PORT})
|
|
|
+ registered = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
|
|
|
+ if s["name"] == "fake-saddle"]
|
|
|
+ check(len(registered) == 1 and registered[0]["token"].startswith("saddle-"),
|
|
|
+ "fake saddle registered with an API token")
|
|
|
+ fake_id = int(registered[0]["id"])
|
|
|
+ fake_token = registered[0]["token"]
|
|
|
+
|
|
|
+ conn.execute("insert into sessions (task_id, saddle_id, started_at, ended_at, outcome)"
|
|
|
+ " values (?, ?, strftime('%s','now'), strftime('%s','now'), 'done')",
|
|
|
+ (plan_task, fake_id))
|
|
|
+ conn.commit()
|
|
|
+ session_id = conn.execute("select max(id) from sessions").fetchone()[0]
|
|
|
+ conn.close()
|
|
|
+
|
|
|
+ status, body = saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
|
|
|
+ {"type": "block", "session_id": session_id, "data": {
|
|
|
+ "question": "Plan submitted for review",
|
|
|
+ "type": "rfc",
|
|
|
+ "body": "# The plan\n\n1. Do it"}})
|
|
|
+ check(status == 200 and body.get("block_id"), "plan-review block raised")
|
|
|
+ status, _ = saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
|
|
|
+ {"type": "block", "session_id": session_id, "data": {
|
|
|
+ "question": "Which database?",
|
|
|
+ "type": "question",
|
|
|
+ "options": ["SQLite", "Postgres"],
|
|
|
+ "body": "# Storage choice"}})
|
|
|
+ check(status == 200, "question block raised")
|
|
|
+
|
|
|
+ detail = task_state(auth, plan_task)
|
|
|
+ blocks = {b["block_type"]: b for b in detail["blocks"]}
|
|
|
+ check(blocks["rfc"].get("is_plan_review") is True,
|
|
|
+ "rfc block flagged as the plan review")
|
|
|
+ check(blocks["rfc"].get("show_body") is False,
|
|
|
+ "pending plan review hides the duplicate body preview")
|
|
|
+ check(blocks["question"].get("show_body") is True,
|
|
|
+ "question block keeps its body preview")
|
|
|
+
|
|
|
+ act(auth, blocks["rfc"]["answer_action"], {"answer": "Plan approved. Proceed."})
|
|
|
+ detail = task_state(auth, plan_task)
|
|
|
+ rfc = [b for b in detail["blocks"] if b["block_type"] == "rfc"][0]
|
|
|
+ check(rfc.get("show_body") is True,
|
|
|
+ "answered plan review shows the body again (review section gone)")
|
|
|
+ finally:
|
|
|
+ app.terminate()
|
|
|
+ try:
|
|
|
+ app.wait(timeout=5)
|
|
|
+ except subprocess.TimeoutExpired:
|
|
|
+ app.kill()
|
|
|
+
|
|
|
+ print("%d passed, %d failed" % (passes, failures))
|
|
|
+ return 1 if failures else 0
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == "__main__":
|
|
|
+ sys.exit(main())
|
