freelancer/flowboard

Boards starter for the Dodil Boards Challenge. FastAPI + DataK3 + appid + Models + MCP.

• dodil-challenge
• dodil-challenge-boards

A remix template · 0 remixes · Remix freelancer/flowboard

tests/test_boards.py

"""Test board rules through the use cases and web/MCP doors.

The repository and AI are in-memory fakes; no Dodil account is required.
"""

import json

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.main import create_app


OWNER, MEMBER, VIEWER, STRANGER = (
    Person(email)
    for email in ("ana@x.io", "bo@x.io", "cy@x.io", "dee@x.io")
)


class MemoryBoards:
    """In-memory repository for isolated tests."""

    def __init__(self):
        self.boards = {}
        self.cards = {}
        self.events = []

    def get(self, board_id):
        return self.boards.get(board_id)

    def boards_for(self, email):
        return [
            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] = board

    def get_card(self, card_id):
        return self.cards.get(card_id)

    def cards_on(self, board_id):
        return [
            card
            for card in self.cards.values()
            if card.board_id == board_id
        ]

    def save_card(self, card):
        self.cards[card.id] = card

    def log(self, board_id, card_id, actor, action):
        self.events.append((board_id, card_id, actor, action))

    def activity_for(self, board_id, limit=100):
        """Return recent events for one board, newest first."""
        events = [
            {
                "id": str(index),
                "board_id": event[0],
                "card_id": event[1],
                "actor": event[2],
                "action": event[3],
                "event_at": None,
            }
            for index, event in enumerate(self.events, start=1)
            if event[0] == board_id
        ]
        return list(reversed(events))[:limit]


class FakeAI:
    """Predictable AI fake for checklist tests."""

    def checklist(self, title):
        return ["step one", "step two"]


@pytest.fixture
def svc():
    """Create a board with one member and one viewer."""
    service = Boards(MemoryBoards(), 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, board["id"]


def test_member_adds_and_moves_cards_and_it_is_logged(svc):
    service, 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(
        "moved from To do to Done" in event[3]
        for event in service.repo.events
        if event[1] == card["id"]
    )


def test_viewer_cannot_edit(svc):
    service, board_id = svc

    with pytest.raises(Forbidden):
        service.add_card(VIEWER, board_id, "nope")


def test_only_owner_manages_members(svc):
    service, board_id = svc

    with pytest.raises(Forbidden):
        service.add_member(MEMBER, board_id, "eve@x.io", "member")


def test_stranger_cannot_see_board(svc):
    service, board_id = svc

    with pytest.raises(NotFound):
        service.show_board(STRANGER, board_id)

    assert service.my_boards(STRANGER) == []


def test_unknown_list_is_refused(svc):
    service, board_id = svc
    card = service.add_card(OWNER, board_id, "x")

    with pytest.raises(Invalid):
        service.move_card(OWNER, card["id"], "Someday")


def _h(person):
    """Build test identity headers for the identity door."""
    return {
        "X-Dodil-User": json.dumps(
            {"sub": person.email, "email": person.email}
        )
    }


def test_web_and_mcp_doors_share_the_rules(svc):
    service, board_id = svc

    with TestClient(create_app(service)) as client:
        assert client.get("/api/boards").status_code == 401

        response = client.get("/api/boards", headers=_h(VIEWER))
        assert response.status_code == 200
        assert response.json()[0]["role"] == "viewer"

        response = client.post(
            f"/api/boards/{board_id}/cards",
            json={"title": "t"},
            headers=_h(VIEWER),
        )
        assert response.status_code == 403

        # Unauthenticated MCP access must be rejected.
        response = client.post(
            "/mcp",
            json={
                "jsonrpc": "2.0",
                "id": 1,
                "method": "tools/list",
            },
        )
        assert response.status_code == 401
        assert "resource_metadata" in response.headers["www-authenticate"]

        mcp_headers = {
            **_h(MEMBER),
            "accept": "application/json, text/event-stream",
        }

        initialize = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-06-18",
                "capabilities": {},
                "clientInfo": {"name": "t", "version": "0"},
            },
        }

        response = client.post(
            "/mcp",
            json=initialize,
            headers=mcp_headers,
        )
        assert response.status_code == 200

        # Create a card through MCP.
        add_call = {
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/call",
            "params": {
                "name": "add_card",
                "arguments": {
                    "board_id": board_id,
                    "title": "via mcp",
                },
            },
        }

        response = client.post(
            "/mcp",
            json=add_call,
            headers=mcp_headers,
        )
        output = response.json()
        assert not output["result"].get("isError"), output

        created_card = next(
            card
            for card in service.repo.cards.values()
            if card.title == "via mcp"
        )

        # Update the card through MCP.
        update_call = {
            "jsonrpc": "2.0",
            "id": 3,
            "method": "tools/call",
            "params": {
                "name": "update_card",
                "arguments": {
                    "card_id": created_card.id,
                    "description": "Updated through MCP",
                    "labels": ["mcp", "test"],
                },
            },
        }

        response = client.post(
            "/mcp",
            json=update_call,
            headers=mcp_headers,
        )
        output = response.json()
        assert not output["result"].get("isError"), output
        assert created_card.description == "Updated through MCP"
        assert created_card.labels == ["mcp", "test"]

        # Retrieve activity history through MCP.
        history_call = {
            "jsonrpc": "2.0",
            "id": 4,
            "method": "tools/call",
            "params": {
                "name": "activity_history",
                "arguments": {"board_id": board_id, "limit": 20},
            },
        }

        response = client.post(
            "/mcp",
            json=history_call,
            headers=mcp_headers,
        )
        output = response.json()
        assert not output["result"].get("isError"), output

        # Verify the tool call did not return an MCP error.
        assert "result" in output


def test_update_card_details_and_log_changes(svc):
    service, board_id = svc
    card = service.add_card(MEMBER, board_id, "Prepare report")

    updated = service.update_card(
        MEMBER,
        card["id"],
        title="Prepare final report",
        description="Review and submit",
        assignee="ANA@X.IO",
        due_date="2026-11-15",
        labels=["urgent", "research"],
    )

    assert updated["title"] == "Prepare final report"
    assert updated["description"] == "Review and submit"
    assert updated["assignee"] == "ana@x.io"
    assert updated["due_date"] == "2026-11-15"
    assert updated["labels"] == ["urgent", "research"]

    update_events = [
        event
        for event in service.repo.events
        if event[1] == card["id"]
        and "updated card fields:" in event[3]
    ]

    assert update_events
    assert update_events[-1][2] == MEMBER.email

    for field_name in (
        "title",
        "description",
        "assignee",
        "due_date",
        "labels",
    ):
        assert field_name in update_events[-1][3]


def test_update_card_can_clear_assignee_and_due_date(svc):
    service, board_id = svc
    card = service.add_card(MEMBER, board_id, "Prepare report")

    service.update_card(
        MEMBER,
        card["id"],
        assignee="ana@x.io",
        due_date="2026-11-15",
    )

    updated = service.update_card(
        MEMBER,
        card["id"],
        assignee="",
        due_date="",
    )

    assert updated["assignee"] is None
    assert updated["due_date"] is None


def test_viewer_cannot_update_card(svc):
    service, board_id = svc
    card = service.add_card(MEMBER, board_id, "Restricted edit")

    with pytest.raises(Forbidden):
        service.update_card(
            VIEWER,
            card["id"],
            title="Unauthorized change",
        )

    assert service.repo.cards[card["id"]].title == "Restricted edit"


def test_update_card_rejects_invalid_due_date(svc):
    service, board_id = svc
    card = service.add_card(OWNER, board_id, "Date validation")

    with pytest.raises(Invalid, match="YYYY-MM-DD"):
        service.update_card(
            OWNER,
            card["id"],
            due_date="15/11/2026",
        )


def test_update_card_rejects_blank_title(svc):
    service, board_id = svc
    card = service.add_card(OWNER, board_id, "Keep valid title")

    with pytest.raises(Invalid, match="title"):
        service.update_card(
            OWNER,
            card["id"],
            title="   ",
        )

    assert service.repo.cards[card["id"]].title == "Keep valid title"


def test_activity_history_returns_recent_events(svc):
    service, board_id = svc

    service.add_card(MEMBER, board_id, "First task")
    service.add_card(OWNER, board_id, "Second task")

    result = service.activity_history(OWNER, board_id)

    assert result["board_id"] == board_id
    assert result["board_title"] == "Launch"
    assert result["activity"]
    assert result["activity"][0]["actor"] == OWNER.email
    assert result["activity"][0]["action"] == "added card 'Second task'"


def test_stranger_cannot_read_activity_history(svc):
    service, board_id = svc
    service.add_card(OWNER, board_id, "Private task")

    with pytest.raises(NotFound):
        service.activity_history(STRANGER, board_id)


def test_activity_history_validates_limit(svc):
    service, board_id = svc

    with pytest.raises(Invalid, match="between 1 and 500"):
        service.activity_history(OWNER, board_id, 0)

    with pytest.raises(Invalid, match="between 1 and 500"):
        service.activity_history(OWNER, board_id, 501)


def test_web_activity_endpoint_requires_authentication_and_board_access(svc):
    service, board_id = svc
    service.add_card(OWNER, board_id, "History test")

    with TestClient(create_app(service)) as client:
        assert client.get(
            f"/api/boards/{board_id}/activity"
        ).status_code == 401

        response = client.get(
            f"/api/boards/{board_id}/activity",
            headers=_h(OWNER),
        )
        assert response.status_code == 200
        assert response.json()["board_id"] == board_id
        assert response.json()["activity"]

        response = client.get(
            f"/api/boards/{board_id}/activity",
            headers=_h(STRANGER),
        )
        assert response.status_code == 404