eduard/cobra-boards
Cobra Boards, team task boards you can run from Claude or ChatGPT. FastAPI + DataK3 + appid + Dodil Models + MCP.
A remix template · 1 remixes · Remix eduard/cobra-boards
tests/test_v2.py
"""v2: comments with @mentions, notifications, the live board stamp, files on cards, archive and
restore, board templates, and the MCP prompts. Same rules as everything else: viewers read,
members write, strangers see nothing, and every change leaves a line in the history."""
import json
import pytest
from conftest import MEMBER, OWNER, STRANGER, TODAY, VIEWER, headers
from fastapi.testclient import TestClient
from test_doors import MCP_ACCEPT, mcp
from boards.application.use_cases import Boards, NotFound
from boards.domain.model import Forbidden, Invalid, clean_filename, mentions_in
from boards.infrastructure.memory import MemoryBoards
from boards.main import create_app
PEOPLE = [OWNER.email, MEMBER.email, VIEWER.email]
PNG = b"\x89PNG\r\n\x1a\n" + b"\x00" * 64
# ---------------------------------------------------------------- templates
def test_templates_shape_the_first_lists(svc):
s, _ = svc
names = {t["name"] for t in s.templates()}
assert {"basic", "sprint", "launch", "onboarding", "support"} <= names
b = s.create_board(OWNER, "Sprint 12", template="Sprint")
assert b["lists"] == ["Backlog", "This sprint", "In progress", "Review", "Done"]
done = [l for l in s.show_board(OWNER, b["id"])["lists"] if l["is_done"]]
assert [l["name"] for l in done] == ["Done"]
with pytest.raises(Invalid, match="no template"):
s.create_board(OWNER, "x", template="kanban-pro")
# ---------------------------------------------------------------- mentions
@pytest.mark.parametrize("text, who", [
("@bo can you check?", [MEMBER.email]),
("thanks @bo@cobra.test", [MEMBER.email]),
("@ana and @cy, please", [OWNER.email, VIEWER.email]),
("@BO, look", [MEMBER.email]),
("@zed is not here", []),
("mail me at ana@cobra.test", []), # an address inside text is not a mention
("@bo @bo twice", [MEMBER.email]),
])
def test_mentions_find_people_on_the_board(text, who):
assert mentions_in(text, PEOPLE) == who
def test_an_ambiguous_mention_mentions_nobody():
assert mentions_in("@bo hi", ["bob@x.test", "bonnie@x.test"]) == []
assert mentions_in("@bob hi", ["bob@x.test", "bonnie@x.test"]) == ["bob@x.test"]
@pytest.mark.parametrize("raw, clean", [("report.pdf", "report.pdf"),
("../../etc/passwd", "passwd"),
("C:\\Users\\me\\plan v2.docx", "plan v2.docx"),
('bad"<name>.txt', "badname.txt")])
def test_file_names_are_made_safe(raw, clean):
assert clean_filename(raw) == clean
@pytest.mark.parametrize("raw", ["", " ", "..", "folder/"])
def test_a_file_needs_a_name(raw):
with pytest.raises(Invalid):
clean_filename(raw)
# ---------------------------------------------------------------- comments & notifications
def test_members_comment_viewers_read_strangers_see_nothing(svc):
s, bid = svc
card = s.add_card(OWNER, bid, "Write copy")
out = s.add_comment(MEMBER, card["id"], "Draft is in the doc")
assert out["comment"]["author"] == MEMBER.email
assert [c["body"] for c in s.card_detail(VIEWER, card["id"])["comments"]] == ["Draft is in the doc"]
with pytest.raises(Forbidden):
s.add_comment(VIEWER, card["id"], "can I?")
with pytest.raises(NotFound):
s.add_comment(STRANGER, card["id"], "hello")
with pytest.raises(NotFound):
s.comments(STRANGER, card["id"])
for bad in ("", " ", "x" * 2001):
with pytest.raises(Invalid):
s.add_comment(MEMBER, card["id"], bad)
assert s.card_history(OWNER, card["id"])[0]["action"] == "commented"
def test_a_mention_notifies_the_person_and_only_them(svc):
s, bid = svc
card = s.add_card(OWNER, bid, "Invoice Acme")
out = s.add_comment(OWNER, card["id"], "@bo please send it today, cc @ana")
assert out["mentioned"] == [MEMBER.email, OWNER.email] and out["notified"] == 1 # not myself
n = s.my_notifications(MEMBER)
assert n["unread"] == 1 and n["items"][0]["kind"] == "mention"
assert n["items"][0]["actor"] == OWNER.email and n["items"][0]["card"] == 1
assert s.my_notifications(OWNER)["items"] == []
assert s.my_notifications(VIEWER)["items"] == []
def test_the_assignee_hears_about_comments_on_their_card(svc):
s, bid = svc
card = s.add_card(OWNER, bid, "QA pass", assignee=MEMBER.email)
kinds = lambda: [i["kind"] for i in s.my_notifications(MEMBER)["items"]]
assert kinds() == ["assigned"] # a card made for you by someone else
s.add_comment(OWNER, card["id"], "Found two bugs")
assert kinds() == ["comment", "assigned"]
s.add_comment(OWNER, card["id"], "@bo see above") # mentioned: one notification, not two
assert kinds() == ["mention", "comment", "assigned"]
s.add_comment(MEMBER, card["id"], "On it") # your own comment tells you nothing
assert len(kinds()) == 3
def test_being_assigned_by_someone_else_is_a_notification(svc):
s, bid = svc
card = s.add_card(OWNER, bid, "Plan")
s.update_card(OWNER, card["id"], assignee=MEMBER.email)
item = s.my_notifications(MEMBER)["items"][0]
assert item["kind"] == "assigned" and "assigned you #1" in item["summary"]
s.update_card(OWNER, card["id"], assignee="me") # taking it yourself tells nobody
assert s.my_notifications(OWNER)["items"] == []
def test_notifications_can_be_marked_read_one_or_all(svc):
s, bid = svc
for t in "abc":
c = s.add_card(OWNER, bid, t)
s.add_comment(OWNER, c["id"], "@bo ping")
items = s.my_notifications(MEMBER)["items"]
assert len(items) == 3
assert s.mark_notifications_read(MEMBER, [items[0]["id"]])["marked"] == 1
assert s.my_notifications(MEMBER)["unread"] == 2
assert len(s.my_notifications(MEMBER, unread_only=True)["items"]) == 2
assert s.mark_notifications_read(MEMBER)["marked"] == 2
assert s.my_notifications(MEMBER)["unread"] == 0
assert s.mark_notifications_read(VIEWER, [items[1]["id"]])["marked"] == 0 # not theirs
def test_notifications_from_a_board_you_left_disappear(svc):
s, bid = svc
c = s.add_card(OWNER, bid, "x")
s.add_comment(OWNER, c["id"], "@bo ping")
s.remove_member(OWNER, bid, MEMBER.email)
assert s.my_notifications(MEMBER)["items"] == []
# ---------------------------------------------------------------- live stamp
def test_the_stamp_moves_when_the_board_changes(svc):
s, bid = svc
first = s.board_stamp(VIEWER, bid)
assert s.board_stamp(VIEWER, bid)["stamp"] == first["stamp"] # nothing changed
card = s.add_card(MEMBER, bid, "New idea")
after = s.board_stamp(VIEWER, bid)
assert after["stamp"] != first["stamp"] and after["by"] == MEMBER.email
assert "New idea" in after["summary"] and after["card_id"] == card["id"]
s.move_card(OWNER, card["id"], "Done")
assert s.board_stamp(VIEWER, bid)["by"] == OWNER.email
with pytest.raises(NotFound):
s.board_stamp(STRANGER, bid)
# ---------------------------------------------------------------- attachments
def test_files_go_to_object_storage_and_back(svc):
s, bid = svc
card = s.add_card(OWNER, bid, "Brand kit")
out = s.add_attachment(MEMBER, card["id"], "logo.png", PNG, "image/png")
att = out["attachment"]
assert att["name"] == "logo.png" and att["size"] == len(PNG)
assert len(s.objects.blobs) == 1
got = s.download_attachment(VIEWER, att["id"])
assert got["data"] == PNG and got["inline"] is True and got["content_type"] == "image/png"
assert s.card_detail(VIEWER, card["id"])["attachments"][0]["name"] == "logo.png"
assert s.card_history(OWNER, card["id"])[0]["action"] == "attached"
def test_a_file_that_could_run_in_the_page_is_only_downloaded(svc):
s, bid = svc
card = s.add_card(OWNER, bid, "x")
att = s.add_attachment(OWNER, card["id"], "page.html", b"<script>alert(1)</script>",
"text/html")["attachment"]
got = s.download_attachment(MEMBER, att["id"])
assert got["inline"] is False and got["content_type"] == "application/octet-stream"
def test_file_rules(svc):
s, bid = svc
card = s.add_card(OWNER, bid, "x")
with pytest.raises(Forbidden):
s.add_attachment(VIEWER, card["id"], "a.txt", b"hi")
with pytest.raises(NotFound):
s.add_attachment(STRANGER, card["id"], "a.txt", b"hi")
with pytest.raises(Invalid, match="empty"):
s.add_attachment(OWNER, card["id"], "a.txt", b"")
with pytest.raises(Invalid, match="10 MB"):
s.add_attachment(OWNER, card["id"], "big.bin", b"0" * (10 * 1024 * 1024 + 1))
for i in range(20):
s.add_attachment(OWNER, card["id"], f"f{i}.txt", b"x")
with pytest.raises(Invalid, match="at most 20"):
s.add_attachment(OWNER, card["id"], "one-more.txt", b"x")
def test_a_stranger_cannot_fetch_a_file_even_with_its_id(svc):
s, bid = svc
card = s.add_card(OWNER, bid, "x")
att = s.add_attachment(OWNER, card["id"], "secret.pdf", b"%PDF", "application/pdf")["attachment"]
with pytest.raises(NotFound):
s.download_attachment(STRANGER, att["id"])
with pytest.raises(NotFound):
s.download_attachment(OWNER, "no-such-id")
def test_removing_a_file_removes_the_bytes_and_writes_history(svc):
s, bid = svc
card = s.add_card(OWNER, bid, "x")
att = s.add_attachment(OWNER, card["id"], "old.txt", b"x")["attachment"]
with pytest.raises(Forbidden):
s.delete_attachment(VIEWER, att["id"])
assert s.delete_attachment(MEMBER, att["id"]) == {"removed": "old.txt"}
assert s.objects.blobs == {} and s.card_detail(OWNER, card["id"])["attachments"] == []
assert "removed the file 'old.txt'" in s.card_history(OWNER, card["id"])[0]["summary"]
def test_without_object_storage_files_are_refused_politely():
s = Boards(MemoryBoards(), None, today=lambda: TODAY)
bid = s.create_board(OWNER, "B")["id"]
card = s.add_card(OWNER, bid, "x")
with pytest.raises(Invalid, match="not configured"):
s.add_attachment(OWNER, card["id"], "a.txt", b"hi")
# ---------------------------------------------------------------- archive
def test_an_archived_card_leaves_every_view_but_keeps_its_history(svc):
s, bid = svc
ship = s.add_card(OWNER, bid, "Ship", assignee=MEMBER.email, due_date="2026-11-01")
qa = s.add_card(OWNER, bid, "QA", due_date="2026-11-02")
s.add_blocker(OWNER, ship["id"], qa["id"])
s.archive_card(MEMBER, qa["id"])
titles = [c["title"] for l in s.show_board(VIEWER, bid)["lists"] for c in l["cards"]]
assert titles == ["Ship"]
assert [c["title"] for c in s.overdue(OWNER)] == ["Ship"]
assert s.search_cards(OWNER, "QA") == []
assert s.what_is_blocking(OWNER, ship["id"])["blocked"] is False
s.archive_card(OWNER, ship["id"])
assert s.my_cards(MEMBER)["overdue"] == []
detail = s.card_detail(VIEWER, qa["id"])
assert detail["archived"] is True and detail["archived_by"] == MEMBER.email
assert [c["number"] for c in s.archived_cards(VIEWER, bid)] == [1, 2]
assert s.card_history(OWNER, qa["id"])[0]["action"] == "archived"
def test_an_archived_card_is_frozen_until_restored(svc):
s, bid = svc
card = s.add_card(OWNER, bid, "Old", list_name="Doing")
s.archive_card(OWNER, card["id"])
assert s.archive_card(OWNER, card["id"])["already"] is True
with pytest.raises(Invalid, match="restore it first"):
s.move_card(OWNER, card["id"], "Done")
with pytest.raises(Invalid, match="restore it first"):
s.update_card(OWNER, card["id"], title="New")
out = s.restore_card(MEMBER, card["id"])
assert out["list"] == "Doing" and out["archived"] is False
assert s.move_card(OWNER, card["id"], "Done")["list"] == "Done"
with pytest.raises(Invalid, match="not archived"):
s.restore_card(OWNER, card["id"])
assert s.card_history(OWNER, card["id"])[1]["action"] == "restored"
def test_viewers_cannot_archive_or_restore(svc):
s, bid = svc
card = s.add_card(OWNER, bid, "x")
with pytest.raises(Forbidden):
s.archive_card(VIEWER, card["id"])
s.archive_card(OWNER, card["id"])
with pytest.raises(Forbidden):
s.restore_card(VIEWER, card["id"])
with pytest.raises(NotFound):
s.archived_cards(STRANGER, bid)
def test_archived_cards_are_not_suggested_as_similar(svc):
s, bid = svc
target = s.add_card(OWNER, bid, "Send invoice to Acme")
old = s.add_card(OWNER, bid, "Chase unpaid bill")
s.add_card(OWNER, bid, "Design the logo")
s.archive_card(OWNER, old["id"])
titles = [c["title"] for c in s.similar_cards(MEMBER, target["id"])["cards"]]
assert "Chase unpaid bill" not in titles
# ---------------------------------------------------------------- the web door
@pytest.fixture
def client(svc):
s, bid = svc
with TestClient(create_app(s)) as c:
yield c, s, bid
def test_web_comments_and_notifications(client):
c, s, bid = client
card = s.add_card(OWNER, bid, "Plan")
r = c.post(f"/api/cards/{card['id']}/comments", json={"body": "@bo over to you"},
headers=headers(OWNER))
assert r.status_code == 201 and r.json()["mentioned"] == [MEMBER.email]
assert c.post(f"/api/cards/{card['id']}/comments", json={"body": "x"},
headers=headers(VIEWER)).status_code == 403
assert c.get(f"/api/cards/{card['id']}/comments", headers=headers(STRANGER)).status_code == 404
n = c.get("/api/me/notifications", headers=headers(MEMBER)).json()
assert n["unread"] == 1
r = c.post("/api/me/notifications/read", json={}, headers=headers(MEMBER))
assert r.json() == {"marked": 1}
assert c.get("/api/me/notifications?unread=true", headers=headers(MEMBER)).json()["items"] == []
def test_web_stamp_and_archive_and_templates(client):
c, s, bid = client
st = c.get(f"/api/boards/{bid}/stamp", headers=headers(VIEWER)).json()
card = s.add_card(MEMBER, bid, "Old")
assert c.get(f"/api/boards/{bid}/stamp", headers=headers(VIEWER)).json()["stamp"] != st["stamp"]
assert c.post(f"/api/cards/{card['id']}/archive", headers=headers(VIEWER)).status_code == 403
assert c.post(f"/api/cards/{card['id']}/archive", headers=headers(MEMBER)).json()["archived"]
arch = c.get(f"/api/boards/{bid}/archived", headers=headers(VIEWER)).json()
assert [x["title"] for x in arch] == ["Old"]
assert c.post(f"/api/cards/{card['id']}/restore", headers=headers(MEMBER)).json()["list"] == "To do"
t = c.get("/api/templates", headers=headers(VIEWER)).json()
assert "sprint" in [x["name"] for x in t]
assert c.get("/api/templates").status_code == 401
b = c.post("/api/boards", json={"title": "Launch Q1", "template": "launch"},
headers=headers(MEMBER)).json()
assert b["lists"][-1] == "Launched"
def test_web_upload_download_delete_a_file(client):
c, s, bid = client
card = s.add_card(OWNER, bid, "Kit")
r = c.post(f"/api/cards/{card['id']}/attachments?name=logo.png", content=PNG,
headers={**headers(MEMBER), "content-type": "image/png"})
assert r.status_code == 201
att = r.json()["attachment"]
got = c.get(att["url"], headers=headers(VIEWER))
assert got.status_code == 200 and got.content == PNG
assert got.headers["content-type"] == "image/png"
assert got.headers["x-content-type-options"] == "nosniff"
assert got.headers["content-disposition"].startswith("inline")
assert c.get(att["url"], headers=headers(STRANGER)).status_code == 404
assert c.post(f"/api/cards/{card['id']}/attachments?name=a.txt", content=b"x",
headers={**headers(VIEWER), "content-type": "text/plain"}).status_code == 403
assert c.delete(att["url"], headers=headers(VIEWER)).status_code == 403
assert c.delete(att["url"], headers=headers(MEMBER)).json() == {"removed": "logo.png"}
assert c.get(att["url"], headers=headers(MEMBER)).status_code == 404
def test_web_html_files_download_and_big_files_are_refused(client):
c, s, bid = client
card = s.add_card(OWNER, bid, "x")
att = c.post(f"/api/cards/{card['id']}/attachments?name=x.html", content=b"<b>hi</b>",
headers={**headers(OWNER), "content-type": "text/html"}).json()["attachment"]
got = c.get(att["url"], headers=headers(OWNER))
assert got.headers["content-type"] == "application/octet-stream"
assert got.headers["content-disposition"].startswith("attachment")
big = c.post(f"/api/cards/{card['id']}/attachments?name=big.bin", content=b"0" * (10 * 1024 * 1024 + 10),
headers={**headers(OWNER), "content-type": "application/octet-stream"})
assert big.status_code == 413
# ---------------------------------------------------------------- the MCP door
def test_mcp_comment_with_a_mention_and_the_notification(client):
c, s, bid = client
s.add_card(OWNER, bid, "Invoice Acme")
err, out = mcp(c, OWNER, "add_comment", card="1", board="Launch", text="@bo can you send it?")
assert not err and "Mentioned bo@cobra.test" in out and "1 person(s) notified" in out
err, out = mcp(c, MEMBER, "my_notifications")
assert not err and out.startswith("1 unread") and "mentioned you on #1" in out
err, out = mcp(c, MEMBER, "mark_notifications_read")
assert "Marked 1" in out
err, out = mcp(c, MEMBER, "my_notifications")
assert out == "No notifications. You're all caught up."
err, out = mcp(c, VIEWER, "add_comment", card="1", board="Launch", text="me too")
assert err and "Not allowed" in out
def test_mcp_archive_restore_and_show_card(client):
c, s, bid = client
card = s.add_card(OWNER, bid, "Old idea")
s.add_comment(OWNER, card["id"], "keep it for later")
s.add_attachment(OWNER, card["id"], "notes.txt", b"x" * 2048, "text/plain")
err, out = mcp(c, MEMBER, "archive_card", card="1", board="Launch")
assert not err and out.startswith("Archived #1")
err, out = mcp(c, VIEWER, "show_card", card="1", board="Launch")
for bit in ("ARCHIVED by bo@cobra.test", "Files: notes.txt (2 KB)", "Comments (1):",
"keep it for later"):
assert bit in out, bit
err, out = mcp(c, VIEWER, "archived_cards", board="Launch")
assert "#1 Old idea" in out
err, out = mcp(c, MEMBER, "restore_card", card="1", board="Launch")
assert out == "Restored #1 Old idea to To do."
def test_mcp_board_from_a_template(client):
c, _, _ = client
err, out = mcp(c, MEMBER, "create_board", title="Help desk", template="support")
assert not err and "New, Investigating, Waiting on customer, Resolved" in out
def _rpc(c, person, method, params=None):
h = {**headers(person), **MCP_ACCEPT}
c.post("/mcp", headers=h, json={"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {
"protocolVersion": "2025-06-18", "capabilities": {},
"clientInfo": {"name": "t", "version": "0"}}})
return c.post("/mcp", headers=h, json={"jsonrpc": "2.0", "id": 2, "method": method,
"params": params or {}}).json()["result"]
def test_mcp_offers_ready_made_prompts(client):
c, _, _ = client
names = {p["name"] for p in _rpc(c, MEMBER, "prompts/list")["prompts"]}
assert {"weekly_review", "standup", "plan_goal"} <= names
got = _rpc(c, MEMBER, "prompts/get", {"name": "plan_goal",
"arguments": {"board": "Launch", "goal": "Ship v2"}})
text = json.dumps(got)
assert "Launch" in text and "Ship v2" in text and "Only after I say yes" in text
def test_the_board_shows_how_many_comments_and_files_a_card_has(svc):
s, bid = svc
a = s.add_card(OWNER, bid, "a")
s.add_card(OWNER, bid, "b")
s.add_comment(OWNER, a["id"], "one")
s.add_comment(MEMBER, a["id"], "two")
s.add_attachment(OWNER, a["id"], "x.txt", b"x")
cards = {c["title"]: c for l in s.show_board(VIEWER, bid)["lists"] for c in l["cards"]}
assert (cards["a"]["comments"], cards["a"]["files"]) == (2, 1)
assert (cards["b"]["comments"], cards["b"]["files"]) == (0, 0)
def test_a_model_that_does_not_answer_is_a_clear_message_not_a_crash():
from boards.application.ports import AssistantUnavailable
class Down:
def checklist(self, title, description):
raise AssistantUnavailable("503")
def summary(self, board_title, facts):
raise AssistantUnavailable("503")
def embed(self, text):
raise AssistantUnavailable("timeout")
s = Boards(MemoryBoards(), Down(), today=lambda: TODAY)
bid = s.create_board(OWNER, "B")["id"]
card = s.add_card(OWNER, bid, "Plan")
with pytest.raises(Invalid, match="try the suggestion again"):
s.suggest_checklist(OWNER, card["id"])
with pytest.raises(Invalid, match="plain search still works"):
s.similar_cards(OWNER, card["id"])
assert s.daily_summary(OWNER, bid)["summary"].startswith("Date:") # falls back to the facts
assert s.search_cards(OWNER, "plan") # and the board itself keeps working