stubborn-otter/otter-boards

Otter Boards, a team task board you can run from the web or from Claude and ChatGPT. FastAPI + DataK3 + appid + Models + MCP.

• boards
• tasks
• fastapi
• mcp
• dodil-challenge
• dodil-challenge-boards

A remix template · 1 remixes · Remix stubborn-otter/otter-boards

tests/conftest.py

"""Shared fixtures. By default the rules run on the in-memory fake.

Set TEST_DATABASE_URL to a throwaway Postgres loaded with schema.sql and every rules test runs a
second time on the real SqlBoards repository. tests/e2e/run.sh does that for you.
"""
import json
import os
from datetime import datetime, timedelta

import pytest
from fastapi.testclient import TestClient
from fakes import FakeAI, MemoryBoards

from boards.application.use_cases import Boards
from boards.domain.model import 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"))
NOW = datetime(2026, 6, 10, 9, 0, 0)  # a Wednesday
TODAY = "2026-06-10"


def _repo(kind):
    if kind == "memory":
        return MemoryBoards()
    import psycopg

    from boards.infrastructure.repository import SqlBoards
    url = os.environ["TEST_DATABASE_URL"]
    with psycopg.connect(url, autocommit=True) as db:   # every test starts from an empty bucket
        db.execute("TRUNCATE boards, memberships, lists, cards, activity, card_embeddings")
    return SqlBoards(lambda: psycopg.connect(url, autocommit=True))


def pytest_generate_tests(metafunc):
    if "repo" in metafunc.fixturenames:
        kinds = ["memory"] + (["sql"] if os.environ.get("TEST_DATABASE_URL") else [])
        metafunc.parametrize("repo", kinds, indirect=True)


@pytest.fixture
def repo(request):
    return _repo(request.param)


class Clock:
    """A clock that ticks, so events in one test still sort in order."""

    def __init__(self):
        self.value = NOW

    def __call__(self):
        self.value += timedelta(microseconds=10)
        return self.value


@pytest.fixture
def ai():
    return FakeAI()


@pytest.fixture
def svc(repo, ai):
    """A board with an owner, a member and a viewer."""
    s = Boards(repo, ai, now=Clock())
    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 header(person):
    return {"X-Dodil-User": json.dumps({"sub": person.email, "email": person.email})}


class McpCaller:
    """Speaks real MCP JSON-RPC to the app's /mcp, as a client would."""

    def __init__(self, client, person):
        self.client, self.n = client, 0
        self.headers = {**header(person), "accept": "application/json, text/event-stream"}
        r = self._send("initialize", {"protocolVersion": "2025-06-18", "capabilities": {},
                                      "clientInfo": {"name": "tests", "version": "0"}})
        assert r.status_code == 200, r.text

    def _send(self, method, params):
        self.n += 1
        return self.client.post("/mcp", headers=self.headers, json={
            "jsonrpc": "2.0", "id": self.n, "method": method, "params": params})

    def raw(self, tool, **arguments):
        out = self._send("tools/call", {"name": tool, "arguments": arguments}).json()
        assert "result" in out, out
        return out["result"]

    def call(self, tool, **arguments):
        """The tool's data; fails the test if the tool refused."""
        result = self.raw(tool, **arguments)
        assert not result.get("isError"), result
        data = result.get("structuredContent")
        return data["result"] if isinstance(data, dict) and set(data) == {"result"} else data

    def error(self, tool, **arguments):
        """The tool's refusal text; fails the test if the tool worked."""
        result = self.raw(tool, **arguments)
        assert result.get("isError"), result
        return result["content"][0]["text"]

    def tools(self):
        return self._send("tools/list", {}).json()["result"]["tools"]


@pytest.fixture
def web(svc):
    s, _ = svc
    with TestClient(create_app(s)) as client:
        yield client


@pytest.fixture
def mcp_as(web):
    return lambda person: McpCaller(web, person)