mobiledev/taskflow
TaskFlow, a team task board you run from the web or from Claude and ChatGPT. FastAPI + DataK3 + appid + Models + MCP.
A remix template · 0 remixes · Remix mobiledev/taskflow
tests/test_doors.py
"""The web routes and the MCP tools are two doors onto the same use cases: same rules, both ways."""
import json
import pytest
from fastapi.testclient import TestClient
from conftest import MEMBER, OWNER, STRANGER, VIEWER
from taskflow.main import create_app
def as_(person):
return {"X-Dodil-User": json.dumps({"sub": person.email, "email": person.email})}
@pytest.fixture
def client(app):
with TestClient(create_app(app)) as c:
yield c
class Mcp:
"""A tiny MCP client speaking JSON-RPC over the app's /mcp endpoint, as one person."""
def __init__(self, client, person):
self.c, self.h = client, {**as_(person), "accept": "application/json, text/event-stream"}
init = {"jsonrpc": "2.0", "id": 0, "method": "initialize", "params": {
"protocolVersion": "2025-06-18", "capabilities": {},
"clientInfo": {"name": "test", "version": "0"}}}
assert self.c.post("/mcp", json=init, headers=self.h).status_code == 200
def tools(self):
out = self.c.post("/mcp", json={"jsonrpc": "2.0", "id": 1, "method": "tools/list"},
headers=self.h).json()
return {t["name"] for t in out["result"]["tools"]}
def call(self, name, **arguments):
out = self.c.post("/mcp", headers=self.h, json={
"jsonrpc": "2.0", "id": 2, "method": "tools/call",
"params": {"name": name, "arguments": arguments}}).json()["result"]
text = out["content"][0]["text"]
return {"error": text} if out.get("isError") else json.loads(text)
# --- web ---------------------------------------------------------------------------------
def test_web_needs_a_signed_in_person(client):
assert client.get("/api/boards").status_code == 401
page = client.get("/", follow_redirects=False)
assert page.status_code == 303 and "/.dodil/auth/login" in page.headers["location"]
assert client.get("/", headers=as_(OWNER)).text.startswith("<!doctype html>")
assert client.get("/healthz").json() == {"ok": True}
assert client.post("/").json() == {"ok": True, "tables": {}} # counts only, no rows
def test_web_roles_hold(client, ops):
card = client.post(f"/api/boards/{ops}/cards", json={"title": "Send invoice"},
headers=as_(MEMBER)).json()
assert card["number"] == 1
assert client.post(f"/api/boards/{ops}/cards", json={"title": "x"},
headers=as_(VIEWER)).status_code == 403
assert client.post(f"/api/cards/{card['id']}/move", json={"list_name": "Done"},
headers=as_(VIEWER)).status_code == 403
assert client.get(f"/api/boards/{ops}", headers=as_(STRANGER)).status_code == 404
assert client.get(f"/api/cards/{card['id']}", headers=as_(STRANGER)).status_code == 404
assert client.post(f"/api/boards/{ops}/members", json={"email": "z@x.io"},
headers=as_(MEMBER)).status_code == 403
assert client.post(f"/api/boards/{ops}/cards", json={"title": "x", "list_name": "Later"},
headers=as_(MEMBER)).status_code == 422
def test_web_refuses_changes_sent_from_other_sites(client, ops):
card = client.post(f"/api/boards/{ops}/cards", json={"title": "x"}, headers=as_(OWNER)).json()
url = f"/api/cards/{card['id']}/checklist/ai"
other_app = {**as_(OWNER), "sec-fetch-site": "same-site", # another *.ignite app
"origin": "https://someone-else-8080.ignite.dodil.cloud"}
assert client.post(url, headers=other_app).status_code == 403
assert client.post(url, headers={**as_(OWNER), "origin": "https://evil.example"}).status_code == 403
page = {**as_(OWNER), "sec-fetch-site": "same-origin", "origin": "http://testserver"}
assert client.post(url, headers=page).status_code == 200 # the page itself
assert client.post(url, headers=as_(OWNER)).status_code == 200 # no browser headers
assert client.get(f"/api/boards/{ops}", headers=other_app).status_code == 200 # reads ok
def test_web_drag_and_drop_and_partial_edit(client, ops):
ids = [client.post(f"/api/boards/{ops}/cards", json={"title": t}, headers=as_(OWNER))
.json()["id"] for t in ("a", "b")]
client.patch(f"/api/cards/{ids[0]}", json={"due_date": "2026-11-20"}, headers=as_(OWNER))
moved = client.post(f"/api/cards/{ids[0]}/move", json={"list_name": "Doing", "index": 0},
headers=as_(MEMBER)).json()
assert moved["list_name"] == "Doing" and moved["due_date"] == "2026-11-20"
cleared = client.patch(f"/api/cards/{ids[0]}", json={"due_date": None},
headers=as_(OWNER)).json()
assert cleared["due_date"] is None and cleared["title"] == "a" # untouched fields stay
history = client.get(f"/api/cards/{ids[0]}/history", headers=as_(VIEWER)).json()
assert [h["actor"] for h in history][:2] == [OWNER.email, MEMBER.email]
# --- MCP ---------------------------------------------------------------------------------
def test_mcp_without_sign_in_points_to_the_pool(client):
r = client.post("/mcp", json={"jsonrpc": "2.0", "id": 1, "method": "tools/list"})
assert r.status_code == 401 and "resource_metadata" in r.headers["www-authenticate"]
link = r.headers["www-authenticate"].split('resource_metadata="')[1].rstrip('"')
assert not link.endswith("/mcp") # Ignite garbles length-carrying replies under */mcp
for path in ("/.well-known/oauth-protected-resource",
"/.well-known/oauth-protected-resource/mcp"):
assert client.get(path).json()["resource"].endswith("/mcp")
def test_mcp_tools_are_actions_not_tables(client):
names = Mcp(client, OWNER).tools()
assert {"my_cards", "add_card", "move_card", "assign_card", "overdue", "card_history",
"what_blocks", "checklist_from_ai", "similar_cards", "board_summary"} <= names
assert not any("sql" in n or "table" in n or "query" in n for n in names)
assert "delete_board" not in names # deleting a whole board is web-only, on purpose
def test_mcp_speaks_like_people_do(client, app, ops):
bo = Mcp(client, MEMBER)
added = bo.call("add_card", board="Ops", title="Send invoice", list_name="Doing",
assignee=MEMBER.email, due_date="2026-11-04")
assert added["number"] == 1 and added["list_name"] == "Doing"
plate = bo.call("my_cards", due_within_days=7)
assert plate["count"] == 1 and plate["cards"][0]["title"] == "Send invoice"
assert bo.call("move_card", card="12", list_name="Done").get("error") # no card 12
assert bo.call("move_card", card="1", list_name="Done")["list_name"] == "Done"
history = bo.call("card_history", card="#1")
assert history["history"][0]["changes"] == {"list_name": ["Doing", "Done"]}
def test_mcp_only_sees_what_its_person_sees(client, app, ops):
app.add_card(OWNER, ops, "Payroll", due_date="2026-10-01")
stranger = Mcp(client, STRANGER)
assert stranger.call("my_boards") == {"boards": []}
assert stranger.call("overdue") == {"count": 0, "cards": []}
assert "no board" in stranger.call("show_board", board="Ops")["error"]
assert "no card" in stranger.call("move_card", card="1", list_name="Done")["error"]
viewer = Mcp(client, VIEWER)
assert [c["title"] for c in viewer.call("overdue")["cards"]] == ["Payroll"]
assert "viewers" in viewer.call("move_card", card="1", list_name="Done")["error"]