61 lines
3.9 KiB
Python
61 lines
3.9 KiB
Python
"""Stage 12 API checks in the disposable real venue."""
|
|
import json
|
|
import uuid
|
|
|
|
from flow import API, run, wait_for
|
|
|
|
|
|
def check_devices(stand, data):
|
|
venue, prod = stand.apis['venue'], stand.apis['prod']
|
|
public = API(venue.base)
|
|
request = {'deviceID': str(uuid.uuid4()), 'connectionID': str(uuid.uuid4())}
|
|
disabled = API(prod.base).request('/api/devices/heartbeat', {})
|
|
assert disabled['enabled'] is False and disabled['serverTime']
|
|
prod.request('/api/devices', expected=400)
|
|
public.request('/api/devices', expected=401)
|
|
assert public.request('/api/devices/heartbeat', request)['enabled'] is True
|
|
public.request('/api/devices/heartbeat', request)
|
|
devices = venue.request('/api/devices')['devices']
|
|
assert len(devices) == 1 and len(devices[0]['connections']) == 1
|
|
original = devices[0]['connections'][0]
|
|
public.request('/api/devices/heartbeat', {**request, 'organizerGameID': data['game']}, expected=401)
|
|
public.request('/api/devices/heartbeat', {**request, 'deviceID': 'bad'}, expected=400)
|
|
for fields in ({'teamID': '', 'password': 'secret'}, {'password': 'secret'}, {'screenID': '3'},
|
|
{'organizerGameID': data['game'], 'teamID': '019a1234-5678-7abc-8def-000000000001', 'password': 'secret'},
|
|
{'screenID': 'not-an-ID', 'accessToken': 'secret'}):
|
|
public.request('/api/devices/heartbeat', {**request, **fields}, expected=400)
|
|
team = data['teams'][0]
|
|
public.request('/api/devices/heartbeat', {**request, 'teamID': team['ID'], 'password': 'invalid'}, expected=404)
|
|
cancelled = data['teams'][2]
|
|
public.request('/api/devices/heartbeat', {**request, 'teamID': cancelled['ID'], 'password': cancelled['password']}, expected=404)
|
|
public.request('/api/devices/heartbeat', {**request, 'screenID': '2147483647', 'accessToken': 'invalid'}, expected=401)
|
|
assert venue.request('/api/devices')['devices'][0]['connections'][0] == original
|
|
public.request('/api/devices/heartbeat', {**request, 'teamID': team['ID'], 'password': team['password']})
|
|
tab = {**request, 'connectionID': str(uuid.uuid4()), 'organizerGameID': data['game']}
|
|
venue.request('/api/devices/heartbeat', tab)
|
|
screen = public.request('/api/screens', {})
|
|
venue.request(f'/api/screens/{screen["screenID"]}/bind', {**screen, 'gameID': data['game']})
|
|
screen_tab = {**request, 'connectionID': str(uuid.uuid4()), **screen}
|
|
public.request('/api/devices/heartbeat', screen_tab)
|
|
devices = venue.request('/api/devices?gameID=' + str(data['game']))['devices']
|
|
assert len(devices) == 1 and len(devices[0]['connections']) == 3
|
|
assert {c['kind'] for c in devices[0]['connections']} == {'organizer', 'team', 'screen'}
|
|
serialized = json.dumps(devices)
|
|
for secret in (team['password'], screen['accessToken'], 'password', 'accessToken', 'email', 'URL'):
|
|
assert secret not in serialized
|
|
public.request('/api/devices/heartbeat', request)
|
|
assert len(venue.request('/api/devices?gameID=' + str(data['game']))['devices'][0]['connections']) == 2
|
|
browser = next(c for c in venue.request('/api/devices')['devices'][0]['connections'] if c['ID'] == request['connectionID'])
|
|
assert browser['kind'] == 'browser' and not any(browser.get(key) for key in ('gameID', 'teamID', 'screenID'))
|
|
# The stand uses the same test-only JWT signing key for both servers.
|
|
reader = API(venue.base)
|
|
reader.token = prod.token
|
|
assert reader.request('/api/devices')['devices']
|
|
run('docker', 'stop', stand.name + '-venue-app')
|
|
run('docker', 'start', stand.name + '-venue-app')
|
|
wait_for(lambda: public.request('/api/test/ping') == {}, 'presence backend restart')
|
|
assert not venue.request('/api/devices')['devices']
|
|
public.request('/api/devices/heartbeat', request)
|
|
assert len(venue.request('/api/devices')['devices']) == 1
|
|
print('PASS: stage 12 public API, UUID/context validation, credentials/cancellation, tabs/filter/clearing, safe output, prod/venue and backend restart', flush=True)
|