theaio/theaio-boards
Laneway, team task boards you can run from Claude or ChatGPT. FastAPI + DataK3 + appid + Models + MCP.
A remix template · 0 remixes · Remix theaio/theaio-boards
tests/test_schema.py
"""schema.sql, the upgrade migration and SqlBoards' SQL, checked offline.
DataK3 is Postgres on the wire; SQLite stands in for it here (same keyed upserts, same plain
INSERT conflict on a primary key). Only the pgvector queries (<=>, = ANY) are left to DataK3.
"""
import re
import sqlite3
from datetime import date, datetime
from pathlib import Path
import psycopg
import pytest
import boards.infrastructure.repository as repository
from boards.application.use_cases import Boards
from boards.infrastructure.repository import SqlBoards
from conftest import MEMBER, OWNER, VIEWER
from fakes import Clock
ROOT = Path(__file__).resolve().parent.parent
SCHEMA = (ROOT / "schema.sql").read_text()
MIGRATION = (ROOT / "migrations" / "001_board_upgrade.sql").read_text()
# The starter's schema.sql exactly as it shipped: what an existing deployment already has.
STARTER = """
CREATE TABLE IF NOT EXISTS boards (id TEXT PRIMARY KEY, title TEXT NOT NULL,
owner_email TEXT NOT NULL, created_at TIMESTAMP);
CREATE TABLE IF NOT EXISTS memberships (id TEXT PRIMARY KEY, board_id TEXT NOT NULL,
email TEXT NOT NULL, role TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS cards (id TEXT PRIMARY KEY, board_id TEXT NOT NULL,
list_name TEXT NOT NULL, title TEXT NOT NULL, assignee TEXT, due_date DATE, updated_at TIMESTAMP);
CREATE TABLE IF NOT EXISTS activity (id TEXT PRIMARY KEY, board_id TEXT NOT NULL, card_id TEXT,
actor TEXT NOT NULL, action TEXT NOT NULL, event_at TIMESTAMP);
"""
sqlite3.register_adapter(date, lambda d: d.isoformat())
sqlite3.register_adapter(datetime, lambda d: d.isoformat())
sqlite3.register_converter("DATE", lambda b: date.fromisoformat(b.decode()))
sqlite3.register_converter("TIMESTAMP", lambda b: datetime.fromisoformat(b.decode()))
def statements(sql):
sql = re.sub(r"--[^\n]*", "", sql)
return [s.strip() for s in sql.split(";") if s.strip()]
def database(*scripts):
db = sqlite3.connect(":memory:", detect_types=sqlite3.PARSE_DECLTYPES,
check_same_thread=False)
for script in scripts:
for s in statements(script):
db.execute(s)
return db
def columns(db):
tables = [r[0] for r in db.execute("SELECT name FROM sqlite_master WHERE type = 'table'")]
return {t: {(r[1], r[2].upper(), r[5] > 0) for r in db.execute(f"PRAGMA table_info({t})")}
for t in tables}
def test_every_table_has_a_primary_key_and_no_sequences():
for s in statements(SCHEMA):
assert "PRIMARY KEY" in s, s
code = " ".join(statements(SCHEMA) + statements(MIGRATION))
assert not re.search(r"SERIAL|SEQUENCE|AUTOINCREMENT", code, re.I)
def test_the_migration_turns_the_starter_into_the_new_schema():
upgraded = columns(database(STARTER, MIGRATION))
fresh = columns(database(SCHEMA))
assert upgraded.keys() == fresh.keys()
for table in fresh: # same column names, types and keys (NOT NULL on old columns aside)
assert {c[::2] for c in upgraded[table]} == {c[::2] for c in fresh[table]}, table
assert {c[:2] for c in upgraded[table]} == {c[:2] for c in fresh[table]}, table
for s in statements(MIGRATION):
assert s.startswith(("ALTER TABLE", "CREATE TABLE IF NOT EXISTS")), s
class Lite:
"""Just enough of a psycopg connection for SqlBoards: %s placeholders, now(), 23505."""
def __init__(self, db):
self.db = db
def __enter__(self):
return self
def __exit__(self, *exc):
return False
def execute(self, sql, args=()):
sql = sql.replace("%s", "?").replace("now()", "CURRENT_TIMESTAMP")
try:
return self.db.execute(sql, args)
except sqlite3.IntegrityError as e:
raise psycopg.errors.UniqueViolation(str(e))
@pytest.fixture
def sql_boards(monkeypatch):
def make(*scripts):
db = database(*scripts)
monkeypatch.setattr(repository, "connect", lambda: Lite(db))
return db
return make
def test_the_use_cases_run_on_the_real_sql(sql_boards):
sql_boards(SCHEMA)
s = Boards(SqlBoards(), None, clock=Clock())
bid = s.create_board(OWNER, "Launch")["id"]
s.invite_member(OWNER, bid, MEMBER.email)
s.invite_member(OWNER, bid, VIEWER.email, "viewer")
s.add_list(MEMBER, bid, "Review")
assert [s.add_card(MEMBER, "Launch", t)["number"] for t in "abc"] == [1, 2, 3]
s.update_card(MEMBER, 2, assignee="me", due_date="2026-10-01", labels=["x", "y"])
s.move_card(MEMBER, 2, "Review", position=1)
s.block_card(MEMBER, 1, 2)
s.remove_member(OWNER, bid, VIEWER.email)
board = s.show_board(OWNER, bid)
assert [l["name"] for l in board["lists"]] == ["To do", "Doing", "Review", "Done"]
card = board["lists"][2]["cards"][0]
assert (card["number"], card["labels"], card["due_date"]) == (2, ["x", "y"], "2026-10-01")
assert [m["email"] for m in board["members"]] == [OWNER.email, MEMBER.email]
assert [h["field"] for h in s.card_history(OWNER, 2)["history"]] == [
"created", "assignee", "due_date", "labels", "list"]
assert s.what_is_blocking(OWNER, 1)["blocked"] is True
assert s.my_cards(MEMBER, 7)["cards"][0]["number"] == 2
assert s.repo.insert_card(s.repo.get_card(f"{bid}:1")) is False # the number is claimed
def test_a_starter_board_is_adopted_after_the_migration(sql_boards):
db = sql_boards(STARTER)
db.execute("INSERT INTO boards VALUES ('old', 'Old board', 'ana@x.io', NULL)")
db.execute("INSERT INTO memberships VALUES ('old:bo@x.io', 'old', 'bo@x.io', 'member')")
db.execute("INSERT INTO cards VALUES ('u1', 'old', 'Doing', 'legacy one', NULL, NULL, "
"'2026-01-01T00:00:00')")
db.execute("INSERT INTO cards VALUES ('u2', 'old', 'To do', 'legacy two', NULL, NULL, "
"'2026-01-02T00:00:00')")
for s in statements(MIGRATION):
db.execute(s)
s = Boards(SqlBoards(), None, clock=Clock())
board = s.show_board(MEMBER, "Old board")
assert {l["name"]: [c["number"] for c in l["cards"]] for l in board["lists"]} == {
"To do": [2], "Doing": [1], "Done": []}
assert s.add_card(MEMBER, "Old board", "new one")["number"] == 3
assert s.move_card(MEMBER, "legacy one", "Done")["list"] == "Done"