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_usability.py

"""Human references and membership lifecycle: scoped, explicit, and atomic."""
from concurrent.futures import ThreadPoolExecutor
import json

import pytest
from fastapi.testclient import TestClient

from boards.application.use_cases import Boards, NotFound
from boards.domain.model import Card, Conflict, Forbidden, Invalid, Person
from boards.infrastructure.memory import MemoryBoards
from boards.main import create_app

OWNER = Person('owner@example.test')
MEMBER = Person('member@example.test')
VIEWER = Person('viewer@example.test')
OUTSIDER = Person('outsider@example.test')


@pytest.fixture
def sample():
    service = Boards(MemoryBoards())
    bid = service.create_board(OWNER, 'Product Launch')['id']
    service.add_member(OWNER, bid, MEMBER.email)
    service.add_member(OWNER, bid, VIEWER.email, 'viewer')
    card = service.add_card(OWNER, bid, 'Send release notes', assignee=MEMBER.email)
    return service, bid, card


def headers(person):
    return {'X-Dodil-User': json.dumps({'email': person.email})}


def mcp_call(client, person, name, arguments):
    response = client.post('/mcp', headers={**headers(person), 'accept': 'application/json, text/event-stream'},
                           json={'jsonrpc': '2.0', 'id': 1, 'method': 'tools/call',
                                 'params': {'name': name, 'arguments': arguments}})
    assert response.status_code == 200
    return response.json()['result']


def payload(result):
    assert not result.get('isError'), result
    return json.loads(result['content'][0]['text'])


def test_card_reference_is_stable_across_changes_and_not_an_editable_field(sample):
    service, bid, card = sample
    reference = 'RB-' + card['id'][:8].upper()
    assert card['reference'] == reference
    changed = service.update_card(OWNER, card['id'], {'title': 'Renamed'}, card['revision'])
    moved = service.move_card(OWNER, card['id'], 'Doing', changed['revision'])
    assert changed['reference'] == moved['reference'] == reference
    with pytest.raises(Invalid):
        service.update_card(OWNER, card['id'], {'reference': 'RB-00000000'}, moved['revision'])
    assert all('reference' not in e['changes'] for e in service.activity(OWNER, bid))


@pytest.mark.parametrize('kind', ['board', 'card'])
def test_lookup_exact_names_are_trimmed_case_insensitive_and_read_only(sample, kind):
    service, bid, card = sample
    before = service.repo.snapshot()
    if kind == 'board':
        result = service.resolve_board(VIEWER, '  PRODUCT LAUNCH ')
        assert result['matches'][0]['id'] == bid
    else:
        result = service.resolve_card(VIEWER, '  SEND RELEASE NOTES  ')
        assert result['matches'][0] == {
            'id': card['id'], 'reference': card['reference'], 'title': card['title'],
            'board_id': bid, 'board_title': 'Product Launch', 'list_name': 'To do', 'revision': 1,
        }
    assert result['status'] == 'resolved'
    assert result['total_matches'] == 1 and not result['truncated']
    assert service.repo.snapshot() == before


def test_card_lookup_accepts_full_id_and_reference_not_position_or_partial_title(sample):
    service, _, card = sample
    for reference in (card['id'], card['reference'], card['reference'].lower()):
        assert service.resolve_card(MEMBER, reference)['matches'][0]['id'] == card['id']
    for reference in ('1', 'Send', card['id'][:8]):
        assert service.resolve_card(MEMBER, reference)['status'] == 'not_found'


def test_board_ids_take_priority_over_id_looking_titles(sample):
    service, bid, _ = sample
    service.create_board(OWNER, bid)
    assert service.resolve_board(OWNER, bid)['matches'][0]['id'] == bid
    assert service.resolve_board(OWNER, bid)['status'] == 'resolved'


def test_card_ids_take_priority_over_id_looking_titles(sample):
    service, bid, card = sample
    service.add_card(OWNER, bid, card['id'])
    assert service.resolve_card(OWNER, card['id'])['matches'][0]['id'] == card['id']
    assert service.resolve_card(OWNER, card['id'])['status'] == 'resolved'


def test_duplicate_board_names_are_ambiguous_only_for_people_with_access(sample):
    service, bid, _ = sample
    other = service.create_board(OWNER, 'PRODUCT LAUNCH')['id']
    result = service.resolve_board(OWNER, 'Product Launch')
    assert result['status'] == 'ambiguous'
    assert {b['id'] for b in result['matches']} == {bid, other}
    result = service.resolve_board(MEMBER, 'Product Launch')
    assert result['status'] == 'resolved' and result['total_matches'] == 1


def test_duplicate_card_titles_require_selection_or_board_scope(sample):
    service, bid, card = sample
    other = service.create_board(OWNER, 'Other board')['id']
    duplicate = service.add_card(OWNER, other, card['title'])
    result = service.resolve_card(OWNER, card['title'])
    assert result['status'] == 'ambiguous'
    assert {c['id'] for c in result['matches']} == {card['id'], duplicate['id']}
    assert service.resolve_card(OWNER, card['title'], bid)['status'] == 'resolved'
    # Scope is a canonical board id, not another opportunity to guess a name.
    with pytest.raises(NotFound):
        service.resolve_card(OWNER, card['title'], 'Product Launch')


def test_short_reference_collision_and_alias_looking_title_are_never_guessed(sample):
    service, bid, _ = sample
    for suffix in ('1' * 24, '2' * 24):
        service.repo.save_card(Card('abcdef01' + suffix, bid, 'Collision'))
    result = service.resolve_card(OWNER, 'RB-ABCDEF01')
    assert result['status'] == 'ambiguous' and result['total_matches'] == 2
    service.add_card(OWNER, bid, 'RB-ABCDEF01')
    result = service.resolve_card(OWNER, 'RB-ABCDEF01')
    assert result['status'] == 'ambiguous' and result['total_matches'] == 3


def test_hidden_candidates_never_leak_through_lookup_or_ambiguity_counts(sample):
    service, bid, card = sample
    hidden_bid = service.create_board(OUTSIDER, 'Product Launch')['id']
    hidden = service.add_card(OUTSIDER, hidden_bid, card['title'])
    assert service.resolve_card(MEMBER, card['title'])['total_matches'] == 1
    assert service.resolve_board(MEMBER, 'Product Launch')['total_matches'] == 1
    for reference in (hidden['id'], hidden['reference']):
        assert service.resolve_card(MEMBER, reference) == {
            'status': 'not_found', 'matches': [], 'total_matches': 0, 'truncated': False,
        }
    with pytest.raises(NotFound, match='no such board'):
        service.resolve_card(MEMBER, card['title'], hidden_bid)
    assert service.resolve_board(MEMBER, hidden_bid)['status'] == 'not_found'


@pytest.mark.parametrize('kind', ['board', 'card'])
def test_ambiguous_results_are_bounded_and_keep_the_true_count(sample, kind):
    service, bid, _ = sample
    for _ in range(22):
        if kind == 'board': service.create_board(OWNER, 'Duplicate')
        else: service.add_card(OWNER, bid, 'Duplicate')
    result = getattr(service, 'resolve_' + kind)(OWNER, 'Duplicate')
    assert result['status'] == 'ambiguous' and result['total_matches'] == 22
    assert result['truncated'] is True and len(result['matches']) == 20


@pytest.mark.parametrize('reference', ['', '  ', None, 1, 'x' * 181])
def test_lookup_rejects_invalid_references(sample, reference):
    service, _, _ = sample
    for method in (service.resolve_board, service.resolve_card):
        with pytest.raises(Invalid): method(OWNER, reference)


def test_removal_unassigns_all_board_cards_with_revisions_and_audit(sample):
    service, bid, card = sample
    done = service.add_card(OWNER, bid, 'Finished work', list_name='Done', assignee=MEMBER.email)
    other = service.create_board(OWNER, 'Unrelated')['id']
    service.add_member(OWNER, other, MEMBER.email)
    unrelated = service.add_card(OWNER, other, 'Keep assignment', assignee=MEMBER.email)
    result = service.remove_member(OWNER, bid, ' MEMBER@EXAMPLE.TEST ')
    assert result == {'board_id': bid, 'email': MEMBER.email, 'removed': True, 'unassigned_cards': 2}
    board = service.show_board(OWNER, bid)
    assert MEMBER.email not in {m['email'] for m in board['members']}
    for c in board['cards']:
        assert c['assignee'] is None and c['revision'] == 2
    assert service.show_board(MEMBER, other)['cards'][0] == unrelated
    events = service.activity(OWNER, bid)
    removal = events[0]
    assert removal['actor'] == OWNER.email and removal['event_at']
    assert removal['changes']['membership'] == {'before': {'email': MEMBER.email, 'role': 'member'}, 'after': None}
    unassigned = [e for e in events if e['action'] == 'unassigned after member removal']
    assert {e['card_id'] for e in unassigned} == {card['id'], done['id']}
    assert all(e['changes']['assignee'] == {'before': MEMBER.email, 'after': None} for e in unassigned)
    with pytest.raises(Conflict):
        service.update_card(OWNER, card['id'], {'title': 'Stale'}, card['revision'])


@pytest.mark.parametrize('action', ['board', 'activity', 'card_edit', 'lookup', 'add_card', 'remove'])
def test_removed_member_loses_all_access(sample, action):
    service, bid, card = sample
    service.remove_member(OWNER, bid, MEMBER.email)
    calls = {
        'board': lambda: service.show_board(MEMBER, bid),
        'activity': lambda: service.activity(MEMBER, bid),
        'card_edit': lambda: service.update_card(MEMBER, card['id'], {'title': 'Denied'}, card['revision']),
        'lookup': lambda: service.resolve_card(MEMBER, card['reference'], bid),
        'add_card': lambda: service.add_card(MEMBER, bid, 'Denied'),
        'remove': lambda: service.remove_member(MEMBER, bid, VIEWER.email),
    }
    with pytest.raises(NotFound): calls[action]()
    assert service.my_boards(MEMBER) == []
    assert service.my_cards(MEMBER) == []
    assert service.resolve_card(MEMBER, card['reference'])['status'] == 'not_found'
    assert service.resolve_board(MEMBER, 'Product Launch')['status'] == 'not_found'


@pytest.mark.parametrize('person,error', [(MEMBER, Forbidden), (VIEWER, Forbidden), (OUTSIDER, NotFound)])
def test_members_cannot_remove_themselves_or_others(sample, person, error):
    service, bid, _ = sample
    before = service.repo.snapshot()
    for email in (person.email, MEMBER.email, OWNER.email):
        with pytest.raises(error): service.remove_member(person, bid, email)
    assert service.repo.snapshot() == before


@pytest.mark.parametrize('email', [OWNER.email, OWNER.email.upper(), 'invalid', ''])
def test_removal_rejects_owner_and_invalid_email_without_writes(sample, email):
    service, bid, _ = sample
    before = service.repo.snapshot()
    with pytest.raises(Invalid): service.remove_member(OWNER, bid, email)
    assert service.repo.snapshot() == before


def test_repeated_removal_is_noop_and_readding_does_not_restore_assignments(sample):
    service, bid, _ = sample
    assert service.remove_member(OWNER, bid, MEMBER.email)['removed']
    before = service.repo.snapshot()
    assert service.remove_member(OWNER, bid, MEMBER.email) == {
        'board_id': bid, 'email': MEMBER.email, 'removed': False, 'unassigned_cards': 0,
    }
    assert service.repo.snapshot() == before
    service.add_member(OWNER, bid, MEMBER.email)
    assert service.show_board(MEMBER, bid)['cards'][0]['assignee'] is None


def test_removing_viewer_preserves_other_members(sample):
    service, bid, card = sample
    assert service.remove_member(OWNER, bid, VIEWER.email)['unassigned_cards'] == 0
    assert service.show_board(MEMBER, bid)['cards'][0] == card
    with pytest.raises(NotFound): service.show_board(VIEWER, bid)


@pytest.mark.parametrize('failure_stage', ['delete_membership', 'save_card', 'card_audit', 'membership_audit'])
def test_removal_rolls_back_membership_assignments_and_history(sample, monkeypatch, failure_stage):
    service, bid, _ = sample
    before = service.repo.snapshot()
    original_log = service.repo.log
    def fail(*args, **kwargs): raise RuntimeError('simulated storage failure')
    if failure_stage.endswith('_audit'):
        def log(board_id, card_id, *args, **kwargs):
            if (card_id is None) == (failure_stage == 'membership_audit'): fail()
            original_log(board_id, card_id, *args, **kwargs)
        monkeypatch.setattr(service.repo, 'log', log)
    else:
        monkeypatch.setattr(service.repo, failure_stage, fail)
    with pytest.raises(RuntimeError): service.remove_member(OWNER, bid, MEMBER.email)
    assert service.repo.snapshot() == before


def test_concurrent_assignment_and_removal_cannot_leave_removed_assignee(sample):
    service, bid, card = sample
    unassigned = service.assign_card(OWNER, card['id'], None, card['revision'])
    def assign():
        try:
            service.assign_card(OWNER, card['id'], MEMBER.email, unassigned['revision'])
            return 'assigned before removal'
        except Invalid:
            return 'rejected after removal'
    with ThreadPoolExecutor(max_workers=2) as pool:
        one = pool.submit(assign)
        two = pool.submit(service.remove_member, OWNER, bid, MEMBER.email)
        assert one.result() in ('assigned before removal', 'rejected after removal')
        assert two.result()['removed'] is True
    assert service.show_board(OWNER, bid)['cards'][0]['assignee'] is None
    with pytest.raises(NotFound): service.show_board(MEMBER, bid)


def test_web_lookup_and_removal_share_identity_and_domain_checks(sample):
    service, bid, card = sample
    path = f'/api/boards/{bid}/members'
    with TestClient(create_app(service)) as client:
        assert client.get('/api/resolve/boards', params={'reference': 'Product Launch'}).status_code == 401
        found = client.get('/api/resolve/cards', params={'reference': card['reference']}, headers=headers(VIEWER))
        assert found.status_code == 200 and found.json()['matches'][0]['revision'] == card['revision']
        denied = client.get('/api/resolve/cards', params={'reference': card['reference'], 'board_id': bid}, headers=headers(OUTSIDER))
        assert denied.status_code == 404
        assert client.request('DELETE', path, json={'email': MEMBER.email}).status_code == 401
        assert client.request('DELETE', path, json={'email': MEMBER.email}, headers=headers(VIEWER)).status_code == 403
        assert client.request('DELETE', path, json={'email': OWNER.email}, headers=headers(OWNER)).status_code == 422
        assert client.request('DELETE', path, json={'email': MEMBER.email, 'actor': OWNER.email}, headers=headers(OWNER)).status_code == 422
        response = client.request('DELETE', path, json={'email': MEMBER.email}, headers=headers(OWNER))
        assert response.status_code == 200 and response.json()['removed']
        assert client.get(f'/api/boards/{bid}', headers=headers(MEMBER)).status_code == 404


def test_mcp_lookup_ambiguity_and_removal_are_consistent_with_web(sample):
    service, bid, card = sample
    service.add_card(OWNER, bid, card['title'])
    with TestClient(create_app(service)) as client:
        found = payload(mcp_call(client, VIEWER, 'resolve_board', {'reference': 'product launch'}))
        assert found['matches'][0]['id'] == bid
        ambiguous = payload(mcp_call(client, OWNER, 'resolve_card', {'reference': card['title']}))
        assert ambiguous['status'] == 'ambiguous'
        assert mcp_call(client, MEMBER, 'remove_member', {'board_id': bid, 'email': VIEWER.email}).get('isError')
        removed = payload(mcp_call(client, OWNER, 'remove_member', {'board_id': bid, 'email': MEMBER.email}))
        assert removed['removed'] and removed['unassigned_cards'] == 1
        assert payload(mcp_call(client, MEMBER, 'resolve_card', {'reference': card['reference']}))['status'] == 'not_found'
        assert mcp_call(client, MEMBER, 'show_board', {'board_id': bid}).get('isError')