north-scape/relayboard
Relayboard: accessible team boards with audited actions, dependency maps, and shared web/MCP use cases.
A remix template · 0 remixes · Remix north-scape/relayboard
tests/test_sql_contract.py
"""Adapter contract checks using a recording connection, not a SQL engine.
These prove parameter binding/transaction plumbing; live DataK3 validation is still required.
"""
from contextlib import contextmanager
from datetime import date, datetime, timezone
import json
import pytest
import psycopg
from boards.domain.model import Board, Card, Conflict, Role
from boards.infrastructure import repository as sql
class Result:
def __init__(self, rows=()): self.rows = list(rows)
def fetchone(self): return self.rows[0] if self.rows else None
def fetchall(self): return self.rows
def __iter__(self): return iter(self.rows)
class Connection:
def __init__(self, responder=None):
self.calls = []
self.responder = responder or (lambda query, params: [])
self.transactions = []
self.fail_commit = False
def __enter__(self): return self
def __exit__(self, *args): pass
def execute(self, query, params=()):
self.calls.append((query, params))
return Result(self.responder(query, params))
@contextmanager
def transaction(self):
self.transactions.append('begin')
try:
yield
if self.fail_commit: raise psycopg.errors.SerializationFailure('conflict')
except BaseException:
self.transactions.append('rollback')
raise
else:
self.transactions.append('commit')
@pytest.fixture
def connection(monkeypatch):
conn = Connection()
opened = []
def connect():
opened.append(conn)
return conn
monkeypatch.setattr(sql, 'connect', connect)
return conn, opened
def test_mutation_and_audit_use_one_connection_and_transaction(connection):
conn, opened = connection
repo = sql.SqlBoards()
card = Card('card', 'board', "Robert'); DROP TABLE cards;--", description='literal content', labels=['risk'])
with repo.atomic():
repo.save_card(card)
repo.log('board', 'card', 'owner@example.test', 'added card', {'title': {'after': card.title}})
assert len(opened) == 1
assert conn.transactions == ['begin', 'commit']
assert sql._connection.get() is None
inserts = [(q, p) for q, p in conn.calls if q.startswith('INSERT')]
assert len(inserts) == 2
assert all(card.title not in q for q, _ in inserts)
assert card.title in inserts[0][1]
assert json.loads(inserts[1][1][-1])['title']['after'] == card.title
def test_failed_mutation_rolls_back_and_resets_context(connection):
conn, _ = connection
with pytest.raises(RuntimeError):
with sql.SqlBoards().atomic():
raise RuntimeError('failed audit write')
assert conn.transactions == ['begin', 'rollback']
assert sql._connection.get() is None
def test_serialization_conflict_becomes_safe_domain_conflict(connection):
conn, _ = connection
conn.fail_commit = True
with pytest.raises(Conflict, match='reload'):
with sql.SqlBoards().atomic():
pass
assert sql._connection.get() is None
assert conn.transactions == ['begin', 'rollback']
def test_mutating_board_read_locks_and_touches_aggregate_guard(connection):
conn, _ = connection
conn.responder = lambda q, p: [('board', 'Title', 'owner@example.test')] if q.startswith('SELECT id, title') else []
with sql.SqlBoards().atomic():
board = sql.SqlBoards().get('board')
assert board.id == 'board'
assert conn.calls[0][0].endswith('FOR UPDATE')
assert any(q.startswith('UPDATE boards SET revision') for q, _ in conn.calls)
def test_read_only_board_fetch_neither_locks_nor_writes(connection):
conn, _ = connection
conn.responder = lambda q, p: [('board', 'Title', 'owner@example.test')] if q.startswith('SELECT id, title') else []
assert sql.SqlBoards().get('board').title == 'Title'
assert all('FOR UPDATE' not in q and not q.startswith('UPDATE') for q, _ in conn.calls)
assert conn.transactions == []
def test_card_sql_roundtrip_column_order_and_json():
row = ('card', 'board', 'Title', 'Doing', 'member@example.test', date(2026, 11, 1),
'Description', '["critical"]', '["dependency"]', 4, '[{"id":"step","text":"Verify","done":true}]')
card = sql._card(row)
assert card.title == 'Title'
assert card.description == 'Description'
assert card.labels == ['critical']
assert card.blocked_by == ['dependency']
assert card.revision == 4
assert card.checklist == [{'id': 'step', 'text': 'Verify', 'done': True}]
assert card.due_date == date(2026, 11, 1)
assert sql._card(None) is None
def test_update_card_is_explicit_update_not_replace_upsert(connection):
conn, _ = connection
conn.responder = lambda q, p: [('card',)] if q.startswith('SELECT id FROM cards') else []
sql.SqlBoards().save_card(Card('card', 'board', 'Updated'))
statement, params = conn.calls[-1]
assert statement.startswith('UPDATE cards SET')
assert statement.endswith('WHERE id=%s')
assert params[-1] == 'card'
assert 'ON CONFLICT' not in statement
def test_activity_is_board_scoped_and_decodes_structured_audit(connection):
conn, _ = connection
stamp = datetime(2026, 10, 30, tzinfo=timezone.utc)
conn.responder = lambda q, p: [('event', 'board', 'card', 'owner@example.test', 'updated card', '{"title":{"before":"a","after":"b"}}', stamp)]
result = sql.SqlBoards().activity('board')
assert conn.calls[0][1] == ('board',)
assert 'WHERE board_id = %s' in conn.calls[0][0]
assert result[0]['changes']['title'] == {'before': 'a', 'after': 'b'}
assert result[0]['event_at'] == stamp.isoformat()
def test_tls_adapter_defaults_verify_hostname_and_system_ca(monkeypatch):
from boards.infrastructure import dodil
monkeypatch.delenv('PG_SSLMODE', raising=False)
monkeypatch.delenv('PG_SSLROOTCERT', raising=False)
monkeypatch.setenv('BUCKET', 'test-bucket')
monkeypatch.setattr(dodil, 'token', lambda: 'fake-token-for-test')
captured = {}
def connect(**kwargs):
captured.update(kwargs)
return object()
monkeypatch.setattr(dodil.psycopg, 'connect', connect)
dodil.connect()
assert captured['sslmode'] == 'verify-full'
assert captured['sslrootcert'] == '/etc/ssl/certs/ca-certificates.crt'
assert captured['dbname'] == 'test-bucket'
assert captured['connect_timeout'] == 15
assert captured['autocommit'] is True
def test_membership_deletion_is_parameterized_and_board_scoped(connection):
conn, _ = connection
email = "quote'@example.test"
sql.SqlBoards().delete_membership('one-board', email)
statement, params = conn.calls[-1]
assert statement == 'DELETE FROM memberships WHERE board_id = %s AND email = %s'
assert params == ('one-board', email)
assert email not in statement
def test_member_removal_uses_aggregate_guard_and_one_atomic_connection(connection):
from boards.application.use_cases import Boards
from boards.domain.model import Person
conn, opened = connection
member = 'member@example.test'
card_row = ('card', 'board', 'Title', 'Doing', member, None, '', '[]', '[]', 3, '[]')
def respond(query, params):
if query.startswith('SELECT id, title'): return [('board', 'Title', 'owner@example.test')]
if query.startswith('SELECT email, role'): return [(member, 'member')]
if query.startswith('SELECT id, board_id'): return [card_row]
if query.startswith('SELECT id FROM cards'): return [('card',)]
return []
conn.responder = respond
result = Boards(sql.SqlBoards()).remove_member(Person('owner@example.test'), 'board', member)
assert result['removed'] and result['unassigned_cards'] == 1
assert len(opened) == 1 and conn.transactions == ['begin', 'commit']
queries = [q for q, _ in conn.calls]
lock = next(i for i, q in enumerate(queries) if q.endswith('FOR UPDATE'))
guard = next(i for i, q in enumerate(queries) if q.startswith('UPDATE boards SET revision'))
deletion = next(i for i, q in enumerate(queries) if q.startswith('DELETE FROM memberships'))
assert lock < guard < deletion
update = next(p for q, p in conn.calls if q.startswith('UPDATE cards SET'))
assert update[3] is None and update[8] == 4
assert sum(q.startswith('INSERT INTO activity') for q in queries) == 2
assert sql._connection.get() is None
@pytest.mark.parametrize('existing', [False, True])
def test_checklist_storage_is_parameterized_json_on_create_and_update(connection, existing):
conn, _ = connection
conn.responder = lambda q, p: [('card',)] if existing and q.startswith('SELECT id FROM cards') else []
steps = [{'id': 'step', 'text': "Verify'); DROP TABLE cards;--", 'done': True}]
sql.SqlBoards().save_card(Card('card', 'board', 'Title', checklist=steps))
query, params = conn.calls[-1]
assert 'checklist' in query
assert steps[0]['text'] not in query
assert json.loads(params[9]) == steps
assert params[-1] == 'card'
assert query.count('%s') == len(params)
def test_migrated_null_checklist_reads_as_empty():
card = sql._card(('card', 'board', 'Title', 'To do', None, None, '', '[]', '[]', 1, None))
assert card.checklist == []
# These are recording-adapter contracts, not proof of a live engine's locking
# or rollback semantics. A concurrent writer is simulated between the first
# card lookup and the aggregate lock; the second lookup must be authoritative.
CARD_ACTIONS = ('update_card', 'assign_card', 'move_card', 'set_blockers', 'update_checklist')
OMITTED_REVISION = object()
def card_action(service, name, revision=OMITTED_REVISION):
from boards.domain.model import Person
args = {
'update_card': {'title': 'Changed'},
'assign_card': 'owner@example.test',
'move_card': 'Done',
'set_blockers': [],
'update_checklist': [{'text': 'Reviewed step'}],
}
# update_checklist already has a required Python argument. Other application
# entry points translate omission into Invalid after permission checks.
kwargs = {} if revision is OMITTED_REVISION else {'expected_revision': revision}
return getattr(service, name)(Person('owner@example.test'), 'card', args[name], **kwargs)
def changing_card_responder(conn):
locked = False
def row(revision):
return ('card', 'board', 'Original', 'Doing', None, None, '', '[]', '["dependency"]', revision, '[]')
def respond(query, params):
nonlocal locked
if query.startswith('SELECT id, title'):
assert query.endswith('FOR UPDATE')
locked = True
return [('board', 'Title', 'owner@example.test')]
if query.startswith('SELECT id, board_id'):
return [row(2 if locked else 1)]
if query.startswith('SELECT id FROM cards'):
return [('card',)]
return []
conn.responder = respond
def assert_locked_card_reread(conn):
queries = [q for q, _ in conn.calls]
reads = [i for i, q in enumerate(queries) if q.startswith('SELECT id, board_id') and 'WHERE id = %s' in q]
lock = next(i for i, q in enumerate(queries) if q.endswith('FOR UPDATE'))
guard = next(i for i, q in enumerate(queries) if q.startswith('UPDATE boards SET revision'))
assert len(reads) == 2
assert reads[0] < lock < guard < reads[1]
return queries, reads[1]
@pytest.mark.parametrize('action', CARD_ACTIONS)
def test_card_mutations_refuse_stale_revision_from_before_aggregate_lock(connection, action):
from boards.application.use_cases import Boards
conn, opened = connection
changing_card_responder(conn)
with pytest.raises(Conflict, match='reload'):
card_action(Boards(sql.SqlBoards()), action, 1)
queries, _ = assert_locked_card_reread(conn)
assert not any(q.startswith(('UPDATE cards', 'INSERT INTO cards', 'INSERT INTO activity')) for q in queries)
# The aggregate guard was touched, but its transaction must not commit.
assert conn.transactions == ['begin', 'rollback']
assert len(opened) == 1 and sql._connection.get() is None
@pytest.mark.parametrize('action', CARD_ACTIONS)
@pytest.mark.parametrize('revision', [None, True, False, '2', 2.0, 2.5])
def test_malformed_revision_rolls_back_guard_without_card_or_history_writes(connection, action, revision):
from boards.application.use_cases import Boards
from boards.domain.model import Invalid
conn, opened = connection
changing_card_responder(conn)
with pytest.raises(Invalid, match='expected_revision'):
card_action(Boards(sql.SqlBoards()), action, revision)
queries, _ = assert_locked_card_reread(conn)
assert not any(q.startswith(('UPDATE cards', 'INSERT INTO cards', 'INSERT INTO activity')) for q in queries)
assert conn.transactions == ['begin', 'rollback']
assert len(opened) == 1 and sql._connection.get() is None
@pytest.mark.parametrize('action', CARD_ACTIONS[:-1])
def test_omitted_revision_rolls_back_guard_without_card_or_history_writes(connection, action):
from boards.application.use_cases import Boards
from boards.domain.model import Invalid
conn, opened = connection
changing_card_responder(conn)
with pytest.raises(Invalid, match='expected_revision'):
card_action(Boards(sql.SqlBoards()), action)
queries, _ = assert_locked_card_reread(conn)
assert not any(q.startswith(('UPDATE cards', 'INSERT INTO cards', 'INSERT INTO activity')) for q in queries)
assert conn.transactions == ['begin', 'rollback']
assert len(opened) == 1 and sql._connection.get() is None
@pytest.mark.parametrize('action', CARD_ACTIONS)
def test_current_locked_revision_commits_card_and_history_in_one_transaction(connection, action):
from boards.application.use_cases import Boards
conn, opened = connection
changing_card_responder(conn)
result = card_action(Boards(sql.SqlBoards()), action, 2)
assert result['revision'] == 3
queries, last_read = assert_locked_card_reread(conn)
writes = [(i, p) for i, (q, p) in enumerate(conn.calls) if q.startswith('UPDATE cards')]
audits = [(i, p) for i, (q, p) in enumerate(conn.calls) if q.startswith('INSERT INTO activity')]
assert len(writes) == len(audits) == 1
assert last_read < writes[0][0] < audits[0][0]
assert writes[0][1][8] == 3 and writes[0][1][-1] == 'card'
assert audits[0][1][1:4] == ('board', 'card', 'owner@example.test')
assert json.loads(audits[0][1][-1])
assert conn.transactions == ['begin', 'commit']
assert len(opened) == 1 and sql._connection.get() is None