dachdigitalwerksorg/kanvaflow
KanvaFlow — Remixable team task board with RBAC, drag-and-drop Kanban, dependencies, Dodil Models AI, and MCP tools.
A remix template · 0 remixes · Remix dachdigitalwerksorg/kanvaflow
tests/test_boards.py
"""Comprehensive Domain, Use-Case, Web Door, and MCP Door test suite for KanvaFlow.
Runs completely in-memory with deterministic fakes (zero Dodil account or network required).
"""
import copy
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")
)
FIXED_TODAY = date(2026, 11, 1)
class MemoryBoards:
def __init__(self):
self.boards = {}
self.cards = {}
self.deps = {} # board_id -> list[(card_id, blocked_by_id)]
self.events = []
def get(self, board_id):
b = self.boards.get(board_id)
return copy.deepcopy(b) if b else None
def boards_for(self, email):
return [
copy.deepcopy(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] = copy.deepcopy(board)
def remove_member_row(self, board_id, email):
b = self.boards.get(board_id)
if b and email in b.members:
del b.members[email]
def get_card(self, card_id):
c = self.cards.get(card_id)
return copy.deepcopy(c) if c else None
def cards_on(self, board_id):
return [copy.deepcopy(c) for c in self.cards.values() if c.board_id == board_id]
def save_card(self, card):
self.cards[card.id] = copy.deepcopy(card)
def delete_card(self, card_id):
c = self.cards.pop(card_id, None)
if c:
self.deps[c.board_id] = [
pair
for pair in self.deps.get(c.board_id, [])
if pair[0] != card_id and pair[1] != card_id
]
def dependencies_on(self, board_id):
out = {}
for cid, bid in self.deps.get(board_id, []):
out.setdefault(cid, []).append(bid)
return out
def add_dependency(self, board_id, card_id, blocked_by_id):
pairs = self.deps.setdefault(board_id, [])
if (card_id, blocked_by_id) not in pairs:
pairs.append((card_id, blocked_by_id))
def remove_dependency(self, card_id, blocked_by_id):
for bid, pairs in self.deps.items():
self.deps[bid] = [p for p in pairs if p != (card_id, blocked_by_id)]
def log(self, board_id, card_id, actor, action):
self.events.append({
"id": str(len(self.events) + 1),
"board_id": board_id,
"card_id": card_id,
"actor": actor,
"action": action,
"event_at": "2026-11-01T12:00:00Z",
})
def activity_for_board(self, board_id, limit=40):
return [e for e in reversed(self.events) if e["board_id"] == board_id][:limit]
def activity_for_card(self, card_id, limit=40):
return [e for e in reversed(self.events) if e["card_id"] == card_id][:limit]
def health_counts(self):
return {"boards": len(self.boards), "cards": len(self.cards), "activity": len(self.events)}
class FakeAI:
def checklist(self, title, description=""):
return [f"Draft {title}", f"Verify {title}", "Publish"]
def summarize_board(self, board_title, snapshot):
return f"Daily summary for {board_title}: focus on overdue and blocked items."
def embed(self, text):
t = text.lower()
# Deterministic 3D semantic vector: [billing/finance, auth/security, deploy/ops]
return [
1.0 if any(w in t for w in ("invoice", "billing", "payment", "tax")) else 0.05,
1.0 if any(w in t for w in ("oauth", "login", "token", "security")) else 0.05,
1.0 if any(w in t for w in ("deploy", "release", "launch", "ci")) else 0.05,
]
@pytest.fixture
def svc():
s = Boards(MemoryBoards(), FakeAI(), clock=lambda: FIXED_TODAY)
b = s.create_board(OWNER, "Ops", description="Operations board")
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_updates_and_moves_cards_with_full_audit_log(svc):
s, bid = svc
card = s.add_card(
MEMBER,
"Ops", # Board lookup by title works naturally for MCP
"Send invoice",
list_name="To do",
description="October retainer",
assignee=MEMBER.email,
due_date="2026-11-04",
labels=["finance", "Q4"],
)
assert card["labels"] == ["finance", "Q4"]
moved = s.move_card(MEMBER, card["id"], "done") # case-insensitive canonicalization
assert moved["list_name"] == "Done"
history = s.card_history(VIEWER, card["id"])
assert any("moved 'Send invoice' from To do to Done" in h["action"] for h in history)
def test_viewer_is_strictly_read_only(svc):
s, bid = svc
card = s.add_card(OWNER, bid, "Protected task")
with pytest.raises(Forbidden):
s.add_card(VIEWER, bid, "Nope")
with pytest.raises(Forbidden):
s.move_card(VIEWER, card["id"], "Doing")
with pytest.raises(Forbidden):
s.update_card(VIEWER, card["id"], title="Tamper")
with pytest.raises(Forbidden):
s.delete_card(VIEWER, card["id"])
with pytest.raises(Forbidden):
s.add_list(VIEWER, bid, "QA")
with pytest.raises(Forbidden):
s.suggest_checklist(VIEWER, card["id"], apply=True)
# Read-only AI suggestion without mutating the card is permitted for viewers
suggestion = s.suggest_checklist(VIEWER, card["id"], apply=False)
assert len(suggestion["suggested_checklist"]) == 3
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)
def test_stranger_is_completely_isolated_across_all_queries(svc):
s, bid = svc
s.add_card(OWNER, bid, "Secret invoice", due_date="2026-10-20", assignee=OWNER.email)
with pytest.raises(NotFound):
s.show_board(STRANGER, bid)
assert s.my_boards(STRANGER) == []
assert s.list_overdue(STRANGER) == []
assert s.search_similar_cards(STRANGER, "invoice") == []
with pytest.raises(NotFound):
s.what_is_blocking(STRANGER, "Secret")
def test_custom_lists_and_unknown_list_rejection(svc):
s, bid = svc
s.add_list(MEMBER, bid, "QA Review", is_done=False)
c = s.add_card(MEMBER, bid, "Test login flow", list_name="qa review")
assert c["list_name"] == "QA Review"
with pytest.raises(Invalid):
s.remove_list(MEMBER, bid, "QA Review") # non-empty list cannot be removed
s.move_card(MEMBER, c["id"], "Done")
s.remove_list(MEMBER, bid, "QA Review")
with pytest.raises(Invalid):
s.move_card(MEMBER, c["id"], "Someday")
def test_overdue_detection_and_personal_weekly_agenda(svc):
s, bid = svc
overdue_card = s.add_card(OWNER, bid, "Pay VAT", assignee=MEMBER.email, due_date="2026-10-28")
done_past = s.add_card(OWNER, bid, "Old settled bill", list_name="Done", assignee=MEMBER.email, due_date="2026-10-15")
week_card = s.add_card(OWNER, bid, "Prepare demo", assignee=MEMBER.email, due_date="2026-11-05")
assert overdue_card["overdue"] is True
assert done_past["overdue"] is False
overdue_list = s.list_overdue(MEMBER)
assert [c["id"] for c in overdue_list] == [overdue_card["id"]]
agenda = s.my_agenda(MEMBER, days=7)
assert [c["id"] for c in agenda["overdue"]] == [overdue_card["id"]]
assert [c["id"] for c in agenda["due_this_week"]] == [week_card["id"]]
def test_dependencies_cycle_detection_and_what_is_blocking_launch(svc):
s, bid = svc
db_mig = s.add_card(OWNER, bid, "Run DB migration", list_name="To do")
qa_sign = s.add_card(OWNER, bid, "QA sign-off", list_name="To do")
launch = s.add_card(OWNER, bid, "Public Launch", list_name="To do")
s.add_blocker(MEMBER, qa_sign["id"], db_mig["id"])
s.add_blocker(MEMBER, launch["id"], qa_sign["id"])
# Self-dependency and 3-node cycle must be rejected
with pytest.raises(Invalid):
s.add_blocker(MEMBER, launch["id"], launch["id"])
with pytest.raises(Invalid):
s.add_blocker(MEMBER, db_mig["id"], launch["id"])
report = s.what_is_blocking(VIEWER, "launch")
match = report["matches"][0]
assert match["is_blocked"] is True
assert [c["id"] for c in match["direct_blockers"]] == [qa_sign["id"]]
assert [c["id"] for c in match["upstream_blockers"]] == [db_mig["id"]]
# Completing both upstream cards unblocks the launch automatically
s.move_card(MEMBER, db_mig["id"], "Done")
s.move_card(MEMBER, qa_sign["id"], "Done")
report_after = s.what_is_blocking(VIEWER, "launch")
assert report_after["matches"][0]["is_blocked"] is False
def test_ai_semantic_search_checklist_apply_and_board_summary(svc):
s, bid = svc
c1 = s.add_card(OWNER, bid, "Settle supplier payment", description="Wire Q4 invoice")
s.add_card(OWNER, bid, "Rotate OAuth signing keys", description="Security token rotation")
results = s.search_similar_cards(MEMBER, "billing and tax", limit=3)
assert results[0]["id"] == c1["id"]
applied = s.suggest_checklist(MEMBER, c1["id"], apply=True)
assert applied["applied"] is True
assert len(applied["checklist"]) == 3
summary = s.summarize_board(VIEWER, bid)
assert "Daily summary for Ops" in summary["summary"]
def _h(p: Person) -> dict:
return {"X-Dodil-User": json.dumps({"sub": p.email, "email": p.email})}
def test_web_and_mcp_doors_enforce_identical_rules_and_self_check(svc):
s, bid = svc
with TestClient(create_app(s)) as c:
assert c.get("/healthz").status_code == 200
assert c.post("/").json()["ok"] is True
assert c.get("/api/boards").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": "nope"}, headers=_h(VIEWER)).status_code == 403
# Unauthenticated MCP returns 401 with RFC 9728 OAuth resource_metadata discovery header
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": "claude-test", "version": "1.0"},
},
}
assert c.post("/mcp", json=init, headers=mcp_h).status_code == 200
add_call = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "add_card",
"arguments": {
"board_id": "Ops",
"title": "Send invoice",
"list_name": "Doing",
"assignee": MEMBER.email,
"due_date": "2026-10-30",
},
},
}
out = c.post("/mcp", json=add_call, headers=mcp_h).json()
assert not out["result"].get("isError"), out
overdue_call = {
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {"name": "list_overdue", "arguments": {}},
}
ov_out = c.post("/mcp", json=overdue_call, headers=mcp_h).json()
assert not ov_out["result"].get("isError"), ov_out
assert "Send invoice" in json.dumps(ov_out["result"])