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

"""Adversarial acceptance checks. No real account, database, or AI calls."""
import json
from datetime import date, timedelta

import pytest
from fastapi.testclient import TestClient

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

OWNER = Person('owner@example.test')
MEMBER = Person('member@example.test')
VIEWER = Person('viewer@example.test')
STRANGER = Person('stranger@example.test')


class RecordingAI:
    def __init__(self):
        self.calls = []

    def checklist(self, title):
        self.calls.append(('checklist', title))
        return ['Confirm scope', 'Deliver and verify']


@pytest.fixture
def extended():
    ai = RecordingAI()
    service = Boards(MemoryBoards(), ai)
    bid = service.create_board(OWNER, 'Launch')['id']
    service.add_member(OWNER, bid, MEMBER.email, 'member')
    service.add_member(OWNER, bid, VIEWER.email, 'viewer')
    card = service.add_card(OWNER, bid, 'Ship the release')
    return service, bid, card['id'], ai


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


@pytest.mark.parametrize('action', ['add', 'move', 'edit', 'assign', 'dependencies'])
def test_viewer_cannot_mutate_any_card_action(extended, action):
    service, bid, cid, _ = extended
    before = service.show_board(OWNER, bid)
    calls = {
        'add': lambda: service.add_card(VIEWER, bid, 'Forbidden'),
        'move': lambda: service.move_card(VIEWER, cid, 'Done', 1),
        'edit': lambda: service.update_card(VIEWER, cid, {'description': 'Forbidden'}, 1),
        'assign': lambda: service.assign_card(VIEWER, cid, VIEWER.email, 1),
        'dependencies': lambda: service.set_blockers(VIEWER, cid, [], 1),
    }
    with pytest.raises(Forbidden):
        calls[action]()
    assert service.show_board(OWNER, bid) == before


@pytest.mark.parametrize('action', ['board', 'history', 'edit', 'move', 'checklist', 'blockers'])
def test_outsider_gets_no_private_board_or_card_information(extended, action):
    service, bid, cid, ai = extended
    calls = {
        'board': lambda: service.show_board(STRANGER, bid),
        'history': lambda: service.activity(STRANGER, bid),
        'edit': lambda: service.update_card(STRANGER, cid, {'title': 'Stolen'}, 1),
        'move': lambda: service.move_card(STRANGER, cid, 'Done', 1),
        'checklist': lambda: service.suggest_checklist(STRANGER, cid),
        'blockers': lambda: service.blockers(STRANGER, cid),
    }
    with pytest.raises(NotFound):
        calls[action]()
    assert service.my_boards(STRANGER) == []
    assert service.my_cards(STRANGER) == []
    assert service.overdue(STRANGER) == []
    assert ai.calls == []


@pytest.mark.parametrize('title', ['', '  ', '\n\t'])
def test_blank_card_titles_are_rejected_without_writes(extended, title):
    service, bid, _, _ = extended
    before = service.show_board(OWNER, bid)
    with pytest.raises(Invalid):
        service.add_card(OWNER, bid, title)
    assert service.show_board(OWNER, bid) == before


@pytest.mark.parametrize('role', ['owner', 'admin', '', 'VIEWER'])
def test_invalid_member_roles_are_domain_errors(extended, role):
    service, bid, _, _ = extended
    with pytest.raises(Invalid):
        service.add_member(OWNER, bid, 'new@example.test', role)


@pytest.mark.parametrize('email', ['', ' ', 'not-an-email'])
def test_invalid_membership_email_is_rejected(extended, email):
    service, bid, _, _ = extended
    with pytest.raises(Invalid):
        service.add_member(OWNER, bid, email, 'member')


def test_only_owner_can_change_roles(extended):
    service, bid, _, _ = extended
    for person in (MEMBER, VIEWER):
        with pytest.raises(Forbidden):
            service.add_member(person, bid, VIEWER.email, 'member')
    assert service.show_board(VIEWER, bid)['role'] == 'viewer'


def test_card_metadata_roundtrip_and_membership_assignment(extended):
    service, bid, cid, _ = extended
    changed = service.update_card(MEMBER, cid, {
        'title': 'Send invoice', 'description': 'Check line items',
        'labels': ['finance', 'urgent'], 'due_date': '2026-11-05',
    }, 1)
    assert changed['title'] == 'Send invoice'
    assert changed['description'] == 'Check line items'
    assert changed['labels'] == ['finance', 'urgent']
    assert changed['due_date'] == '2026-11-05'
    assigned = service.assign_card(MEMBER, cid, MEMBER.email, changed['revision'])
    assert assigned['assignee'] == MEMBER.email
    with pytest.raises(Invalid):
        service.assign_card(OWNER, cid, STRANGER.email, assigned['revision'])
    assert service.assign_card(OWNER, cid, None, assigned['revision'])['assignee'] is None


def test_invalid_update_is_all_or_nothing(extended):
    service, bid, cid, _ = extended
    before = service.show_board(OWNER, bid)
    with pytest.raises(Invalid):
        service.update_card(OWNER, cid, {'title': 'Do not save me', 'due_date': 'not-a-date'}, 1)
    assert service.show_board(OWNER, bid) == before


def test_overdue_boundary_and_done_exclusion(extended):
    service, bid, cid, _ = extended
    today = date(2026, 10, 30)
    service.today = lambda: today
    changed = service.update_card(OWNER, cid, {'due_date': today.isoformat()}, 1)
    assert service.overdue(OWNER) == []
    overdue = service.update_card(OWNER, cid, {'due_date': (today - timedelta(days=1)).isoformat()}, changed['revision'])
    assert cid in {c['id'] for c in service.overdue(OWNER)}
    service.move_card(OWNER, cid, 'Done', overdue['revision'])
    assert service.overdue(OWNER) == []


def test_my_cards_only_contains_current_person_and_visible_boards(extended):
    service, bid, cid, _ = extended
    due = (date.today() + timedelta(days=2)).isoformat()
    changed = service.update_card(OWNER, cid, {'due_date': due}, 1)
    service.assign_card(OWNER, cid, MEMBER.email, changed['revision'])
    assert cid in {c['id'] for c in service.my_cards(MEMBER)}
    assert cid not in {c['id'] for c in service.my_cards(OWNER)}
    assert service.my_cards(STRANGER) == []


def test_dependencies_refuse_self_cross_board_and_cycles(extended):
    service, bid, cid, _ = extended
    second = service.add_card(OWNER, bid, 'Second')['id']
    third = service.add_card(OWNER, bid, 'Third')['id']
    other = service.create_board(OWNER, 'Separate')['id']
    outside = service.add_card(OWNER, other, 'Outside')['id']
    with pytest.raises(Invalid):
        service.set_blockers(OWNER, cid, [cid], 1)
    with pytest.raises((Invalid, NotFound)):
        service.set_blockers(OWNER, cid, [outside], 1)
    service.set_blockers(OWNER, cid, [second], 1)
    service.set_blockers(OWNER, second, [third], 1)
    with pytest.raises(Invalid):
        service.set_blockers(OWNER, third, [cid], 1)


def test_ai_checklist_is_read_only(extended):
    service, bid, cid, ai = extended
    before = service.show_board(OWNER, bid)
    response = service.suggest_checklist(VIEWER, cid)
    assert response['suggested_checklist'] == ['Confirm scope', 'Deliver and verify']
    assert len(ai.calls) == 1
    assert service.show_board(OWNER, bid) == before


def test_web_private_reads_and_writes_are_hidden(extended):
    service, bid, cid, _ = extended
    with TestClient(create_app(service)) as client:
        for path in [f'/api/boards/{bid}', f'/api/boards/{bid}/activity']:
            assert client.get(path, headers=headers(STRANGER)).status_code == 404
            assert client.get(path).status_code == 401
        assert client.patch(f'/api/cards/{cid}', json={'title': 'No', 'expected_revision': 1},
                            headers=headers(VIEWER)).status_code == 403
        assert client.post(f'/api/boards/{bid}/members', json={'email': 'x@example.test', 'role': 'admin'},
                           headers=headers(OWNER)).status_code == 422


@pytest.mark.parametrize('raw', ['[]', 'null', '42', '"text"', '{"email": []}', '{"email": 12}', '{'])
def test_malformed_identity_never_causes_server_error(extended, raw):
    service, _, _, _ = extended
    with TestClient(create_app(service)) as client:
        response = client.get('/api/boards', headers={'X-Dodil-User': raw})
        assert response.status_code == 401


def test_html_looking_content_roundtrips_as_data(extended):
    service, bid, _, _ = extended
    payload = '<img src=x onerror="window.injected=true">'
    with TestClient(create_app(service)) as client:
        response = client.post(f'/api/boards/{bid}/cards', json={'title': payload}, headers=headers(OWNER))
        assert response.status_code == 201
        assert response.json()['title'] == payload
        assert 'application/json' in response.headers['content-type']


def test_successful_changes_have_actor_timestamp_and_before_after(extended):
    service, bid, cid, _ = extended
    baseline = len(service.activity(OWNER, bid))
    service.update_card(MEMBER, cid, {'description': 'Ready for review'}, 1)
    event = service.activity(OWNER, bid)[0]
    assert len(service.activity(OWNER, bid)) == baseline + 1
    assert event['actor'] == MEMBER.email
    assert event['card_id'] == cid
    assert event['event_at']
    assert event['changes']['description'] == {'before': '', 'after': 'Ready for review'}
    before = service.activity(OWNER, bid)
    with pytest.raises(Forbidden):
        service.move_card(VIEWER, cid, 'Done', 1)
    assert service.activity(OWNER, bid) == before


def test_storage_error_rolls_back_mutation_and_event(extended, monkeypatch):
    service, bid, cid, _ = extended
    snapshot = service.repo.snapshot()
    def fail(*args, **kwargs):
        raise RuntimeError('simulated audit storage failure')
    monkeypatch.setattr(service.repo, 'log', fail)
    with pytest.raises(RuntimeError):
        service.update_card(OWNER, cid, {'title': 'Must roll back'}, 1)
    assert service.repo.snapshot() == snapshot


def test_stale_revision_rejected_without_writing(extended):
    from boards.domain.model import Conflict
    service, bid, cid, _ = extended
    revision = service.show_board(OWNER, bid)['cards'][0]['revision']
    changed = service.update_card(OWNER, cid, {'description': 'Newer'}, expected_revision=revision)
    assert changed['revision'] == revision + 1
    snapshot = service.repo.snapshot()
    with pytest.raises(Conflict):
        service.update_card(MEMBER, cid, {'title': 'Stale'}, expected_revision=revision)
    assert service.repo.snapshot() == snapshot


def test_demotion_clears_assignment_and_logs_change(extended):
    service, bid, cid, _ = extended
    service.assign_card(OWNER, cid, MEMBER.email, 1)
    service.add_member(OWNER, bid, MEMBER.email, 'viewer')
    card = service.show_board(OWNER, bid)['cards'][0]
    assert card['assignee'] is None
    assert any(e['card_id'] == cid and e['changes'].get('assignee', {}).get('before') == MEMBER.email
               and e['changes']['assignee']['after'] is None for e in service.activity(OWNER, bid))
    with pytest.raises(Forbidden):
        service.move_card(MEMBER, cid, 'Done', 1)


def test_done_dependencies_stop_blocking(extended):
    service, bid, cid, _ = extended
    blocker = service.add_card(OWNER, bid, 'Approval')['id']
    service.set_blockers(OWNER, cid, [blocker], 1)
    assert service.blockers(OWNER, cid)['is_blocked']
    service.move_card(OWNER, blocker, 'Done', 1)
    assert service.blockers(OWNER, cid)['blockers'] == []
    assert not service.blockers(OWNER, cid)['is_blocked']


def test_summary_and_semantic_search_never_include_another_board(extended):
    service, bid, cid, _ = extended
    class ScopedAI:
        summary_payload = None
        texts = None
        def summary(self, payload):
            self.summary_payload = payload
            return 'A concise summary.'
        def embeddings(self, texts):
            self.texts = texts
            return [[1.0, 0.0]] * len(texts)
    ai = ScopedAI()
    service.assistant = ai
    service.add_card(OWNER, bid, 'Public to members')
    hidden_bid = service.create_board(OWNER, 'Private finance')['id']
    hidden = service.add_card(OWNER, hidden_bid, 'SECRET ACQUISITION')
    service.board_summary(MEMBER, bid)
    assert 'SECRET ACQUISITION' not in json.dumps(ai.summary_payload)
    result = service.similar_cards(MEMBER, cid)
    assert hidden['id'] not in {m['card']['id'] for m in result['matches']}
    assert 'SECRET ACQUISITION' not in '\n'.join(ai.texts)
    for action in [lambda: service.board_summary(MEMBER, hidden_bid),
                   lambda: service.similar_cards(MEMBER, hidden['id'])]:
        with pytest.raises(NotFound):
            action()


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


def test_mcp_visibility_and_edit_controls_match_web(extended):
    service, bid, cid, _ = extended
    with TestClient(create_app(service)) as client:
        assert mcp_call(client, STRANGER, 'show_board', {'board_id': bid}).get('isError')
        assert mcp_call(client, VIEWER, 'move_card', {'card_id': cid, 'list_name': 'Done', 'expected_revision': 1}).get('isError')
        assert mcp_call(client, MEMBER, 'add_member', {'board_id': bid, 'email': 'new@example.test'}).get('isError')
        assert not mcp_call(client, MEMBER, 'move_card', {'card_id': cid, 'list_name': 'Doing', 'expected_revision': 1}).get('isError')
    assert service.show_board(OWNER, bid)['cards'][0]['list_name'] == 'Doing'


def test_two_simultaneous_stale_edits_have_exactly_one_winner(extended):
    from concurrent.futures import ThreadPoolExecutor
    from boards.domain.model import Conflict
    service, bid, cid, _ = extended
    revision = service.show_board(OWNER, bid)['cards'][0]['revision']
    def edit(title):
        try:
            service.update_card(MEMBER, cid, {'title': title}, expected_revision=revision)
            return 'saved'
        except Conflict:
            return 'conflict'
    with ThreadPoolExecutor(max_workers=2) as pool:
        results = list(pool.map(edit, ['First editor', 'Second editor']))
    assert sorted(results) == ['conflict', 'saved']


def test_idempotent_move_has_no_duplicate_history_event(extended):
    service, bid, cid, _ = extended
    moved = service.move_card(OWNER, cid, 'Doing', 1)
    before = service.repo.snapshot()
    service.move_card(OWNER, cid, 'Doing', moved['revision'])
    assert service.repo.snapshot() == before


@pytest.mark.parametrize('changes', [
    {'owner_email': 'attacker@example.test'}, {'board_id': 'other'},
    {'labels': 'urgent'}, {'labels': ['']}, {'title': None},
    {'title': 'x' * 181}, {'description': 'x' * 6001},
])
def test_invalid_fields_are_rejected_before_write(extended, changes):
    service, _, cid, _ = extended
    before = service.repo.snapshot()
    with pytest.raises(Invalid):
        service.update_card(OWNER, cid, changes, 1)
    assert service.repo.snapshot() == before


def test_browser_security_headers_and_cross_origin_refusal(extended):
    service, _, _, _ = extended
    with TestClient(create_app(service)) as client:
        response = client.get('/api/boards', headers=headers(OWNER))
        assert response.headers['cache-control'] == 'no-store'
        assert response.headers['x-content-type-options'] == 'nosniff'
        assert "frame-ancestors 'none'" in response.headers['content-security-policy']
        response = client.post('/api/boards', json={'title': 'Cross-origin'},
                               headers={**headers(OWNER), 'Origin': 'https://evil.example'})
        assert response.status_code == 403
        assert len(service.my_boards(OWNER)) == 1


def test_web_does_not_accept_actor_or_owner_from_body(extended):
    service, bid, _, _ = extended
    with TestClient(create_app(service)) as client:
        response = client.post(f'/api/boards/{bid}/cards', headers=headers(MEMBER),
                               json={'title': 'Spoof', 'actor': OWNER.email})
        assert response.status_code == 422
        response = client.post('/api/boards', headers=headers(MEMBER),
                               json={'title': 'Spoof', 'owner_email': OWNER.email})
        assert response.status_code == 422


def test_anonymous_health_and_metadata_do_not_expose_data(extended):
    service, _, _, _ = extended
    with TestClient(create_app(service)) as client:
        assert client.get('/healthz').json() == {'ok': True}
        metadata = client.get('/.well-known/oauth-protected-resource/mcp')
        assert metadata.status_code == 200
        assert 'Launch' not in metadata.text
        assert client.get('/docs').status_code == 404
        assert client.get('/openapi.json').status_code == 404
        assert client.get('/static/app.js').status_code == 401