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/test_repository.py

"""The parts of the real repository the rules can't reach, and the promises schema.sql and the
Dodil files make. The rules themselves run on SqlBoards via tests/e2e/run.sh."""
import re
import time
from datetime import datetime, timedelta, timezone
from pathlib import Path

import psycopg
import pytest

from boards.infrastructure import repository
from boards.infrastructure.repository import SqlBoards

ROOT = Path(__file__).parent.parent
SCHEMA = (ROOT / "schema.sql").read_text()
TABLES = re.findall(r"CREATE TABLE IF NOT EXISTS (\w+) \((.*?)\n\);", SCHEMA, re.S)


class FakeConnection:
    closed, description = False, None

    def __init__(self, fails=False):
        self.fails, self.statements = fails, []

    def execute(self, statement, params=()):
        if self.fails:
            raise psycopg.OperationalError("server closed the connection")
        self.statements.append(statement)
        return self

    def close(self):
        self.closed = True


@pytest.fixture(autouse=True)
def no_retry_pauses(monkeypatch):
    monkeypatch.setattr(repository, "RETRY_PAUSES", (0, 0))


def repo_with(*connections):
    opened = list(connections)
    repo = SqlBoards(lambda: opened.pop(0))
    return repo, opened


def test_one_connection_serves_a_thread_across_queries():
    good = FakeConnection()
    repo, _ = repo_with(good)
    repo.boards_for("a@x.io")
    repo.boards_for("a@x.io")
    assert len(good.statements) == 4


def test_a_stale_connection_is_replaced_and_the_query_still_runs():
    stale, fresh = FakeConnection(fails=True), FakeConnection()
    repo, _ = repo_with(stale, fresh)
    repo.boards_for("a@x.io")
    assert stale.closed and len(fresh.statements) == 2


def test_a_connection_drop_that_lasts_two_tries_is_survived():
    """Seen on DataK3: "tcp connect error" twice in a row, then fine."""
    first, second = FakeConnection(fails=True), FakeConnection(fails=True)
    fresh = FakeConnection()
    repo, _ = repo_with(first, second, fresh)
    repo.boards_for("a@x.io")
    assert first.closed and second.closed and len(fresh.statements) == 2


def test_a_database_that_stays_down_is_reported_not_hidden():
    repo, _ = repo_with(*(FakeConnection(fails=True) for _ in range(3)))
    with pytest.raises(psycopg.OperationalError):
        repo.boards_for("a@x.io")


def test_datak3_timestamps_become_plain_utc():
    """DataK3 returns TIMESTAMP with a zone, Postgres without. The rest of the app must not care:
    a zone broke the dates in history and crashed the board summary."""
    chicago = timezone(timedelta(hours=-5))
    row = ("e1", "b1", "c1", "a@x.io", "moved it", datetime(2026, 11, 1, 16, 53, tzinfo=chicago),
           "list", "To do", "Doing")
    assert repository._event(row).at == datetime(2026, 11, 1, 21, 53)
    plain = ("e1", "b1", "c1", "a@x.io", "moved it", datetime(2026, 11, 1, 21, 53), "", "", "")
    assert repository._event(plain).at == datetime(2026, 11, 1, 21, 53)


def test_an_old_connection_is_retired(monkeypatch):
    old, fresh = FakeConnection(), FakeConnection()
    repo, _ = repo_with(old, fresh)
    repo.boards_for("a@x.io")
    later = time.time() + repository.MAX_CONNECTION_AGE + 1
    monkeypatch.setattr(repository.time, "time", lambda: later)
    repo.boards_for("a@x.io")
    assert old.closed and fresh.statements


def test_every_table_has_a_chosen_primary_key():
    assert len(TABLES) == 6
    for name, body in TABLES:
        assert "PRIMARY KEY" in body, name


def test_schema_stays_plain_sql():
    code = "\n".join(line.partition("--")[0] for line in SCHEMA.splitlines()).upper()
    for feature in ("SERIAL", "SEQUENCE", "TRIGGER", "FUNCTION", "GENERATED", "JSONB", "RETURNING",
                    "REFERENCES", "EXTENSION"):
        assert feature not in code, feature


def test_schema_comments_can_not_split_a_statement():
    """A remix tool may split on semicolons, so comments mustn't contain one."""
    assert not any(";" in line.partition("--")[2] for line in SCHEMA.splitlines())


def test_repository_uses_only_selects_and_inserts():
    source = Path(repository.__file__).read_text()
    statements = re.findall(r'"((?:SELECT|INSERT|UPDATE|DELETE|WITH|CREATE|ALTER|DROP)\b)', source)
    assert set(statements) <= {"SELECT", "INSERT"}, set(statements)
    for feature in ("RETURNING", "UNION", "JOIN", "BEGIN", "COMMIT", ".transaction("):
        assert feature not in source, feature


def test_contest_files_still_fit_the_rules():
    remix = (ROOT / ".dodil/remix.yaml").read_text()
    deploy = (ROOT / ".dodil/deploy.yaml").read_text()
    assert '"${{ dodil.repo }}-data"' in remix and '"${{ dodil.repo }}-data"' in deploy
    assert '"${{ dodil.repo }}-users"' in remix and '"${{ dodil.repo }}-users"' in deploy
    assert "on_conflict" not in remix        # the default, fail, is the rule
    assert "schema: schema.sql" in remix
    assert "OPENAI" not in deploy.upper() and "ANTHROPIC" not in deploy.upper()
    assert len("otter-boards") <= 18


def test_no_outside_ai_or_database_keys_in_the_app():
    for path in (ROOT / "app" / "boards").rglob("*.py"):
        text = path.read_text().upper()
        assert "OPENAI_API_KEY" not in text and "ANTHROPIC_API_KEY" not in text, path