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_revisions.py
"""Mandatory optimistic concurrency across the application, JSON and MCP doors.
Only local in-memory state: every refusal preserves cards, memberships and the
entire audit history, not merely the endpoint's response status.
"""
from concurrent.futures import ThreadPoolExecutor
import json
from threading import Barrier
import pytest
from fastapi.testclient import TestClient
from boards.application.use_cases import Boards, NotFound
from boards.domain.model import Conflict, Forbidden, Invalid, Person
from boards.infrastructure.memory import MemoryBoards
from boards.main import create_app
OWNER, MEMBER, VIEWER, OUTSIDER = [Person(x + '@example.test') for x in ('owner', 'member', 'viewer', 'outsider')]
ACTIONS = ('update_card', 'assign_card', 'move_card', 'set_blockers', 'update_checklist')
MISSING = object()
BAD_REVISIONS = [pytest.param(MISSING, id='omitted'), None, True, False, '1', 1.0, 1.5, [], {}]
@pytest.fixture
def setup():
service = Boards(MemoryBoards())
bid = service.create_board(OWNER, 'Revision contract')['id']
service.add_member(OWNER, bid, MEMBER.email)
service.add_member(OWNER, bid, VIEWER.email, 'viewer')
card = service.add_card(OWNER, bid, 'Observed task')
blocker = service.add_card(OWNER, bid, 'Observed prerequisite')
return service, bid, card, blocker
def value(action, blocker, current=None):
if action == 'update_card': return {'title': current['title'] if current else 'Reviewed title'}
if action == 'assign_card': return current['assignee'] if current else MEMBER.email
if action == 'move_card': return current['list_name'] if current else 'Doing'
if action == 'set_blockers': return current['blocked_by'] if current else [blocker['id']]
return current['checklist'] if current else [{'text': 'Reviewed step'}]
def revision_arg(revision):
return {} if revision is MISSING else {'expected_revision': revision}
def invoke(service, action, card, blocker, revision, person=MEMBER, current=None):
return getattr(service, action)(person, card['id'], value(action, blocker, current), **revision_arg(revision))
def headers(person):
return {'X-Dodil-User': json.dumps({'email': person.email}), 'accept': 'application/json, text/event-stream'}
def web_call(client, action, card, blocker, revision, person=MEMBER, current=None):
field = {'update_card': None, 'assign_card': 'assignee', 'move_card': 'list_name',
'set_blockers': 'blocked_by', 'update_checklist': 'checklist'}[action]
data = value(action, blocker, current)
body = {**(data if field is None else {field: data}), **revision_arg(revision)}
method, suffix = {'move_card': ('POST', '/move'), 'set_blockers': ('POST', '/blockers')}.get(action, ('PATCH', ''))
return client.request(method, '/api/cards/' + card['id'] + suffix, json=body, headers=headers(person) if person else {})
def mcp_call(client, action, card, blocker, revision, person=MEMBER, current=None):
field = {'update_card': 'changes', 'assign_card': 'email', 'move_card': 'list_name',
'set_blockers': 'blocked_by', 'update_checklist': 'items'}[action]
response = client.post('/mcp', headers=headers(person) if person else {}, json={
'jsonrpc': '2.0', 'id': 1, 'method': 'tools/call',
'params': {'name': action, 'arguments': {'card_id': card['id'], field: value(action, blocker, current), **revision_arg(revision)}},
})
assert response.status_code == 200, response.text
return response.json()['result']
@pytest.mark.parametrize('action', ACTIONS)
@pytest.mark.parametrize('revision', BAD_REVISIONS)
def test_application_refuses_missing_or_noninteger_revisions_without_writes(setup, action, revision):
service, _, card, blocker = setup
before = service.repo.snapshot()
# update_checklist already has a mandatory Python argument; its omission is
# a call-signature error. The other entry points report a domain error only
# after checking permissions, preserving the existing authorization order.
error = TypeError if action == 'update_checklist' and revision is MISSING else Invalid
with pytest.raises(error): invoke(service, action, card, blocker, revision)
assert service.repo.snapshot() == before
@pytest.mark.parametrize('action', ACTIONS)
@pytest.mark.parametrize('revision', BAD_REVISIONS)
def test_web_refuses_missing_or_coerced_revisions_without_writes(setup, action, revision):
service, _, card, blocker = setup
before = service.repo.snapshot()
with TestClient(create_app(service)) as client:
response = web_call(client, action, card, blocker, revision)
assert response.status_code == 422, response.text
assert service.repo.snapshot() == before
@pytest.mark.parametrize('action', ACTIONS)
@pytest.mark.parametrize('revision', BAD_REVISIONS)
def test_mcp_refuses_missing_or_coerced_revisions_without_writes(setup, action, revision):
service, _, card, blocker = setup
before = service.repo.snapshot()
with TestClient(create_app(service)) as client:
result = mcp_call(client, action, card, blocker, revision)
assert result.get('isError'), result
assert 'expected_revision' in result['content'][0]['text']
assert service.repo.snapshot() == before
@pytest.mark.parametrize('action', ACTIONS)
@pytest.mark.parametrize('transport', ['application', 'web', 'mcp'])
@pytest.mark.parametrize('revision', [0, -1, 1, 3])
def test_stale_future_and_nonpositive_revisions_never_write(setup, action, transport, revision):
service, _, card, blocker = setup
service.update_card(OWNER, card['id'], {'description': 'A newer edit'}, card['revision'])
before = service.repo.snapshot()
if transport == 'application':
with pytest.raises(Conflict, match='changed'):
invoke(service, action, card, blocker, revision)
else:
with TestClient(create_app(service)) as client:
if transport == 'web':
response = web_call(client, action, card, blocker, revision)
assert response.status_code == 409 and 'changed' in response.text
else:
result = mcp_call(client, action, card, blocker, revision)
assert result.get('isError') and 'changed' in result['content'][0]['text']
assert service.repo.snapshot() == before
@pytest.mark.parametrize('action', ACTIONS)
@pytest.mark.parametrize('transport', ['application', 'web', 'mcp'])
def test_current_revision_commits_once_and_current_noop_adds_no_history(setup, action, transport):
service, bid, card, blocker = setup
event_count = len(service.repo.events)
with TestClient(create_app(service)) as client:
def perform(revision, current=None):
if transport == 'application': return invoke(service, action, card, blocker, revision, current=current)
if transport == 'web':
response = web_call(client, action, card, blocker, revision, current=current)
assert response.status_code == 200, response.text
return response.json()
result = mcp_call(client, action, card, blocker, revision, current=current)
assert not result.get('isError'), result
return json.loads(result['content'][0]['text'])
saved = perform(card['revision'])
assert saved['revision'] == card['revision'] + 1
assert len(service.repo.events) == event_count + 1
event = service.activity(OWNER, bid)[0]
assert event['actor'] == MEMBER.email and event['card_id'] == card['id'] and event['changes']
before = service.repo.snapshot()
assert perform(saved['revision'], saved) == saved
assert service.repo.snapshot() == before
# Even a no-op must not accept an old observation or silently turn a
# lost-response retry into a success acknowledgment.
if transport == 'application':
with pytest.raises(Conflict): perform(card['revision'], saved)
elif transport == 'web':
assert web_call(client, action, card, blocker, card['revision'], current=saved).status_code == 409
else:
assert mcp_call(client, action, card, blocker, card['revision'], current=saved).get('isError')
assert service.repo.snapshot() == before
@pytest.mark.parametrize('action', ACTIONS)
@pytest.mark.parametrize('revision', [None, True, 0, 1])
@pytest.mark.parametrize('person,error', [(VIEWER, Forbidden), (OUTSIDER, NotFound)])
def test_application_checks_visibility_and_permission_before_revision(setup, action, revision, person, error):
service, _, card, blocker = setup
before = service.repo.snapshot()
with pytest.raises(error): invoke(service, action, card, blocker, revision, person)
assert service.repo.snapshot() == before
@pytest.mark.parametrize('action', ACTIONS)
@pytest.mark.parametrize('person,status,message', [(VIEWER, 403, 'viewers'), (OUTSIDER, 404, 'no such card')])
def test_doors_hide_private_cards_and_preserve_viewer_permissions(setup, action, person, status, message):
service, _, card, blocker = setup
before = service.repo.snapshot()
with TestClient(create_app(service)) as client:
# A well-formed but stale revision cannot be used as an existence or
# revision oracle by a viewer or outsider.
response = web_call(client, action, card, blocker, 0, person)
assert response.status_code == status and message in response.text
result = mcp_call(client, action, card, blocker, 0, person)
assert result.get('isError') and message in result['content'][0]['text']
assert web_call(client, action, card, blocker, 1, None).status_code == 401
assert service.repo.snapshot() == before
@pytest.mark.parametrize('action', ACTIONS)
def test_simultaneous_different_actions_on_same_revision_have_one_winner(setup, action):
service, _, card, blocker = setup
barrier = Barrier(2)
event_count = len(service.repo.events)
def edit(which):
barrier.wait()
try:
if which == 'competing':
service.update_card(OWNER, card['id'], {'description': 'Competing edit'}, card['revision'])
else:
invoke(service, action, card, blocker, card['revision'])
return 'saved'
except Conflict:
return 'conflict'
with ThreadPoolExecutor(max_workers=2) as executor:
results = list(executor.map(edit, ['competing', 'requested']))
assert sorted(results) == ['conflict', 'saved']
assert service.repo.get_card(card['id']).revision == 2
assert len(service.repo.events) == event_count + 1
def test_web_and_mcp_publish_required_integer_revision_schema():
with TestClient(create_app(Boards(MemoryBoards()))) as client:
response = client.post('/mcp', headers=headers(OWNER), json={'jsonrpc': '2.0', 'id': 1, 'method': 'tools/list'})
tools = {tool['name']: tool for tool in response.json()['result']['tools']}
for action in ACTIONS:
schema = tools[action]['inputSchema']
assert 'expected_revision' in schema['required']
assert schema['properties']['expected_revision']['type'] == 'integer'
assert 'default' not in schema['properties']['expected_revision']
from boards.doors.web import Blockers, Move, Patch
for model in (Patch, Move, Blockers):
schema = model.model_json_schema()
assert 'expected_revision' in schema['required']
assert schema['properties']['expected_revision']['type'] == 'integer'
assert 'default' not in schema['properties']['expected_revision']