damien-kaiser/taskloft

Taskloft — team task boards (owner / member / viewer), card history, my plate, overdue, blockers, AI checklists and summaries, usable from Claude and ChatGPT over MCP.

A remix template · 0 remixes · Remix damien-kaiser/taskloft

tests/test_boards.py

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

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(2026, 10, 29)


class MemoryBoards:
    def __init__(self):
        self.boards, self.cards, self.events, self.numbers = {}, {}, [], {}
        self.blockers, self.items = set(), {}

    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 remove_member(self, board_id, email):
        self.boards[board_id].members.pop(email, None)

    def claim_number(self, card_id):
        n = max(self.numbers, default=0) + 1
        assert n not in self.numbers          # a claim never overwrites
        self.numbers[n] = card_id
        return n

    def _with_blockers(self, c):
        c.blocked_by = [b for (a, b) in sorted(self.blockers) if a == c.id]
        return c

    def get_card(self, card_id):
        c = self.cards.get(card_id)
        return self._with_blockers(c) if c else None

    def card_by_number(self, number):
        return next((self._with_blockers(c) for c in self.cards.values() if c.number == number), None)

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

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

    def set_blocker(self, card_id, blocked_by, on):
        (self.blockers.add if on else self.blockers.discard)((card_id, blocked_by))

    def checklist(self, card_id):
        return sorted((dict(i) for i in self.items.values() if i["card_id"] == card_id), key=lambda i: i["position"])

    def save_checklist_item(self, item):
        self.items[item["id"]] = dict(item)

    def log(self, board_id, card_id, actor, action, changes=None):
        self.events.append({"board_id": board_id, "card_id": card_id, "actor": actor, "action": action,
                            "changes": changes, "at": f"2026-10-29T10:{len(self.events):02d}:00"})

    def history(self, board_id, card_id=None, limit=50):
        ev = [e for e in self.events if e["board_id"] == board_id and (card_id is None or e["card_id"] == card_id)]
        return list(reversed(ev))[:limit]

    vectors = None
    summaries = None

    def cached_summary(self, key):
        return (self.summaries or {}).get(key)

    def save_summary(self, key, board_id, text):
        self.summaries = self.summaries or {}
        self.summaries[key] = text

    def vector_hashes(self, board_ids):
        self.vectors = self.vectors or {}
        return {cid: h for cid, (b, h, _) in self.vectors.items() if b in board_ids}

    def save_vector(self, card_id, board_id, text_hash, vector):
        self.vectors = self.vectors or {}
        self.embedded = getattr(self, "embedded", 0) + 1
        self.vectors[card_id] = (board_id, text_hash, vector)

    def nearest(self, board_ids, vector, k, exclude=None):
        import math
        def dist(a, b):
            na, nb = math.sqrt(sum(x * x for x in a)), math.sqrt(sum(x * x for x in b))
            return 1.0 - (sum(x * y for x, y in zip(a, b)) / (na * nb) if na and nb else 0.0)
        hits = [(cid, dist(v, vector)) for cid, (b, _, v) in (self.vectors or {}).items()
                if b in board_ids and cid != exclude]
        return sorted(hits, key=lambda x: x[1])[:k]


class FakeAI:
    """Deterministic 'meaning': a bag of word stems, so 'bill' lands near 'invoice' via a tiny synonym map."""
    SYN = {"bill": "invoice", "billing": "invoice", "invoices": "invoice", "client": "customer"}

    def embed(self, texts):
        vocab = ["invoice", "customer", "design", "launch", "copy", "bank", "pay"]
        out = []
        for t in texts:
            words = [self.SYN.get(w, w) for w in t.lower().replace("·", " ").split()]
            v = [float(sum(1 for w in words if w.startswith(k))) for k in vocab]
            out.append(v if any(v) else [0.0] * (len(vocab) - 1) + [0.01])
        return out

    def checklist(self, title, description=""):
        return ["Draft it", "Check it", "Send it"]

    def summary(self, board_title, facts):
        return f"{board_title}: " + ("overdue work" if "OVERDUE" in facts else "on track")


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


# --- roles hold ---------------------------------------------------------------------------------
def test_viewer_reads_but_cannot_change_anything(svc):
    s, bid = svc
    card = s.add_card(MEMBER, bid, "Send invoice")
    assert s.show_board(VIEWER, bid)["role"] == "viewer"
    for attempt in (lambda: s.add_card(VIEWER, bid, "nope"),
                    lambda: s.move_card(VIEWER, card["number"], "Done"),
                    lambda: s.update_card(VIEWER, card["number"], title="x"),
                    lambda: s.assign_card(VIEWER, card["number"], MEMBER.email),
                    lambda: s.add_checklist_items(VIEWER, card["number"], ["a"]),
                    lambda: s.add_list(VIEWER, bid, "Later")):
        with pytest.raises(Forbidden):
            attempt()


def test_only_the_owner_manages_people(svc):
    s, bid = svc
    with pytest.raises(Forbidden):
        s.invite_member(MEMBER, bid, "eve@x.io", "member")
    with pytest.raises(Forbidden):
        s.remove_member(MEMBER, bid, VIEWER.email)
    with pytest.raises(Invalid):
        s.invite_member(OWNER, bid, "eve@x.io", "owner")
    s.invite_member(OWNER, bid, VIEWER.email, "member")          # role change
    assert {m["email"]: m["role"] for m in s.members(OWNER, bid)["members"]}[VIEWER.email] == "member"


def test_nobody_sees_a_board_they_are_not_on(svc):
    s, bid = svc
    card = s.add_card(OWNER, bid, "Secret plan", due_date="2026-10-01")
    with pytest.raises(NotFound):
        s.show_board(STRANGER, bid)
    with pytest.raises(NotFound):
        s.card(STRANGER, card["number"])
    with pytest.raises(NotFound):
        s.show_board(STRANGER, "Ops")                          # not by title either
    assert s.my_boards(STRANGER) == []
    assert s.search_cards(STRANGER, "secret") == []
    assert s.overdue(STRANGER) == []


def test_removed_member_loses_the_board_and_their_cards(svc):
    s, bid = svc
    card = s.add_card(OWNER, bid, "Ship", assignee=MEMBER.email)
    s.remove_member(OWNER, bid, MEMBER.email)
    with pytest.raises(NotFound):
        s.show_board(MEMBER, bid)
    assert s.card(OWNER, card["number"])["assignee"] is None


# --- boards, lists, cards -----------------------------------------------------------------------
def test_cards_have_numbers_and_move_by_names(svc):
    s, bid = svc
    a = s.add_card(MEMBER, bid, "Send invoice", list_ref="Doing")
    b = s.add_card(MEMBER, "ops board", "Call the bank")         # board by title
    assert (a["number"], b["number"]) == (1, 2) and a["list"] == "Doing"
    moved = s.move_card(MEMBER, "#2", "done")                    # card by number, list by name
    assert moved["list"] == "Done" and moved["done"]
    with pytest.raises(Invalid):
        s.move_card(MEMBER, 2, "Someday")


def test_drag_and_drop_order_inside_a_list(svc):
    s, bid = svc
    one, two, three = (s.add_card(OWNER, bid, t)["number"] for t in ("one", "two", "three"))
    s.move_card(OWNER, three, "To do", before_ref=one)           # dropped above "one"
    order = [c["title"] for c in s.show_board(OWNER, bid)["lists"][0]["cards"]]
    assert order == ["three", "one", "two"]


def test_every_card_change_is_recorded_field_by_field(svc):
    s, bid = svc
    card = s.add_card(MEMBER, bid, "Send invoice")
    s.update_card(MEMBER, card["number"], title="Send invoice", description="To ACME", labels=["Finance"])
    s.assign_card(OWNER, card["number"], MEMBER.email)
    s.move_card(MEMBER, card["number"], "Doing")
    hist = s.card_history(OWNER, card["number"])
    assert [h["actor"] for h in hist] == [MEMBER.email, OWNER.email, MEMBER.email, MEMBER.email]
    edit = next(h for h in hist if h["action"].startswith("edited"))
    assert edit["changes"] == {"description": ["", "To ACME"], "labels": [[], ["finance"]]}   # title unchanged: not logged
    assert hist[0]["changes"] == {"list": ["To do", "Doing"]}


def test_assignee_must_be_able_to_work_on_the_board(svc):
    s, bid = svc
    card = s.add_card(OWNER, bid, "x")
    with pytest.raises(Invalid):
        s.assign_card(OWNER, card["number"], VIEWER.email)
    with pytest.raises(Invalid):
        s.assign_card(OWNER, card["number"], STRANGER.email)
    assert s.assign_card(MEMBER, card["number"], "me")["assignee"] == MEMBER.email


# --- what needs me ------------------------------------------------------------------------------
def test_my_plate_this_week_and_overdue(svc):
    s, bid = svc
    s.add_card(OWNER, bid, "Late", assignee=MEMBER.email, due_date="2026-10-20")
    s.add_card(OWNER, bid, "Friday", assignee=MEMBER.email, due_date="2026-10-30")
    s.add_card(OWNER, bid, "Next month", assignee=MEMBER.email, due_date="2026-11-30")
    s.add_card(OWNER, bid, "Whenever", assignee=MEMBER.email)
    done = s.add_card(OWNER, bid, "Finished late", assignee=MEMBER.email, due_date="2026-10-01")
    s.move_card(OWNER, done["number"], "Done")
    s.add_card(OWNER, bid, "Not mine", assignee=OWNER.email, due_date="2026-10-21")
    plate = s.my_plate(MEMBER)
    assert [c["title"] for c in plate["overdue"]] == ["Late"]
    assert [c["title"] for c in plate["due_soon"]] == ["Friday"]
    assert [c["title"] for c in plate["no_due_date"]] == ["Whenever"]
    assert [c["title"] for c in s.overdue(VIEWER)] == ["Late", "Not mine"]   # done cards are never overdue


# --- blockers -----------------------------------------------------------------------------------
def test_blockers_refuse_loops_and_report_the_chain(svc):
    s, bid = svc
    launch, site, copy = (s.add_card(OWNER, bid, t)["number"] for t in ("Launch", "Website", "Copy"))
    s.set_blocked_by(OWNER, launch, site)
    s.set_blocked_by(OWNER, site, copy)
    with pytest.raises(Invalid):
        s.set_blocked_by(OWNER, copy, launch)                    # would be a loop
    with pytest.raises(Invalid):
        s.set_blocked_by(OWNER, launch, launch)
    r = s.what_is_blocking(VIEWER, launch)
    assert [(c["title"], c["depth"]) for c in r["blocking"]] == [("Website", 1), ("Copy", 2)]
    assert s.card(OWNER, site)["waiting_on"] == [copy]
    s.move_card(OWNER, copy, "Done")
    assert [c["title"] for c in s.what_is_blocking(OWNER, launch)["blocking"]] == ["Website"]
    lists = {lst["name"]: lst["cards"] for lst in s.show_board(OWNER, bid)["lists"]}
    assert {c["number"]: c["waiting_on"] for c in lists["To do"]} == {launch: [site], site: []}   # a finished blocker stops blocking
    other = s.create_board(OWNER, "Other")
    far = s.add_card(OWNER, other["id"], "Elsewhere")["number"]
    with pytest.raises(Invalid):
        s.set_blocked_by(OWNER, launch, far)


# --- checklist and AI ---------------------------------------------------------------------------
def test_turn_a_card_into_a_checklist_and_tick_it(svc):
    s, bid = svc
    card = s.add_card(MEMBER, bid, "Send invoice")["number"]
    out = s.make_checklist(MEMBER, card)
    assert [i["text"] for i in out["checklist"]] == ["Draft it", "Check it", "Send it"]
    out = s.tick_checklist_item(MEMBER, card, "2")
    assert [i["done"] for i in out["checklist"]] == [False, True, False]
    first = s.board_summary(VIEWER, bid)
    assert "overdue work" not in first["summary"] and first["cached"] is False
    assert s.board_summary(MEMBER, bid)["cached"] is True              # same board, same facts: no second wait
    s.add_card(MEMBER, bid, "Late", due_date="2026-10-01")
    again = s.board_summary(VIEWER, bid)
    assert again["cached"] is False and "overdue work" in again["summary"]   # the board changed: written again


# --- meaning ----------------------------------------------------------------------------------
def test_find_cards_like_this_one_only_among_visible_boards(svc):
    s, bid = svc
    inv = s.add_card(OWNER, bid, "Send invoice to ACME")["number"]
    s.add_card(OWNER, bid, "Bill the client for October")
    s.add_card(OWNER, bid, "Design the plan cards")
    other = s.create_board(STRANGER, "Private")
    s.add_card(STRANGER, other["id"], "Invoice the customer twice")      # close in meaning, but not visible
    sim = s.similar_cards(VIEWER, inv)
    assert sim[0]["title"] == "Bill the client for October" and inv not in [c["number"] for c in sim]
    assert all(c["board_id"] == bid for c in sim)
    assert s.search_by_meaning(MEMBER, "billing")[0]["title"] in ("Send invoice to ACME", "Bill the client for October")
    n = s.repo.embedded
    s.similar_cards(VIEWER, inv)
    assert s.repo.embedded == n                                         # unchanged cards are not re-embedded
    s.update_card(OWNER, inv, title="Send invoice to ACME and Globex")
    s.similar_cards(VIEWER, inv)
    assert s.repo.embedded == n + 1                                     # an edited card is


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


def _mcp(c, who, name, args, i=[10]):
    i[0] += 1
    h = {**_h(who), "accept": "application/json, text/event-stream"}
    return c.post("/mcp", json={"jsonrpc": "2.0", "id": i[0], "method": "tools/call",
                                "params": {"name": name, "arguments": args}}, headers=h).json()


def _data(out):
    """A tool's JSON answer (a list answer arrives as one content item per element)."""
    items = [json.loads(x["text"]) for x in out["result"]["content"]]
    return items[0] if len(items) == 1 and isinstance(items[0], dict) else items


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.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
        assert c.get(f"/api/boards/{bid}", headers=_h(STRANGER)).status_code == 404
        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
        tools = c.post("/mcp", json={"jsonrpc": "2.0", "id": 2, "method": "tools/list"}, headers=mcp_h).json()
        names = {t["name"] for t in tools["result"]["tools"]}
        assert {"my_plate", "add_card", "move_card", "overdue", "assign_card", "what_is_blocking"} <= names
        # "add a card 'Send invoice' to Doing on the Ops board"
        out = _mcp(c, MEMBER, "add_card", {"board": "Ops board", "title": "Send invoice", "list": "Doing",
                                           "assignee": MEMBER.email, "due_date": "2026-10-30"})
        assert not out["result"].get("isError"), out
        num = _data(out)["number"]
        # "what's on my plate this week?"
        plate = _data(_mcp(c, MEMBER, "my_plate", {}))
        assert [x["number"] for x in plate["due_soon"]] == [num]
        # "move card N to Done", and a viewer can't
        assert _mcp(c, VIEWER, "move_card", {"card": str(num), "list": "Done"})["result"]["isError"]
        assert not _mcp(c, MEMBER, "move_card", {"card": str(num), "list": "Done"})["result"].get("isError")
        # the AI only sees boards the signed-in person can see
        assert _mcp(c, STRANGER, "show_board", {"board": "Ops"})["result"]["isError"]
        assert _data(_mcp(c, STRANGER, "my_boards", {})) == []
        # web PATCH: only sent fields change, due_date null clears it
        r = c.post(f"/api/boards/{bid}/cards", json={"title": "Tagged", "list": "Doing", "labels": ["Finance"]},
                   headers=_h(MEMBER))
        assert r.status_code == 201 and r.json()["labels"] == ["finance"], r.text
        r = c.patch(f"/api/cards/{num}", json={"due_date": None}, headers=_h(OWNER)).json()
        assert r["due_date"] is None and r["title"] == "Send invoice"
        assert c.get("/", headers=_h(OWNER)).status_code == 200