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.
A remix template · 1 remixes · Remix stubborn-otter/otter-boards
tests/e2e/check.py
"""Exercises the running server from outside, the way people and Claude would: plain HTTP for the
web door, the MCP SDK's real streamable-HTTP client for /mcp. Prints every check; any failure
stops the run with a non-zero exit.
"""
import asyncio
import json
import os
import sys
from concurrent.futures import ThreadPoolExecutor
from datetime import date, timedelta
from pathlib import Path
import httpx
import httpx2
import psycopg
from mcp.client import Client
from mcp.client.streamable_http import streamable_http_client
BASE = os.environ["APP_URL"]
DEV = os.environ["DEV_USER"]
OWNER, MEMBER, VIEWER, STRANGER = DEV, "bo@otter.test", "cy@otter.test", "dee@otter.test"
TODAY = date.today()
checks = 0
def ok(what, passed=True, detail=""):
global checks
if not passed:
print(f"FAILED {what} {detail}")
sys.exit(1)
checks += 1
print(f" ok {what}")
def who(email):
return {"X-Dodil-User": json.dumps({"sub": email, "email": email})}
def api(method, path, as_=None, want=None, **kw):
headers = who(as_) if as_ else {}
r = httpx.request(method, BASE + path, headers=headers, timeout=30, **kw)
if want is not None and r.status_code != want:
ok(f"{method} {path} as {as_}", False, f"got {r.status_code}, wanted {want}: {r.text[:300]}")
return r
class Mcp:
"""A signed-in MCP client over real HTTP."""
def __init__(self, email=None):
self.headers = who(email) if email else {}
def __enter__(self):
return self
def __exit__(self, *exc):
return False
async def _run(self, steps):
http = httpx2.AsyncClient(headers=self.headers, timeout=30)
async with http, Client(streamable_http_client(f"{BASE}/mcp", http_client=http)) as client:
return await steps(client)
def run(self, steps):
return asyncio.run(self._run(steps))
def call(self, tool, **arguments):
async def steps(client):
return await client.call_tool(tool, arguments)
result = self.run(steps)
if result.is_error:
ok(f"mcp {tool}", False, str(result.content))
data = result.structured_content
return data["result"] if isinstance(data, dict) and set(data) == {"result"} else data
def refused(self, tool, **arguments):
async def steps(client):
return await client.call_tool(tool, arguments)
result = self.run(steps)
if not result.is_error:
ok(f"mcp {tool} should be refused", False, str(result.content))
return result.content[0].text
def main():
print("-- identity and the gate")
ok("healthz is public", api("GET", "/healthz").json() == {"ok": True})
ok("metadata is public", api("GET", "/.well-known/oauth-protected-resource/mcp",
want=200).json()["resource"].endswith("/mcp"))
ok("DEV_USER is who you are with no header", api("GET", "/api/me").json()["email"] == DEV)
ok("the page is served", "Otter Boards" in api("GET", "/", want=200).text)
ok("app.js is served", api("GET", "/static/app.js", want=200).text != "")
ok("api docs are not public", api("GET", "/openapi.json").status_code == 404)
print("-- the core over HTTP")
board = api("POST", "/api/boards", OWNER, 201, json={"title": "Ops"}).json()
bid = board["id"]
for email, role in ((MEMBER, "member"), (VIEWER, "viewer")):
api("POST", f"/api/boards/{bid}/members", OWNER, 201, json={"email": email, "role": role})
shown = api("GET", f"/api/boards/{bid}", OWNER, 200).json()
names = [x["name"] for x in shown["lists"]]
ok("a new board has its three lists", names == ["To do", "Doing", "Done"], names)
ok("three people on the board", len(shown["members"]) == 3)
review = api("POST", f"/api/boards/{bid}/lists", MEMBER, 201, json={"name": "Review"}).json()
api("PATCH", f"/api/boards/{bid}/lists/{review['id']}", MEMBER, 200, json={"name": "QA"})
api("POST", f"/api/boards/{bid}/lists/{review['id']}/move", MEMBER, 200,
json={"before": shown["lists"][2]["id"]})
names = [x["name"] for x in api("GET", f"/api/boards/{bid}", OWNER, 200).json()["lists"]]
ok("a list was added, renamed and moved", names == ["To do", "Doing", "QA", "Done"], names)
due = (TODAY + timedelta(days=3)).isoformat()
late = (TODAY - timedelta(days=2)).isoformat()
made = api("POST", f"/api/boards/{bid}/cards", MEMBER, 201, json={
"title": "Quarterly report", "description": "Numbers & <b>words</b>", "assignee": MEMBER,
"due_date": due, "labels": ["finance", "q3"]}).json()
ok("a card keeps every field", (made["description"], made["assignee"], made["due_date"],
made["labels"]) == ("Numbers & <b>words</b>", MEMBER, due,
["finance", "q3"]))
second = api("POST", f"/api/boards/{bid}/cards", OWNER, 201, json={
"title": "Late thing", "due_date": late, "assignee": OWNER}).json()
ok("card numbers count up", (made["number"], second["number"]) == (1, 2))
ok("a late card is flagged overdue", second["overdue"] is True)
moved = api("POST", f"/api/cards/{made['id']}/move", MEMBER, 200,
json={"list_name": "Doing"}).json()
ok("drag-and-drop route moves a card", moved["list"] == "Doing")
third = api("POST", f"/api/boards/{bid}/cards", OWNER, 201,
json={"title": "Above the report", "list_name": "Doing"}).json()
api("POST", f"/api/cards/{third['id']}/move", OWNER, 200,
json={"list_name": "Doing", "before": made["id"]})
doing = api("GET", f"/api/boards/{bid}", OWNER, 200).json()["lists"][1]["cards"]
ok("a drop lands at the chosen spot",
[c["number"] for c in doing] == [3, 1], [c["number"] for c in doing])
api("PATCH", f"/api/cards/{made['id']}", MEMBER, 200,
json={"title": "Quarterly report v2", "due_date": None})
history = api("GET", f"/api/cards/{made['id']}/history", VIEWER, 200).json()
actions = [e["action"] for e in history]
ok("history has who, when and what", actions == [
"added the card to To do", "moved from To do to Doing",
"changed title from Quarterly report to Quarterly report v2",
f"cleared due date (was {due})"] and history[1]["who"] == MEMBER
and history[1]["at"].endswith("Z"), actions)
print("-- roles over HTTP")
api("POST", f"/api/boards/{bid}/cards", VIEWER, 403, json={"title": "nope"})
api("POST", f"/api/cards/{made['id']}/move", VIEWER, 403, json={"list_name": "Done"})
api("PATCH", f"/api/cards/{made['id']}", VIEWER, 403, json={"title": "nope"})
api("POST", f"/api/boards/{bid}/members", VIEWER, 403, json={"email": "x@y.io"})
api("POST", f"/api/boards/{bid}/members", MEMBER, 403, json={"email": "x@y.io"})
api("DELETE", f"/api/boards/{bid}/members/{VIEWER}", MEMBER, 403)
ok("viewer can't write and member can't manage people")
for path in (f"/api/boards/{bid}", f"/api/boards/{bid}/members", f"/api/cards/{made['id']}",
f"/api/cards/{made['id']}/history", f"/api/boards/{bid}/summary"):
api("GET", path, STRANGER, 404)
api("POST", f"/api/cards/{made['id']}/move", STRANGER, 404, json={"list_name": "Done"})
ok("a stranger's board list is empty", api("GET", "/api/boards", STRANGER, 200).json() == [])
ok("a stranger gets 404 everywhere")
api("GET", "/api/boards", None).status_code == 200 # DEV_USER: the owner
r = httpx.get(f"{BASE}/api/boards", headers={"X-Dodil-User": "garbage"})
ok("a bad identity header is refused, not defaulted", r.status_code == 401, r.status_code)
print("-- card numbers under load")
with ThreadPoolExecutor(max_workers=12) as pool:
results = list(pool.map(lambda i: api(
"POST", f"/api/boards/{bid}/cards", OWNER, 201,
json={"title": f"Race {i}", "list_name": "QA"}).json()["number"], range(24)))
ok("24 parallel adds got 24 different numbers", len(set(results)) == 24, sorted(results))
board_now = api("GET", f"/api/boards/{bid}", OWNER, 200).json()
qa = next(x for x in board_now["lists"] if x["name"] == "QA")
ok("and all 24 are on the board", len(qa["cards"]) == 24, len(qa["cards"]))
for c in qa["cards"]:
api("POST", f"/api/cards/{c['id']}/archive", OWNER, 200)
print("-- a real MCP client")
owner, member, viewer, stranger = Mcp(), Mcp(MEMBER), Mcp(VIEWER), Mcp(STRANGER)
async def list_tools(client):
return (await client.list_tools()).tools
tools = owner.run(list_tools)
ok(f"the client lists {len(tools)} tools", len(tools) >= 25)
ok("every tool describes itself", all(t.description for t in tools))
ok("no tool exposes raw tables", not any(
word in t.name for t in tools for word in ("sql", "query", "table", "select")))
async def unsigned():
async with httpx.AsyncClient() as http:
return await http.post(f"{BASE}/mcp", headers={"X-Dodil-User": "garbage"}, json={
"jsonrpc": "2.0", "id": 1, "method": "tools/list"})
r = asyncio.run(unsigned())
ok("no person: 401 pointing at sign-in metadata",
r.status_code == 401 and "resource_metadata" in r.headers["www-authenticate"])
ok("whoami over the DEV_USER path", owner.call("whoami")["email"] == DEV)
card = member.call("add_card", board="Ops", title="Send invoice", list_name="Doing",
assignee="me", due_date=due)
ok("add a card 'Send invoice' to Doing on the Ops board",
(card["list"], card["board"]) == ("Doing", "Ops"), card)
number = card["number"]
plate = member.call("my_cards")
ok("what's on my plate this week?",
"Send invoice" in [c["title"] for c in plate["due_soon"]], plate)
overdue = owner.call("overdue_cards")
ok("what's overdue?", [c["title"] for c in overdue["cards"]] == ["Late thing"], overdue)
moved = member.call("move_card", card=number, list_name="Done")
ok(f"move card {number} to Done", moved["list"] == "Done", moved)
history = viewer.call("card_history", card=number, board="ops")
ok("the move is in the history, under the right person",
history[-1]["who"] == MEMBER and "Doing to Done" in history[-1]["action"], history)
print("-- the same rules over MCP")
ok("viewer can't add", "viewers" in viewer.refused("add_card", board="Ops", title="no"))
ok("viewer can't move", "viewers" in viewer.refused("move_card", card=number, list_name="To do"))
ok("member can't invite", "owner" in member.refused("add_member", board="Ops", email="x@y.io"))
ok("stranger can't see by name", "no such" in stranger.refused("show_board", board="Ops"))
ok("stranger can't reach a card", "no such" in stranger.refused(
"move_card", card=number, list_name="To do"))
ok("stranger's plate is empty", stranger.call("my_boards") == []
and stranger.call("overdue_cards")["count"] == 0)
print("-- extras, through the real client and a stand-in for Dodil Models")
a = owner.call("add_card", board="Ops", title="Launch party")
b = owner.call("add_card", board="Ops", title="Bake the cake")
c = owner.call("add_card", board="Ops", title="Pay the invoice")
owner.call("add_blocker", card="Launch party", blocked_by=b["number"], board="Ops")
report = viewer.call("what_is_blocking", card="launch party")
ok("what is blocking the launch?",
[x["title"] for x in report["open_blockers"]] == ["Bake the cake"], report)
ok("a loop is refused", "itself" in owner.refused(
"add_blocker", card=b["number"], blocked_by=a["number"], board="Ops"))
steps = member.call("suggest_checklist", card=a["number"], board="Ops")["suggested_checklist"]
ok("AI checklist through the envelope", steps == ["Draft it", "Check it", "Send it"], steps)
ok("suggesting saved nothing",
owner.call("show_card", card=a["number"], board="Ops")["checklist"] == [])
saved = member.call("save_checklist", card=a["number"], items=steps[:2], board="Ops")
ok("saving keeps it on the card", len(saved["checklist"]) == 2)
ok("viewer can't ask for a checklist", "viewers" in viewer.refused(
"suggest_checklist", card=a["number"], board="Ops"))
found = viewer.call("find_similar_cards", card=a["number"], board="Ops")
ok("similar cards by meaning", found["method"] == "meaning"
and found["matches"][0]["title"] == "Bake the cake", found)
summary = owner.call("board_summary", board="Ops")
ok("daily summary worded by AI", summary["source"] == "ai", summary)
ok("overdue is in the summary's facts", any("Late thing" in x for x in
summary["facts"]["overdue"]), summary)
Path(os.environ["MODELS_DOWN_FLAG"]).touch()
found = viewer.call("find_similar_cards", card=c["number"], board="Ops")
ok("Models down: vectors already stored still answer by meaning",
found["method"] == "meaning", found)
fresh = owner.call("add_card", board="Ops", title="Chase the invoice")
found = viewer.call("find_similar_cards", card=fresh["number"], board="Ops")
ok("Models down: a new card falls back to shared words", found["method"] == "keywords" and
{m["title"] for m in found["matches"]} == {"Send invoice", "Pay the invoice"}, found)
summary = owner.call("board_summary", board="Ops")
ok("Models down: the summary is the plain facts", summary["source"] == "facts", summary)
ok("Models down: checklist says so plainly", "Models" in member.refused(
"suggest_checklist", card=a["number"], board="Ops"))
r = api("POST", f"/api/cards/{a['id']}/checklist/suggest", OWNER)
ok("Models down: the web door says 503", r.status_code == 503, r.status_code)
ok("Models down: the board still works",
member.call("add_card", board="Ops", title="Still here")["number"] > 0)
Path(os.environ["MODELS_DOWN_FLAG"]).unlink()
print("-- both doors see one truth, stored for real")
web_view = api("GET", f"/api/boards/{bid}", VIEWER, 200).json()
mcp_view = viewer.call("show_board", board=bid)
ok("web and MCP return the same board", web_view == mcp_view)
with psycopg.connect(os.environ["TEST_DATABASE_URL"]) as db:
rows = db.execute("SELECT count(*) FROM activity WHERE board_id = %s", (bid,)).fetchone()[0]
kept = db.execute("SELECT count(*) FROM cards WHERE board_id = %s AND archived",
(bid,)).fetchone()[0]
vectors = db.execute("SELECT count(*) FROM card_embeddings WHERE board_id = %s",
(bid,)).fetchone()[0]
ok(f"{rows} events in the activity table", rows > 30, rows)
ok("archived cards keep their row", kept == 24, kept)
ok(f"{vectors} embeddings stored", vectors > 3, vectors)
print("-- the database drops every connection")
with psycopg.connect(os.environ["TEST_DATABASE_URL"], autocommit=True) as db:
gone = db.execute("SELECT count(pg_terminate_backend(pid)) FROM pg_stat_activity "
"WHERE pid <> pg_backend_pid() AND backend_type = 'client backend'"
).fetchone()[0]
ok(f"{gone} app connections killed", gone > 0, gone)
for _ in range(8): # enough requests to reach the server's worker threads
api("GET", f"/api/boards/{bid}", OWNER, 200)
ok("the app reconnected and kept answering")
api("DELETE", f"/api/boards/{bid}/members/{MEMBER}", OWNER, 200)
api("GET", f"/api/boards/{bid}", MEMBER, 404)
ok("a removed member loses the board at once", "no such" in member.refused(
"show_board", board=bid))
print(f"\nEND-TO-END: {checks} checks passed")
if __name__ == "__main__":
main()