"""Firefox UI checks against the disposable real servers started by flow.py.""" import json import socket import subprocess import time import urllib.parse def check_browser_flow(stand, fixtures): profile = stand.directory / 'firefox' profile.mkdir() downloads = stand.directory / 'downloads' downloads.mkdir() with socket.socket() as reservation: reservation.bind(('127.0.0.1', 0)) port = reservation.getsockname()[1] (profile / 'user.js').write_text('\n'.join([ f'user_pref("marionette.port", {port});', 'user_pref("browser.shell.checkDefaultBrowser", false);', 'user_pref("browser.tabs.warnOnClose", false);', 'user_pref("browser.download.folderList", 2);', f'user_pref("browser.download.dir", {json.dumps(str(downloads))});', 'user_pref("browser.download.useDownloadDir", true);', 'user_pref("browser.helperApps.neverAsk.saveToDisk", "application/zip,application/octet-stream");', 'user_pref("browser.download.always_ask_before_handling_new_types", false);', ])) sock = None with (stand.directory / 'firefox.log').open('w') as log: firefox = subprocess.Popen(['firefox', '--headless', '--no-remote', '--profile', str(profile), '--marionette', '--remote-allow-system-access', 'about:blank'], stdout=log, stderr=log) try: deadline = time.monotonic() + 15 while sock is None and time.monotonic() < deadline: try: sock = socket.create_connection(('127.0.0.1', port), timeout=1) except OSError: time.sleep(.1) assert sock is not None, 'Firefox Marionette did not start' sock.settimeout(30) def read_message(): size = b'' while not size.endswith(b':'): part = sock.recv(1) assert part, 'Firefox disconnected' size += part remaining = int(size[:-1]) data = b'' while remaining: part = sock.recv(remaining) assert part, 'Firefox disconnected' data += part remaining -= len(part) return json.loads(data) read_message() command_id = 0 def command(name, args=None): nonlocal command_id command_id += 1 payload = json.dumps([0, command_id, name, args or {}]).encode() sock.sendall(str(len(payload)).encode() + b':' + payload) result = read_message() if result[2]: raise RuntimeError('Firefox command failed: ' + name) return result[3] def js(script, *args): result = command('WebDriver:ExecuteScript', {'script': script, 'args': list(args), 'newSandbox': False, 'sandbox': None}) return result.get('value') if isinstance(result, dict) and 'value' in result else result def wait(script, description, seconds=15): until = time.monotonic() + seconds while time.monotonic() < until: if js('return ' + script): return time.sleep(.1) raise AssertionError('Browser timed out: ' + description) def navigate(path): command('WebDriver:Navigate', {'url': stand.public + path}) def click(label): assert js("const button=[...document.querySelectorAll('button')].find(b=>b.textContent.trim()===arguments[0] || b.getAttribute('aria-label')===arguments[0]);if(!button || button.disabled)return false;button.click();return true;", label), 'Missing or disabled button: ' + label def input_value(selector, value): assert js("const input=document.querySelector(arguments[0]);if(!input)return false;input.value=arguments[1];input.dispatchEvent(new Event('input',{bubbles:true}));input.dispatchEvent(new Event('change',{bubbles:true}));return true;", selector, value), 'Missing input: ' + selector def login(mode): stand.current = mode navigate('/login') js('localStorage.clear();return true;') navigate('/login') wait("document.querySelector('input[type=password]')!==null", 'login form') input_value('input[placeholder="detective@mail.ru"]', 'organizer@test.invalid') input_value('input[type=password]', 'test-only-password') click('Вход') wait("location.pathname==='/office'", 'login') def check_presence(data): from flow import run, wait_for api = stand.apis['venue'] device_id = js("return localStorage.getItem('evening-detective-device-ID')") assert device_id def connections(): devices = api.request('/api/devices')['devices'] return next((d['connections'] for d in devices if d['ID'] == device_id), []) def has_kind(kind): return any(c['kind'] == kind for c in connections()) wait_for(lambda: has_kind('organizer'), 'organizer presence') organizer_window = command('WebDriver:GetWindowHandle') if isinstance(organizer_window, dict): organizer_window = organizer_window['value'] team_window = command('WebDriver:NewWindow', {'type': 'tab'})['handle'] command('WebDriver:SwitchToWindow', {'handle': team_window}) captain = data['teams'][0] navigate(f'/team-story/{captain["ID"]}?password={urllib.parse.quote(captain["password"])}') wait_for(lambda: has_kind('team'), 'player presence') screen_window = command('WebDriver:NewWindow', {'type': 'tab'})['handle'] command('WebDriver:SwitchToWindow', {'handle': screen_window}) navigate('/screen') wait_for(lambda: has_kind('screen'), 'screen presence') screen_connection = next(c for c in connections() if c['kind'] == 'screen') # Bind the actual browser session, whose credentials stay inside the browser. saved = js("return JSON.parse(sessionStorage.getItem('evening-detective-screen-session'))") assert saved['screenID'] == str(screen_connection['screenID']) api.request(f'/api/screens/{saved["screenID"]}/bind', {**saved, 'gameID': data['game']}) js("window.dispatchEvent(new Event('online'));return true;") wait_for(lambda: any(c['ID'] == screen_connection['ID'] and c.get('gameID') == data['game'] for c in connections()), 'bound screen presence') assert {c['kind'] for c in connections()} >= {'organizer', 'team', 'screen'} command('WebDriver:CloseWindow') command('WebDriver:SwitchToWindow', {'handle': team_window}) command('WebDriver:CloseWindow') command('WebDriver:SwitchToWindow', {'handle': organizer_window}) old_id = next(c['ID'] for c in connections() if c['kind'] == 'organizer') click('Главная') wait("location.pathname==='/'", 'SPA navigation') wait_for(lambda: any(c['ID'] == old_id and c['kind'] == 'browser' for c in connections()), 'SPA clears context on same connection') # Drop only heartbeat requests; gameplay and authorization remain live. js("window.presenceFetch=window.fetch;window.fetch=(url,options)=>String(url).includes('/devices/heartbeat')?Promise.reject(new Error('test offline')):window.presenceFetch(url,options);return true;") before = next(c['lastSeenAt'] for c in connections() if c['ID'] == old_id) time.sleep(22) assert next(c['lastSeenAt'] for c in connections() if c['ID'] == old_id) == before js("window.fetch=window.presenceFetch;window.dispatchEvent(new Event('online'));return true;") wait_for(lambda: next(c['lastSeenAt'] for c in connections() if c['ID'] == old_id) != before, 'network recovery') run('docker', 'stop', stand.name + '-venue-app') run('docker', 'start', stand.name + '-venue-app') wait_for(lambda: api.request('/api/test/ping') == {}, 'browser backend restart') wait_for(lambda: any(c['ID'] == old_id for c in connections()), 'automatic interval re-registration', 30) navigate(f'/games/{data["game"]}') wait_for(lambda: has_kind('organizer'), 'reloaded organizer presence') new_id = next(c['ID'] for c in connections() if c['kind'] == 'organizer') assert new_id != old_id and js("return localStorage.getItem('evening-detective-device-ID')") == device_id print('PASS: stage 12 real Firefox organizer/player/screen, grouped tabs, SPA context clearing, reload UUID, stopped heartbeat/network recovery and interval re-registration after restart', flush=True) command('WebDriver:NewSession', {'capabilities': {'alwaysMatch': {'acceptInsecureCerts': True}}}) for width, data in zip((1440, 390), fixtures, strict=True): game, captain = data['game'], data['teams'][0] command('WebDriver:SetWindowRect', {'width': width, 'height': 900}) login('prod') navigate(f'/games/{game}?tab=settings') wait("document.body.innerText.includes('Подготовить пакет')", 'export button') click('Подготовить пакет') wait("document.body.innerText.includes('Скачать пакет')", 'export ready', 60) previous_downloads = {p.name for p in downloads.glob('*.zip')} click('Скачать пакет') until = time.monotonic() + 15 package = None while time.monotonic() < until: candidates = [p for p in downloads.glob('*.zip') if p.name not in previous_downloads and p.stat().st_size and not p.with_suffix(p.suffix + '.part').exists()] if candidates: package = max(candidates, key=lambda p: p.stat().st_mtime) break time.sleep(.1) assert package is not None, 'ZIP download missing' raw = package.read_bytes() assert raw.startswith(b'PK'), 'Downloaded file is not ZIP' login('venue') navigate('/games') wait("document.body.innerText.includes('Импорт игры')", 'import button') click('Импорт игры') js("const dt=new DataTransfer();dt.items.add(new File([new Uint8Array(arguments[0])],arguments[1],{type:'application/zip'}));const input=document.querySelector('input[type=file]');input.files=dt.files;input.dispatchEvent(new Event('change',{bubbles:true}));return true;", list(raw), package.name) click('Загрузить пакет') wait("document.body.innerText.includes('Игра импортирована')", 'real ZIP import') assert js("const r=document.querySelector('[role=dialog]').getBoundingClientRect();return r.left>=0 && r.right<=innerWidth"), 'Import modal overflows viewport' command('WebDriver:PerformActions', {'actions': [{'type': 'key', 'id': 'keyboard', 'actions': [{'type': 'keyDown', 'value': '\ue004'}, {'type': 'keyUp', 'value': '\ue004'}]}]}) assert js("return document.querySelector('[role=dialog]').contains(document.activeElement)"), 'Modal lost keyboard focus' click('Открыть игру') wait(f"location.pathname==='/games/{game}'", 'open imported game') if width == 1440: check_presence(data) navigate(f'/games/{game}?tab=payment') wait("document.body.innerText.includes('Проверить билет')", 'ticket checking') click('Проверить билет') input_value('input[placeholder="Или вставьте ссылку на билет"]', data['tickets'][1]['URL']) assert js("const button=[...document.querySelectorAll('[role=dialog] button')].find(b=>b.textContent.trim()==='Проверить билет');button.click();return true;") wait("document.body.innerText.includes('Принять билет')", 'active ticket') click('Принять билет') wait("document.body.innerText.includes('уже использован')", 'accepted ticket') navigate(urllib.parse.urlparse(data['tickets'][1]['URL']).path) wait("document.body.innerText.includes('Билет уже использован')", 'original ticket page') navigate(f'/games/{game}?tab=game') wait("document.querySelector('[aria-label=\"Начать игру\"]')!==null", 'start control') click('Начать игру') wait("document.querySelector('[aria-label=\"Поставить игру на паузу\"]')!==null", 'running game') team_path = f'/team-story/{captain["ID"]}?' + urllib.parse.urlencode({'password': captain['password']}) navigate(team_path) wait("document.querySelector('#run')!==null && !document.querySelector('#run').disabled", 'team input') stand.delay_actions = .4 input_value('#run', 'p1') click('Поехали') assert js("return document.querySelector('.game-button-run').disabled"), 'Move submission is not blocked' js("document.querySelector('.game-button-run').click();return true;") wait("document.body.innerText.includes('TEST Place p1')", 'first move') assert len(stand.apis['venue'].story(captain)['actions']) == 1, 'Double submission created duplicate moves' stand.sql('venue', f"UPDATE teams SET password='revoked' WHERE id={int(captain['ID'])};") input_value('#run', 'p2') click('Поехали') wait("document.body.innerText.includes('команда не найдена')", 'refused move error') assert js("return document.querySelector('#run').value==='p2'"), 'Refused move erased the input' escaped_password = captain['password'].replace("'", "''") stand.sql('venue', f"UPDATE teams SET password='{escaped_password}' WHERE id={int(captain['ID'])};") stand.delay_actions = 0 stand.apis['venue'].request(f'/api/games/{game}/pause', {}, 'PUT') wait("document.querySelector('#run').disabled", 'pause polling') assert js("return document.querySelector('.game-button-run').disabled"), 'Prefilled move is still enabled during pause' stand.apis['venue'].request(f'/api/games/{game}/play', {}, 'PUT') wait("!document.querySelector('.game-button-run').disabled", 'resume polling') click('Поехали') wait("document.body.innerText.includes('TEST Place p2')", 'second move') if data['format'] == 'online': input_value('.questions-block textarea', 'Butler') click('Сохранить ответы') wait("document.body.innerText.includes('Ответы сохранены')", 'answer submission') else: assert js("return document.querySelector('.questions-block')===null"), 'Offline game exposes digital answers' navigate(f'/games/{game}?tab=teams') wait("document.body.innerText.includes('Выдать: TEST Clue p1')", 'pending clue') click('Выдать: TEST Clue p1') wait("document.querySelector('[role=dialog]')!==null", 'clue dialog') click('Выдать') navigate(f'/games/{game}?tab=game') wait("document.querySelector('[aria-label=\"Завершить игру\"]')!==null", 'finish control') click('Завершить игру') navigate(f'/games/{game}?tab=answers') wait("document.querySelector('.answers-score-input input')!==null", 'grading form') input_value('.answers-score-input input', '5') click('Сохранить баллы') wait("document.body.innerText.includes('Баллы сохранены')", 'saved scores') navigate(f'/games/{game}/results') wait("document.body.innerText.includes('TEST Captain')", 'results') navigate(team_path) wait("document.body.innerText.includes('Игра завершена')", 'finished player view') assert js("return document.querySelector('.questions-block')===null"), 'Finished game exposes questions' navigate(f'/team-story/{captain["ID"]}?password=wrong') wait("document.body.innerText.includes('Не удалось загрузить дело')", 'wrong-password error') def download_results(): navigate(f'/games/{game}?tab=settings') wait("document.body.innerText.includes('Скачать результаты')", 'results download button') before = {p.name for p in downloads.glob('*.zip')} click('Скачать результаты') until = time.monotonic() + 15 while time.monotonic() < until: files = [p for p in downloads.glob('*.zip') if p.name not in before and p.stat().st_size and not p.with_suffix(p.suffix + '.part').exists()] if files: return max(files, key=lambda p: p.stat().st_mtime).read_bytes() time.sleep(.1) raise AssertionError('Results download missing') def apply_results(raw): navigate(f'/games/{game}?tab=settings') wait("document.body.innerText.includes('Возврат результатов с площадки')", 'return block') assert js("return document.body.innerText.includes('Онлайн-правки этих данных потеряются') && document.body.innerText.includes('Оформление и музыка сохраняются')"), 'Replacement warning missing' js("const dt=new DataTransfer();dt.items.add(new File([new Uint8Array(arguments[0])],'results.zip',{type:'application/zip'}));const input=document.querySelector('input[type=file]');input.files=dt.files;input.dispatchEvent(new Event('change',{bubbles:true}));return true;", list(raw)) stand.delay_results = .5 click('Применить ZIP и перезаписать результаты') assert js("return [...document.querySelectorAll('button')].find(b=>b.textContent.trim()==='Применить ZIP и перезаписать результаты').disabled"), 'Repeat apply is not blocked' wait("document.body.innerText.includes('Результаты приняты')", 'results accepted') stand.delay_results = 0 wait("!document.body.innerText.includes('Игра в офлайн-режиме')", 'offline mark cleared') assert stand.apis['prod'].game(game)['status'] == 'finish' first_results = download_results() login('prod') stand.sql('prod', f"UPDATE games SET name='TEST Preserved presentation', description='TEST Preserved description' WHERE id={game};") apply_results(first_results) assert stand.apis['prod'].game(game)['name'] == 'TEST Preserved presentation' assert stand.apis['prod'].story(captain)['game']['status'] == 'finish' assert stand.apis['prod'].request('/api' + urllib.parse.urlparse(data['tickets'][1]['URL']).path)['ticket']['status'] == 'used' assert stand.apis['prod'].request(f'/api/games/{game}/results')['teams'][0]['totalScore'] == 5 login('venue') navigate(f'/games/{game}?tab=answers') wait("document.querySelector('.answers-score-input input')!==null", 'corrected grading') input_value('.answers-score-input input', '0') click('Сохранить баллы') wait("document.body.innerText.includes('Баллы сохранены')", 'correction saved') corrected_results = download_results() assert corrected_results != first_results, 'Results are not fresh' login('prod') stand.sql('prod', f"UPDATE team_answers SET answer='Online edit', score=9 WHERE team_id={captain['ID']};") apply_results(corrected_results) scores = stand.apis['prod'].request(f'/api/games/{game}/results')['teams'] assert next(t for t in scores if t['team']['ID'] == captain['ID'])['totalScore'] == 0 accepted = stand.sql('prod', f"SELECT applied_at FROM game_returns WHERE source_game_id={game} ORDER BY applied_at DESC LIMIT 1;") stand.restart('prod') apply_results(first_results) assert stand.sql('prod', f"SELECT applied_at FROM game_returns WHERE source_game_id={game} ORDER BY applied_at DESC LIMIT 1;") == accepted assert next(t for t in stand.apis['prod'].request(f'/api/games/{game}/results')['teams'] if t['team']['ID'] == captain['ID'])['totalScore'] == 0 print(f'PASS: real Firefox {width}px {data["format"]}: export/download/import, tickets, moves, pause, clues, grading/results, result ZIP return, correction, online overwrite, preserved presentation, restart repeat, errors and keyboard', flush=True) command('Marionette:Quit', {'flags': ['eForceQuit']}) finally: if sock: sock.close() firefox.terminate() try: firefox.wait(timeout=5) except subprocess.TimeoutExpired: firefox.kill() firefox.wait(timeout=5)