sawera/sawera-boards

Sawera Boards - a team task board for the Dodil Boards Challenge. Boards, lists, cards, roles, activity log, MCP actions and AI extras. FastAPI + DataK3 + appid + Models + MCP.

A remix template · 0 remixes · Remix sawera/sawera-boards

tests/test_boards.py

"""The board's rules, through the use cases and both doors. No Dodil account needed:
the repository and the AI are in-memory fakes."""
import json
from datetime import date, timedelta

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(e) for e in
                                   ("ana@x.io", "bo@x.io", "cy@x.io", "dee@x.io"))
TODAY = date.today()


class MemoryBoards:
    """In-memory fake of the repository port. Boards are stored whole (with their
    lists and members), like one row-set per board."""

    def __init__(self):
        self.boards, self.cards, self.events = {}, {}, []
        self.embeddings = {}  # card_id -> (text_hash, vector)

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

    def boards_for(self, email):
        return [b for b in self.boards.values() if b.owner_email == email or email in b.members]

    def save_board(self, board):
        self.boards[board.id] = board

    def delete_board(self, board_id):
        self.boards.pop(board_id, None)
        gone = {cid for cid, c in self.cards.items() if c.board_id == board_id}
        self.cards = {k: v for k, v in self.cards.items() if k not in gone}
        self.comments = {k: v for k, v in getattr(self, "comments", {}).items()
                         if v.card_id not in gone}
        self.embeddings = {k: v for k, v in self.embeddings.items() if k not in gone}
        self.events = [e for e in self.events if e["board_id"] != board_id]

    def save_lists(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 sorted((c for c in self.cards.values() if c.board_id == board_id),
                      key=lambda c: c.position)

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

    def delete_card(self, card_id):
        self.cards.pop(card_id, None)
        self.comments = {k: v for k, v in getattr(self, "comments", {}).items()
                         if v.card_id != card_id}

    def save_comment(self, comment):
        if not hasattr(self, "comments"):
            self.comments = {}
        self.comments[comment.id] = comment

    def comments_on(self, card_id):
        return [c for c in getattr(self, "comments", {}).values()
                if c.card_id == card_id]

    def comment_counts(self, board_id):
        out = {}
        for c in getattr(self, "comments", {}).values():
            card = self.cards.get(c.card_id)
            if card and card.board_id == board_id:
                out[c.card_id] = out.get(c.card_id, 0) + 1
        return out

    def log(self, board_id, card_id, actor, action):
        self.events.append({"board_id": board_id, "card_id": card_id, "actor": actor,
                            "action": action, "at": "2026-01-01T00:00:00"})

    def activity(self, board_id, limit=50):
        return [e for e in reversed(self.events) if e["board_id"] == board_id][:limit]

    def counts(self):
        return {"boards": len(self.boards), "cards": len(self.cards),
                "activity": len(self.events)}

    def get_embeddings(self, board_id):
        return {cid: v for cid, v in self.embeddings.items()
                if cid in self.cards and self.cards[cid].board_id == board_id}

    def save_embedding(self, card_id, text_hash, vector):
        self.embeddings[card_id] = (text_hash, tuple(vector))


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

    def summarize(self, facts):
        return "All calm on this board."

    _SYNONYMS = {"invoice": "billing", "bill": "billing", "billing": "billing",
                 "money": "billing", "owed": "billing", "payment": "billing",
                 "paid": "billing"}

    def embed(self, texts):
        """A stand-in for real meaning vectors: words are folded to a canonical
        form (invoice = money = billing), then counted into fixed buckets, so
        texts about the same thing land close together and others do not."""
        import hashlib
        out = []
        for text in texts:
            vec = [0.0] * 128
            for word in "".join(ch if ch.isalnum() else " " for ch in text.lower()).split():
                word = self._SYNONYMS.get(word, word)
                vec[int(hashlib.md5(word.encode()).hexdigest(), 16) % 128] += 1.0
            out.append(tuple(vec))
        return out


@pytest.fixture
def svc():
    s = Boards(MemoryBoards(), FakeAI())
    b = s.create_board(OWNER, "Launch")
    s.add_member(OWNER, b["id"], MEMBER.email, "member")
    s.add_member(OWNER, b["id"], VIEWER.email, "viewer")
    return s, b["id"]


def test_member_adds_and_moves_cards_and_it_is_logged(svc):
    s, bid = svc
    card = s.add_card(MEMBER, bid, "Send invoice")
    moved = s.move_card(MEMBER, card["id"], list_name="Done")
    assert moved["list_name"] == "Done"
    assert any("moved" in e["action"] and "Done" in e["action"] for e in s.repo.events)


def test_viewer_cannot_edit(svc):
    s, bid = svc
    with pytest.raises(Forbidden):
        s.add_card(VIEWER, bid, "nope")
    card = s.add_card(OWNER, bid, "real")
    with pytest.raises(Forbidden):
        s.move_card(VIEWER, card["id"], list_name="Done")
    with pytest.raises(Forbidden):
        s.add_list(VIEWER, bid, "Nope")


def test_only_owner_manages_members(svc):
    s, bid = svc
    with pytest.raises(Forbidden):
        s.add_member(MEMBER, bid, "eve@x.io", "member")
    with pytest.raises(Forbidden):
        s.remove_member(MEMBER, bid, VIEWER.email)
    s.remove_member(OWNER, bid, VIEWER.email)
    with pytest.raises(NotFound):
        s.show_board(VIEWER, bid)
    with pytest.raises(Invalid):
        s.remove_member(OWNER, bid, OWNER.email)


def test_stranger_cannot_see_board(svc):
    s, bid = svc
    with pytest.raises(NotFound):
        s.show_board(STRANGER, bid)
    assert s.my_boards(STRANGER) == []
    card = s.add_card(OWNER, bid, "secret plans")
    assert s.search_cards(STRANGER, "secret") == []
    assert s.overdue_cards(STRANGER) == []


def test_board_starts_with_three_lists_and_lists_have_rules(svc):
    s, bid = svc
    board = s.show_board(OWNER, bid)
    assert [l["name"] for l in board["lists"]] == ["To do", "Doing", "Done"]
    review = s.add_list(MEMBER, bid, "Review")
    assert s.show_board(OWNER, bid)["lists"][-1]["name"] == "Review"
    with pytest.raises(Invalid):
        s.add_list(MEMBER, bid, "review")            # duplicate, any case
    with pytest.raises(Invalid):
        s.rename_list(MEMBER, bid, review["id"], "Done")
    card = s.add_card(MEMBER, bid, "in review", list_name="Review")
    with pytest.raises(Invalid):
        s.remove_list(MEMBER, bid, review["id"])     # not empty
    s.delete_card(MEMBER, card["id"])
    s.remove_list(MEMBER, bid, review["id"])
    assert [l["name"] for l in s.show_board(OWNER, bid)["lists"]] == ["To do", "Doing", "Done"]


def test_card_fields_roundtrip_and_labels_are_cleaned(svc):
    s, bid = svc
    card = s.add_card(OWNER, bid, "  Write docs ", description="long text",
                      assignee=" Bo@X.io ", due_date=TODAY + timedelta(days=3),
                      labels=[" Docs ", "docs", "API"])
    assert card["title"] == "Write docs"
    assert card["assignee"] == "bo@x.io"
    assert card["labels"] == ["docs", "api"]
    assert card["due_date"] == (TODAY + timedelta(days=3)).isoformat()
    upd = s.update_card(MEMBER, card["id"], labels=["a", "b", "c", "d", "e", "f"])
    assert upd["labels"] == ["a", "b", "c", "d", "e", "f"]


def test_too_many_labels_refused(svc):
    s, bid = svc
    with pytest.raises(Invalid):
        s.add_card(OWNER, bid, "x", labels=["1", "2", "3", "4", "5", "6", "7"])


def test_my_plate_orders_overdue_first(svc):
    s, bid = svc
    late = s.add_card(OWNER, bid, "late", assignee=MEMBER.email,
                      due_date=TODAY - timedelta(days=2))
    soon = s.add_card(OWNER, bid, "soon", assignee=MEMBER.email,
                      due_date=TODAY + timedelta(days=2))
    s.add_card(OWNER, bid, "not mine", assignee=OWNER.email,
               due_date=TODAY - timedelta(days=9))
    done = s.add_card(OWNER, bid, "finished", assignee=MEMBER.email,
                      due_date=TODAY - timedelta(days=5))
    s.move_card(OWNER, done["id"], list_name="Done")
    plate = s.my_cards(MEMBER)
    assert [c["id"] for c in plate] == [late["id"], soon["id"]]
    assert plate[0]["overdue"] is True


def test_overdue_ignores_done_cards(svc):
    s, bid = svc
    c1 = s.add_card(OWNER, bid, "open late", due_date=TODAY - timedelta(days=1))
    c2 = s.add_card(OWNER, bid, "done late", due_date=TODAY - timedelta(days=1))
    s.move_card(OWNER, c2["id"], list_name="Done")
    assert [c["id"] for c in s.overdue_cards(VIEWER)] == [c1["id"]]


def test_blocked_by_chain_cycle_and_delete(svc):
    s, bid = svc
    a = s.add_card(OWNER, bid, "Launch")
    b = s.add_card(OWNER, bid, "Legal sign-off")
    c = s.add_card(OWNER, bid, "Contract")
    s.set_blockers(OWNER, a["id"], [b["id"]])
    s.set_blockers(OWNER, b["id"], [c["id"]])
    view = s.what_is_blocking(MEMBER, a["id"])
    assert [x["title"] for x in view["blocking"]] == ["Legal sign-off", "Contract"]
    assert s.show_board(OWNER, bid)["cards"][0]["blocked"] is True
    with pytest.raises(Invalid):
        s.set_blockers(OWNER, c["id"], [a["id"]])      # would circle a->b->c->a
    with pytest.raises(Invalid):
        s.set_blockers(OWNER, a["id"], [a["id"]])      # self
    s.move_card(OWNER, c["id"], list_name="Done")
    view = s.what_is_blocking(MEMBER, a["id"])
    assert [x["title"] for x in view["blocking"]] == ["Legal sign-off"]
    s.delete_card(OWNER, b["id"])                       # deleting a blocker frees the card
    assert s.what_is_blocking(MEMBER, a["id"])["blocking"] == []


def test_activity_records_who_did_what(svc):
    s, bid = svc
    card = s.add_card(MEMBER, bid, "Invoice")
    s.assign_card(OWNER, card["id"], MEMBER.email)
    acts = s.board_activity(VIEWER, bid)
    assert any("added card" in e["action"] and e["actor"] == MEMBER.email for e in acts)
    assert any("updated assignee" in e["action"] and e["actor"] == OWNER.email for e in acts)


def test_daily_summary_uses_the_assistant(svc):
    s, bid = svc
    s.add_card(OWNER, bid, "anything")
    assert s.daily_summary(OWNER, bid)["summary"] == "All calm on this board."
    assert s.suggest_checklist(OWNER, s.show_board(OWNER, bid)["cards"][0]["id"])[
        "suggested_checklist"] == ["step one", "step two"]


def _h(p):
    return {"X-Dodil-User": json.dumps({"sub": p.email, "email": p.email})}


def test_web_and_mcp_doors_share_the_rules(svc):
    s, bid = svc
    with TestClient(create_app(s)) as c:
        assert c.get("/api/boards").status_code == 401
        assert c.post("/").status_code == 401
        assert c.post("/", headers=_h(OWNER)).json()["counts"]["boards"] == 1
        assert c.get("/api/boards", headers=_h(VIEWER)).json()[0]["role"] == "viewer"
        assert c.post(f"/api/boards/{bid}/cards", json={"title": "t"},
                      headers=_h(VIEWER)).status_code == 403
        page = c.get("/", headers=_h(OWNER))
        assert page.status_code == 200 and "Sawera Boards" in page.text
        r = c.post("/mcp", json={"jsonrpc": "2.0", "id": 1, "method": "tools/list"})
        assert r.status_code == 401 and "resource_metadata" in r.headers["www-authenticate"]
        mcp_h = {**_h(MEMBER), "accept": "application/json, text/event-stream"}
        init = {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {
            "protocolVersion": "2025-06-18", "capabilities": {},
            "clientInfo": {"name": "t", "version": "0"}}}
        assert c.post("/mcp", json=init, headers=mcp_h).status_code == 200

        def tool(name, args):
            out = c.post("/mcp", headers=mcp_h, json={
                "jsonrpc": "2.0", "id": 3, "method": "tools/call",
                "params": {"name": name, "arguments": args}}).json()
            assert not out["result"].get("isError"), out
            # this MCP server returns one JSON text item per element of a list result
            return [json.loads(i["text"]) for i in out["result"]["content"]
                    if i["type"] == "text"]

        card = tool("add_card", {"board_id": bid, "title": "via mcp",
                                 "due_date": (TODAY - timedelta(days=1)).isoformat()})[0]
        assert card["overdue"] is True
        tool("assign_card", {"card_id": card["id"], "email": MEMBER.email})
        plate = tool("my_cards", {})
        assert [p["id"] for p in plate] == [card["id"]]
        assert [p["id"] for p in tool("overdue_cards", {})] == [card["id"]]


def test_semantic_search_finds_cards_by_meaning(svc):
    s, bid = svc
    invoice = s.add_card(OWNER, bid, "Send the invoice")
    s.add_card(OWNER, bid, "Book the venue")
    s.add_card(OWNER, bid, "First task")  # two words: too little text for meaning
    hits = s.search_cards(OWNER, "money owed to us")
    assert [h["id"] for h in hits] == [invoice["id"]]  # meaning match, no shared words
    assert hits[0]["score"] >= 0.4
    # the vector was cached, and a keyword search still ranks first with no score
    assert s.repo.embeddings
    kw = s.search_cards(OWNER, "invoice")
    assert kw[0]["id"] == invoice["id"] and "score" not in kw[0]


def test_search_without_ai_is_keyword_only(svc):
    s, bid = svc
    bare = Boards(s.repo, None)
    card = s.add_card(OWNER, bid, "Send the invoice")
    assert [h["id"] for h in bare.search_cards(OWNER, "invoice")] == [card["id"]]
    assert bare.search_cards(OWNER, "money owed") == []


def test_search_survives_a_failing_ai(svc):
    class BrokenAI(FakeAI):
        def embed(self, texts):
            raise RuntimeError("models are down")

    s, bid = svc
    broken = Boards(s.repo, BrokenAI())
    card = s.add_card(OWNER, bid, "Send the invoice")
    assert [h["id"] for h in broken.search_cards(OWNER, "invoice")] == [card["id"]]


def test_comments_live_on_the_card_and_follow_the_roles(svc):
    s, bid = svc
    card = s.add_card(OWNER, bid, "Design the invite")
    first = s.add_comment(MEMBER, card["id"], "Draft is in the shared folder.")
    s.add_comment(OWNER, card["id"], "Thanks - two small fixes from me.")
    got = s.card_comments(VIEWER, card["id"])  # viewers may read the discussion
    assert [c["body"] for c in got] == ["Draft is in the shared folder.",
                                        "Thanks - two small fixes from me."]
    assert got[0]["author"] == MEMBER.email and got[0]["id"] == first["id"]
    with pytest.raises(Forbidden):
        s.add_comment(VIEWER, card["id"], "I should not be able to write")
    with pytest.raises(Invalid):
        s.add_comment(MEMBER, card["id"], "   ")
    with pytest.raises(NotFound):
        s.card_comments(STRANGER, card["id"])
    acts = [e["action"] for e in s.repo.events if "commented" in e["action"]]
    assert acts == [f"commented on {'Design the invite'!r}"] * 2
    s.delete_card(OWNER, card["id"])
    assert s.repo.comments_on(card["id"]) == []


def test_only_the_owner_can_delete_the_board_and_it_takes_everything(svc):
    s, bid = svc
    card = s.add_card(MEMBER, bid, "Write the launch notes")
    s.add_comment(MEMBER, card["id"], "Draft started.")
    with pytest.raises(Forbidden):
        s.delete_board(MEMBER, bid)   # a member cannot delete the board
    with pytest.raises(Forbidden):
        s.delete_board(VIEWER, bid)   # nor a viewer
    with pytest.raises(NotFound):
        s.delete_board(STRANGER, bid)  # a stranger can't even see it
    assert s.delete_board(OWNER, bid) == {"deleted": bid}
    assert s.repo.get(bid) is None
    assert s.repo.cards_on(bid) == []
    assert s.repo.comments_on(card["id"]) == []
    assert s.my_boards(MEMBER) == []  # gone for the member too


def test_cards_carry_a_meaning_vector_from_the_moment_they_are_saved(svc):
    s, bid = svc
    card = s.add_card(OWNER, bid, "Send the invoice to the client")
    assert card["id"] in s.repo.embeddings  # written at save time, not at search time
    s.update_card(OWNER, card["id"], title="Send the final bill to the client")
    found = s.search_cards(OWNER, "money owed to us")
    assert found and found[0]["id"] == card["id"]


def test_search_keyword_half_answers_without_the_ai(svc):
    s, bid = svc
    s.add_card(OWNER, bid, "Send the invoice to the client")
    fast = s.search_cards(OWNER, "invoice", semantic=False)
    assert [c["title"] for c in fast] == ["Send the invoice to the client"]
    assert all("score" not in c for c in fast)  # no meaning ranking in the fast half
    assert s.search_cards(OWNER, "money owed", semantic=False) == []