contest/orbit-boards
Orbit Boards, a private, remixable team task board built with FastAPI, DataK3, AppID, Models and MCP.
A remix template · 0 remixes · Remix contest/orbit-boards
tests/test_boards.py
"""Permission and workflow tests for shared use cases, web routes and MCP actions."""
import copy
import json
from datetime import date, timedelta
import psycopg
import pytest
from fastapi.testclient import TestClient
from boards.application.use_cases import Boards, NotFound
from boards.domain.model import Forbidden, Invalid, Person
from boards.infrastructure import repository
from boards.main import create_app
OWNER, MEMBER, VIEWER, STRANGER = (Person(email) for email in
("ana@example.com", "bo@example.com", "cy@example.com", "dee@example.com"))
class MemoryBoards:
"""Isolation-friendly repository fake with the same atomic write contract as DataK3."""
def __init__(self):
self.boards, self.cards, self.events = {}, {}, []
def get(self, board_id):
return copy.deepcopy(self.boards.get(board_id))
def boards_for(self, email):
email = email.lower()
return [copy.deepcopy(board) for board in self.boards.values()
if board.owner_email == email or email in board.members]
def save_board(self, board):
self.boards[board.id] = copy.deepcopy(board)
def save_board_event(self, board, event):
self.save_board(board)
self.events.append({**copy.deepcopy(event), "event_at": "2026-01-01T00:00:00+00:00"})
def get_card(self, card_id):
return copy.deepcopy(self.cards.get(card_id))
def cards_on(self, board_id):
return [copy.deepcopy(card) for card in self.cards.values() if card.board_id == board_id]
def save_card(self, card):
self.cards[card.id] = copy.deepcopy(card)
def save_cards_and_events(self, cards, events):
for card in cards:
self.save_card(card)
for event in events:
self.events.append({**copy.deepcopy(event), "event_at": "2026-01-01T00:00:00+00:00"})
def log(self, board_id, card_id, actor, action):
self.events.append({"board_id": board_id, "card_id": card_id, "actor": actor,
"action": action, "event_at": "2026-01-01T00:00:00+00:00"})
def activity_for(self, board_id, limit=50):
selected = [event for event in self.events if event["board_id"] == board_id]
return list(reversed(selected[-limit:]))
def check_ready(self):
return None
class FakeAI:
def checklist(self, title):
return [f"Prepare {title}", "Review the result", "Close the task"]
@pytest.fixture
def svc():
repo = MemoryBoards()
service = Boards(repo, FakeAI())
board = service.create_board(OWNER, "Launch")
service.add_member(OWNER, board["id"], MEMBER.email, "member")
service.add_member(OWNER, board["id"], VIEWER.email, "viewer")
return service, repo, board["id"]
def test_member_can_create_and_move_cards_and_changes_are_logged(svc):
service, repo, board_id = svc
card = service.add_card(MEMBER, board_id, "Send invoice")
moved = service.move_card(MEMBER, card["id"], "Done")
assert moved["list_name"] == "Done"
assert any(event.get("field") == "list_name" and event["actor"] == MEMBER.email
for event in repo.events)
def test_owner_member_and_viewer_roles_are_enforced(svc):
service, _, board_id = svc
assert service.show_board(OWNER, board_id)["role"] == "owner"
assert service.show_board(MEMBER, board_id)["role"] == "member"
assert service.show_board(VIEWER, board_id)["role"] == "viewer"
with pytest.raises(Forbidden):
service.add_card(VIEWER, board_id, "no edit")
with pytest.raises(Forbidden):
service.add_member(MEMBER, board_id, "eve@example.com", "member")
with pytest.raises(Forbidden):
service.move_card(VIEWER, service.add_card(MEMBER, board_id, "read-only test")["id"], "Done")
def test_board_privacy_hides_other_boards_by_id(svc):
service, _, board_id = svc
with pytest.raises(NotFound, match="no such board"):
service.show_board(STRANGER, board_id)
with pytest.raises(NotFound, match="no such board"):
service.get_activity(STRANGER, board_id)
assert service.my_boards(STRANGER) == []
def test_owner_is_the_only_owner_and_member_email_is_validated(svc):
service, _, board_id = svc
with pytest.raises(Invalid, match="one owner"):
service.add_member(OWNER, board_id, "new@example.com", "owner")
with pytest.raises(Invalid, match="valid member email"):
service.add_member(OWNER, board_id, "not-an-email", "member")
def test_card_fields_and_assignment_are_validated_and_audited(svc):
service, repo, board_id = svc
card = service.add_card(MEMBER, board_id, "Prepare release", description="Ship notes",
assignee=OWNER.email, due_date="2026-12-01", labels=["launch", "qa"])
assert card["description"] == "Ship notes"
assert card["assignee"] == OWNER.email
assert card["labels"] == ["launch", "qa"]
assert card["due_date"] == "2026-12-01"
updated = service.update_card(MEMBER, card["id"], title="Release notes", labels=["ready"])
assert updated["title"] == "Release notes"
assert updated["labels"] == ["ready"]
changed = [event for event in repo.events if event.get("card_id") == card["id"]]
assert {event["field"] for event in changed if event.get("field")} >= {"title", "labels"}
assert all(event["actor"] == MEMBER.email for event in changed)
def test_assignee_must_be_on_the_same_board_and_rejection_does_not_mutate(svc):
service, repo, board_id = svc
card = service.add_card(OWNER, board_id, "Plan launch")
with pytest.raises(Invalid, match="member of this board"):
service.assign_card(OWNER, card["id"], STRANGER.email)
assert repo.get_card(card["id"]).assignee is None
with pytest.raises(Invalid):
service.update_card(OWNER, card["id"], title=" ")
assert repo.get_card(card["id"]).title == "Plan launch"
def test_move_rejects_unknown_list_and_supports_stable_reordering(svc):
service, repo, board_id = svc
first = service.add_card(MEMBER, board_id, "First")
second = service.add_card(MEMBER, board_id, "Second")
third = service.add_card(MEMBER, board_id, "Third")
with pytest.raises(Invalid):
service.move_card(MEMBER, first["id"], "Someday")
service.move_card(MEMBER, third["id"], "To do", position=0)
ordered = sorted(repo.cards_on(board_id), key=lambda card: card.position)
assert [card.title for card in ordered] == ["Third", "First", "Second"]
service.move_card(MEMBER, first["id"], "Doing")
assert repo.get_card(first["id"]).list_name == "Doing"
def test_my_cards_and_overdue_skip_undated_and_completed_cards(svc):
service, _, board_id = svc
late = service.add_card(MEMBER, board_id, "Overdue", assignee=MEMBER.email,
due_date=(date.today() - timedelta(days=1)).isoformat())
soon = service.add_card(MEMBER, board_id, "Due soon", assignee=MEMBER.email,
due_date=(date.today() + timedelta(days=4)).isoformat())
service.add_card(MEMBER, board_id, "No deadline", assignee=MEMBER.email)
done = service.add_card(MEMBER, board_id, "Finished", assignee=MEMBER.email,
due_date=(date.today() - timedelta(days=2)).isoformat(), list_name="Done")
assert {card["id"] for card in service.list_my_cards(MEMBER)} == {late["id"], soon["id"]}
assert {card["id"] for card in service.list_overdue(MEMBER)} == {late["id"]}
assert done["id"] not in {card["id"] for card in service.list_overdue(MEMBER)}
assert {card["id"] for card in service.list_overdue(OWNER, board_id)} == {late["id"]}
def test_activity_is_scoped_and_returns_actor_and_field_values(svc):
service, _, board_id = svc
card = service.add_card(MEMBER, board_id, "Initial title")
service.update_card(MEMBER, card["id"], title="Updated title")
events = service.get_activity(VIEWER, board_id)
assert events[0]["actor"] == MEMBER.email
assert events[0]["field"] == "title"
assert events[0]["old_value"] == "Initial title"
assert events[0]["new_value"] == "Updated title"
def test_invalid_card_creation_and_invalid_limits_are_rejected(svc):
service, _, board_id = svc
with pytest.raises(Invalid, match="needs a title"):
service.add_card(OWNER, board_id, " ")
with pytest.raises(Invalid, match="days"):
service.list_my_cards(OWNER, 999)
with pytest.raises(Invalid, match="limit"):
service.get_activity(OWNER, board_id, 101)
def _headers(person):
return {"X-Dodil-User": json.dumps({"sub": person.email, "email": person.email})}
def test_web_create_board_persists_owner_and_validates_title():
repo = MemoryBoards()
service = Boards(repo, FakeAI())
with TestClient(create_app(service)) as client:
assert client.post("/api/boards", json={"title": "Roadmap"}).status_code == 401
created = client.post("/api/boards", headers=_headers(OWNER), json={"title": " Roadmap "})
assert created.status_code == 201, created.text
board_id = created.json()["id"]
assert created.json()["title"] == "Roadmap"
assert repo.boards[board_id].owner_email == OWNER.email
assert client.get("/api/boards", headers=_headers(OWNER)).json() == [
{"id": board_id, "title": "Roadmap", "role": "owner"}]
assert any(event["board_id"] == board_id and event["actor"] == OWNER.email
for event in repo.events)
assert client.post("/api/boards", headers=_headers(OWNER),
json={"title": " "}).status_code == 422
assert client.post("/api/boards", headers=_headers(OWNER),
json={"title": "x" * 121}).status_code == 422
assert len(repo.boards) == 1
def test_web_database_failure_is_sanitized_and_retryable():
class BrokenRepository(MemoryBoards):
def save_board_event(self, board, event):
raise psycopg.OperationalError("private database diagnostic")
service = Boards(BrokenRepository(), FakeAI())
with TestClient(create_app(service)) as client:
response = client.post("/api/boards", headers=_headers(OWNER), json={"title": "Roadmap"})
assert response.status_code == 503
assert "temporarily unavailable" in response.json()["detail"]
assert "private database diagnostic" not in response.text
def test_readyz_checks_datastore_without_exposing_internal_errors():
class UnreadyRepository(MemoryBoards):
def check_ready(self):
raise RuntimeError("private schema detail")
with TestClient(create_app(Boards(UnreadyRepository(), FakeAI()))) as client:
assert client.get("/healthz").json() == {"ok": True}
response = client.get("/readyz")
assert response.status_code == 503
assert response.json() == {"ok": False, "reason": "data store unavailable"}
assert "private schema detail" not in response.text
def test_sql_readiness_probe_covers_repository_schema(monkeypatch):
statements = []
class Connection:
def __enter__(self):
return self
def __exit__(self, *_):
return False
def execute(self, statement):
statements.append(statement)
monkeypatch.setattr(repository, "connect", lambda: Connection())
repository.SqlBoards().check_ready()
assert len(statements) == 4
assert any("description" in query and "position" in query for query in statements)
assert any("field" in query and "old_value" in query and "new_value" in query
for query in statements)
def test_web_routes_enforce_permissions_and_share_persistence(svc):
service, repo, board_id = svc
app = create_app(service)
with TestClient(app) as client:
assert client.get("/api/boards").status_code == 401
assert client.get("/api/boards", headers=_headers(VIEWER)).json()[0]["role"] == "viewer"
response = client.post(f"/api/boards/{board_id}/cards", json={"title": "forbidden"},
headers=_headers(VIEWER))
assert response.status_code == 403
created = client.post(f"/api/boards/{board_id}/cards", headers=_headers(MEMBER), json={
"title": "Route card", "description": "A real update", "labels": ["ops"],
"due_date": "2026-12-01"})
assert created.status_code == 201, created.text
card_id = created.json()["id"]
moved = client.post(f"/api/cards/{card_id}/move", headers=_headers(MEMBER),
json={"list_name": "Doing"})
assert moved.status_code == 200 and moved.json()["list_name"] == "Doing"
updated = client.patch(f"/api/cards/{card_id}", headers=_headers(MEMBER),
json={"title": "Changed by web"})
assert updated.status_code == 200
assert repo.get_card(card_id).title == "Changed by web"
assert client.get(f"/api/boards/{board_id}/activity", headers=_headers(VIEWER)).status_code == 200
assert client.get(f"/api/boards/{board_id}", headers=_headers(STRANGER)).status_code == 404
assert client.get("/", headers=_headers(OWNER)).status_code == 200
def test_mcp_actions_share_use_cases_and_reject_viewer_mutations(svc):
service, repo, board_id = svc
with TestClient(create_app(service)) as client:
missing = client.post("/mcp", json={"jsonrpc": "2.0", "id": 1, "method": "tools/list"})
assert missing.status_code == 401 and "resource_metadata" in missing.headers["www-authenticate"]
headers = {**_headers(MEMBER), "accept": "application/json, text/event-stream"}
initialize = {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {
"protocolVersion": "2025-06-18", "capabilities": {},
"clientInfo": {"name": "test", "version": "1"}}}
assert client.post("/mcp", json=initialize, headers=headers).status_code == 200
tools = client.post("/mcp", headers=headers,
json={"jsonrpc": "2.0", "id": 2, "method": "tools/list"}).json()
names = {item["name"] for item in tools["result"]["tools"]}
assert {"list_boards", "list_my_cards", "list_overdue", "add_card", "move_card",
"assign_card", "update_card", "get_activity", "add_member"} <= names
add_call = {"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {
"name": "add_card", "arguments": {"board_id": board_id, "title": "via MCP",
"description": "Persistent", "labels": ["assistant"]}}}
result = client.post("/mcp", json=add_call, headers=headers).json()
assert not result["result"].get("isError"), result
card = next(card for card in repo.cards.values() if card.title == "via MCP")
viewer_headers = {**_headers(VIEWER), "accept": "application/json, text/event-stream"}
viewer_init = {**initialize, "id": 4, "params": {**initialize["params"],
"clientInfo": {"name": "viewer", "version": "1"}}}
assert client.post("/mcp", json=viewer_init, headers=viewer_headers).status_code == 200
mutation = {"jsonrpc": "2.0", "id": 5, "method": "tools/call", "params": {
"name": "move_card", "arguments": {"card_id": card.id, "list_name": "Done"}}}
denied = client.post("/mcp", json=mutation, headers=viewer_headers).json()
assert denied["result"]["isError"] is True
assert repo.get_card(card.id).list_name == "To do"