stubborn-otter/otter-boards

Otter Boards, a team task board you can run from the web or from Claude and ChatGPT. FastAPI + DataK3 + appid + Models + MCP.

• boards
• tasks
• fastapi
• mcp
• dodil-challenge
• dodil-challenge-boards

A remix template · 1 remixes · Remix stubborn-otter/otter-boards

tests/e2e/fake_models.py

"""A stand-in for Dodil Models for the local end-to-end run: OpenAI-shaped, with the gateway's
{"status": ..., "data": {...}} wrapper as the brief describes it. Nothing here calls Dodil.

Create the file named by MODELS_DOWN_FLAG and every request fails with 503, to prove the fallbacks.
"""
import json
import os
import sys
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

DOWN = Path(os.environ["MODELS_DOWN_FLAG"])
TOPICS = (("invoice", "bill", "payment"), ("launch", "party", "cake"), ("website", "homepage"))


def vector(text):
    return [float(any(word in text.lower() for word in words)) for words in TOPICS]


class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        body = json.loads(self.rfile.read(int(self.headers["content-length"])))
        if DOWN.exists() or not self.headers.get("authorization", "").startswith("Bearer "):
            self._reply(503, {"status": "error"})
        elif self.path.endswith("/embeddings"):
            rows = [{"object": "embedding", "index": i, "embedding": vector(t)}
                    for i, t in enumerate(body["input"])]
            self._reply(200, {"status": "ok", "data": {"object": "list", "data": rows}})
        else:
            prompt = body["messages"][0]["content"]
            text = ("- Draft it\n- Check it\n3. Send it" if "checklist" in prompt
                    else "Stand-in summary of the board.")
            self._reply(200, {"status": "ok", "data": {
                "choices": [{"message": {"role": "assistant", "content": text}}]}})

    def _reply(self, status, body):
        payload = json.dumps(body).encode()
        self.send_response(status)
        self.send_header("content-type", "application/json")
        self.send_header("content-length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def log_message(self, *args):
        pass


if __name__ == "__main__":
    ThreadingHTTPServer(("127.0.0.1", int(sys.argv[1])), Handler).serve_forever()