105 lines
5.1 KiB
Python
105 lines
5.1 KiB
Python
import json
|
|
import time
|
|
|
|
import pytest
|
|
from websockets.sync.client import connect
|
|
|
|
from test_ready_server import server_url, send, receive
|
|
|
|
|
|
def join_pair(host, guest):
|
|
file_info = {'name': 'fight.mkv', 'size': 1234, 'duration': 60}
|
|
send(host, type='create_room', username='Host', fileInfo=file_info)
|
|
code = receive(host, 'room_created')['code']
|
|
send(guest, type='join_room', username='Guest', code=code)
|
|
receive(guest, 'room_file_check')
|
|
send(guest, type='confirm_join', fileInfo=file_info)
|
|
receive(guest, 'room_joined')
|
|
receive(host, 'user_joined')
|
|
return code, file_info
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['ko', 'bell', 'tap', 'robbery'])
|
|
def test_reactions_are_room_wide_and_preserve_playback(server_url, kind):
|
|
with connect(server_url) as host, connect(server_url) as guest:
|
|
code, file_info = join_pair(host, guest)
|
|
# An unconfirmed client cannot broadcast or consume the cooldown.
|
|
with connect(server_url) as pending:
|
|
send(pending, type='join_room', username='Pending', code=code)
|
|
receive(pending, 'room_file_check')
|
|
send(pending, type='reaction', kind=kind)
|
|
send(pending, type='ping', timestamp=1)
|
|
assert receive(pending, 'pong')['timestamp'] == 1
|
|
send(host, type='reaction', kind='invalid')
|
|
assert 'Unknown' in receive(host, 'reaction_error')['message']
|
|
send(host, type='sync', action='play', position=18)
|
|
receive(guest, 'sync')
|
|
send(host, type='reaction', kind=kind, username='Spoofed')
|
|
reaction = receive(host, 'reaction')
|
|
assert reaction['kind'] == kind and reaction['username'] == 'Host'
|
|
assert reaction['position'] >= 18
|
|
assert receive(guest, 'reaction') == reaction
|
|
assert receive(host, 'chat')['message'] == '🥊 /' + kind
|
|
receive(guest, 'chat')
|
|
if kind == 'robbery':
|
|
assert receive(host, 'fight_poll')['active']
|
|
receive(guest, 'fight_poll')
|
|
send(guest, type='reaction', kind='bell')
|
|
assert 'cooldown' in receive(guest, 'reaction_error')['message']
|
|
send(host, type='request_state')
|
|
state = receive(host, 'state')['state']
|
|
assert state['playing'] and state['position'] >= 18
|
|
with pytest.raises(TimeoutError):
|
|
host.recv(timeout=.1)
|
|
# Reconnecting restores chat history without replaying transient effects.
|
|
guest.close()
|
|
with connect(server_url) as restored:
|
|
send(restored, type='rejoin_room', username='Guest', code=code)
|
|
snapshot = receive(restored, 'room_rejoined')
|
|
assert snapshot['chatHistory'][-1]['message'] == '🥊 /' + kind
|
|
with pytest.raises(TimeoutError):
|
|
restored.recv(timeout=.1)
|
|
|
|
|
|
def test_robbery_votes_are_unique_restored_and_expire(server_url):
|
|
with connect(server_url) as host, connect(server_url) as guest:
|
|
code, file_info = join_pair(host, guest)
|
|
send(host, type='reaction', kind='robbery')
|
|
poll = receive(host, 'fight_poll')
|
|
poll_id = poll['id']
|
|
assert poll['votes'] == {} and poll['active']
|
|
receive(guest, 'fight_poll')
|
|
send(host, type='fight_poll_vote', id=poll_id, choice='yes')
|
|
assert receive(host, 'fight_poll')['votes'] == {'Host': 'yes'}
|
|
receive(guest, 'fight_poll')
|
|
send(host, type='fight_poll_vote', id=poll_id, choice='no')
|
|
assert receive(host, 'fight_poll')['votes'] == {'Host': 'no'}
|
|
receive(guest, 'fight_poll')
|
|
send(guest, type='fight_poll_vote', id=poll_id, choice='yes', username='Host')
|
|
assert receive(host, 'fight_poll')['votes'] == {'Host': 'no', 'Guest': 'yes'}
|
|
receive(guest, 'fight_poll')
|
|
for vote_id, choice in [('stale', 'yes'), (poll_id, 'invalid')]:
|
|
send(guest, type='fight_poll_vote', id=vote_id, choice=choice)
|
|
send(guest, type='ping', timestamp=2)
|
|
assert json.loads(guest.recv(timeout=3))['type'] == 'pong'
|
|
guest.close()
|
|
with connect(server_url) as restored:
|
|
send(restored, type='rejoin_room', username='Guest', code=code)
|
|
snapshot = receive(restored, 'room_rejoined')['fightPoll']
|
|
assert snapshot['id'] == poll_id and snapshot['votes']['Guest'] == 'yes'
|
|
receive(host, 'user_rejoined')
|
|
time.sleep(3.1)
|
|
send(restored, type='reaction', kind='robbery')
|
|
assert 'already open' in receive(restored, 'reaction_error')['message']
|
|
assert receive(restored, 'fight_poll')['id'] == poll_id
|
|
closed = json.loads(host.recv(timeout=35))
|
|
assert closed['type'] == 'fight_poll' and not closed['active']
|
|
assert closed['votes'] == {'Host': 'no', 'Guest': 'yes'}
|
|
assert not receive(restored, 'fight_poll')['active']
|
|
send(restored, type='fight_poll_vote', id=poll_id, choice='no')
|
|
send(restored, type='ping', timestamp=3)
|
|
assert json.loads(restored.recv(timeout=3))['type'] == 'pong'
|
|
send(host, type='reaction', kind='robbery')
|
|
replacement = receive(host, 'fight_poll')
|
|
assert replacement['id'] != poll_id and replacement['votes'] == {}
|