"""Real prod -> ZIP -> venue lifecycle in disposable Docker resources, without WAN.""" import argparse import base64 from concurrent.futures import ThreadPoolExecutor import datetime from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json import hashlib import mimetypes from pathlib import Path import secrets import subprocess import sys import tempfile import threading import time import urllib.error import urllib.parse import urllib.request import uuid import zipfile sys.dont_write_bytecode = True ROOT = Path(__file__).resolve().parents[2] OPENER = urllib.request.build_opener(urllib.request.ProxyHandler({})) MATERIAL = b'%PDF-1.4\n% ED-029 stage 8 test clue\n%%EOF\n' def run(*args, input=None, cwd=None): result = subprocess.run(args, input=input, cwd=cwd, text=True, capture_output=True, timeout=180) if result.returncode: raise RuntimeError(f'{args[0]} {args[1]} failed: {result.stdout[-2000:]}\n{result.stderr[-2000:]}') return result.stdout.strip() def wait_for(check, description, seconds=60): deadline = time.monotonic() + seconds while time.monotonic() < deadline: try: if check(): return except (OSError, urllib.error.URLError, RuntimeError): pass time.sleep(.2) raise AssertionError('Timed out: ' + description) class API: def __init__(self, base): self.base = base self.token = '' def request(self, path, body=None, method=None, raw=False, expected=200): headers = {} if self.token: headers['Authorization'] = 'Bearer ' + self.token if isinstance(body, bytes): headers['Content-Type'] = 'application/zip' data = body else: headers['Content-Type'] = 'application/json' data = None if body is None else json.dumps(body).encode() request = urllib.request.Request(self.base + path, data=data, headers=headers, method=method) try: response = OPENER.open(request, timeout=1800 if "/results-package/" in request.full_url or request.full_url.endswith("/api/games/import") else 30) except urllib.error.HTTPError as error: response = error with response: payload = response.read() assert response.status in (expected if isinstance(expected, tuple) else (expected,)), f'{method or request.get_method()} API: HTTP {response.status}, expected {expected}' if raw: return payload result = json.loads(payload) if response.status != 200: assert result.get('code') or result.get('error'), 'Missing API error' return result def game(self, id): return self.request(f'/api/games/{id}')['game'] def story(self, team, expected=200): query = urllib.parse.urlencode({'password': team['password']}) return self.request(f'/api/teams/{team["ID"]}/story?{query}', expected=expected) def move(self, team, code, expected=200): return self.request(f'/api/teams/{team["ID"]}/actions', {'password': team['password'], 'code': code}, expected=expected) class Stand: def __init__(self, directory): self.directory = Path(directory) self.name = 'ed029-flow-' + uuid.uuid4().hex[:10] self.containers = [] self.network_created = False self.current = 'prod' self.delay_actions = 0 self.delay_results = 0 self.apis = {} stand = self class Handler(BaseHTTPRequestHandler): def log_message(self, *args): pass def do_GET(self): self.handle_request() def do_POST(self): self.handle_request() def do_PUT(self): self.handle_request() def do_DELETE(self): self.handle_request() def handle_request(self): if self.path.startswith('/api/'): if self.command == 'POST' and self.path.endswith('/results-package/apply'): time.sleep(stand.delay_results) if self.command == 'POST' and self.path.endswith('/actions'): time.sleep(stand.delay_actions) headers = {key: value for key, value in self.headers.items() if key.lower() not in ('host', 'connection', 'content-length')} length = int(self.headers.get('Content-Length', 0)) limit = 1024 ** 3 if '/results-package/' in self.path or self.path == '/api/games/import' else 257 * 1024 ** 2 if length < 0 or length > limit: self.send_error(413) return self.connection.settimeout(1800) request = urllib.request.Request(stand.apis[stand.current].base + self.path, data=self.rfile.read(length) if length else None, headers=headers, method=self.command) try: response = OPENER.open(request, timeout=1800 if "/results-package/" in request.full_url or request.full_url.endswith("/api/games/import") else 30) except urllib.error.HTTPError as error: response = error with response: body = response.read() self.send_response(response.status) for key, value in response.headers.items(): if key.lower() not in ('connection', 'transfer-encoding', 'content-length'): self.send_header(key, value) else: dist = stand.directory / 'frontend' filename = (dist / urllib.parse.urlparse(self.path).path.lstrip('/')).resolve() if not filename.is_relative_to(dist.resolve()): self.send_error(404) return if not filename.is_file(): filename = dist / 'index.html' body = filename.read_bytes() self.send_response(200) self.send_header('Content-Type', mimetypes.guess_type(filename)[0] or 'application/octet-stream') self.send_header('Content-Length', str(len(body))) self.end_headers() try: self.wfile.write(body) except (BrokenPipeError, ConnectionResetError): # Navigation may cancel an in-flight browser polling response. pass self.server = ThreadingHTTPServer(('127.0.0.1', 0), Handler) self.public = f'http://127.0.0.1:{self.server.server_port}' self.server_thread = threading.Thread(target=self.server.serve_forever, daemon=True) def container(self, suffix, *args): name = self.name + '-' + suffix run('docker', 'run', '-d', '--name', name, '--pull=never', '--network', self.name, '--label', 'ed029-flow=' + self.name, *args) self.containers.append(name) return name def sql(self, mode, sql): return run('docker', 'exec', '-i', self.name + '-' + mode + '-pg', 'psql', '-U', 'test', '-d', 'test', '-v', 'ON_ERROR_STOP=1', '-At', input=sql) def start(self): print('Building current backend/frontend in temporary directory', flush=True) run('env', 'CGO_ENABLED=0', 'go', 'build', '-o', str(self.directory / 'server'), './cmd/evening_detective_server', cwd=ROOT / 'evening_detective_server') run('env', 'VITE_API_URL=/.', 'npm', 'run', 'build-only', '--', '--outDir', str(self.directory / 'frontend'), cwd=ROOT / 'evening_detective_frontend') run('docker', 'network', 'create', '--internal', self.name) self.network_created = True assert run('docker', 'network', 'inspect', self.name, '--format', '{{.Internal}}') == 'true' # SMTP sink accepts and closes connections; no real mail can leave the test network. self.container('smtp', '--network-alias', 'smtp', 'alpine:3.22', 'nc', '-lk', '-p', '465', '-e', '/bin/true') for mode in ('prod', 'venue'): self.container(mode + '-pg', '--network-alias', mode + '-pg', '-e', 'POSTGRES_USER=test', '-e', 'POSTGRES_PASSWORD=test-only', '-e', 'POSTGRES_DB=test', 'postgres:15') self.container(mode + '-s3', '--network-alias', mode + '-s3', '-e', 'RUSTFS_ACCESS_KEY=test', '-e', 'RUSTFS_SECRET_KEY=test-only-password', '-e', 'RUSTFS_CONSOLE_ENABLE=false', 'rustfs/rustfs:latest') wait_for(lambda: self.sql(mode, 'SELECT 1;') == '1', mode + ' PostgreSQL') wait_for(lambda: run('docker', 'exec', self.name + '-' + mode + '-s3', 'curl', '-fsS', 'http://127.0.0.1:9000/health') is not None, mode + ' RustFS') for migration in sorted((ROOT / 'evening_detective_server/migrations').glob('*.sql')): self.sql(mode, migration.read_text().split('-- +goose Down')[0]) bootstrap = (ROOT / 'evening_detective_server/scripts/dev-user.sql').read_text() run('docker', 'exec', '-i', self.name + '-' + mode + '-pg', 'psql', '-U', 'test', '-d', 'test', '-v', 'ON_ERROR_STOP=1', '-v', 'dev_username=TEST organizer', '-v', 'dev_email=organizer@test.invalid', '-v', 'dev_password=test-only-password', input=bootstrap) if mode == 'venue': self.sql(mode, "UPDATE users SET roles='[\"organizer\",\"user\"]'::jsonb;") self.container(mode + '-app', '-v', str(self.directory / 'server') + ':/server:ro', '-w', '/tmp', '-e', 'APP_MODE=' + mode, '-e', f'DATABASE_URL=postgresql://test:test-only@{mode}-pg:5432/test?sslmode=disable', '-e', 'APP_BASE_URL=' + self.public, '-e', 'FILE_PREFIX_DOMAIN=' + self.public + '/api/files/', '-e', 'JWT_SECRET=test-only-jwt-secret', '-e', 'PARTICIPATION_SECRET=test-only-participation-secret', '-e', 'S3_HOST=http://' + mode + '-s3:9000', '-e', 'S3_USER=test', '-e', 'S3_PASSWORD=test-only-password', '-e', 'S3_BUCKET=test', '-e', 'SMTP_HOST=smtp', '-e', 'SMTP_PORT=465', '-e', 'SMTP_USER=test@test.invalid', '-e', 'SMTP_PASSWORD=test-only', '-e', 'SMTP_TIMEOUT=1s', '-e', 'GAME_TIMEZONE=Asia/Barnaul', 'alpine:3.22', '/server') address = json.loads(run('docker', 'inspect', self.name + '-' + mode + '-app'))[0]['NetworkSettings']['Networks'][self.name]['IPAddress'] api = self.apis[mode] = API('http://' + address + ':8090') wait_for(lambda: api.request('/api/test/ping') == {}, mode + ' API') api.token = api.request('/api/auth/login', {'email': 'organizer@test.invalid', 'password': 'test-only-password'})['accessToken'] assert api.request('/api/ui/permissions')['environment'] == mode self.server_thread.start() print('PASS: separate prod/venue DB and RustFS in internal Docker network', flush=True) def restart(self, mode='venue'): for suffix in (mode + '-app', mode + '-pg', mode + '-s3'): run('docker', 'stop', self.name + '-' + suffix) for suffix in (mode + '-pg', mode + '-s3', mode + '-app'): run('docker', 'start', self.name + '-' + suffix) if suffix == mode + '-pg': wait_for(lambda: self.sql(mode, 'SELECT 1;') == '1', 'restarted PostgreSQL') elif suffix == mode + '-s3': wait_for(lambda: run('docker', 'exec', self.name + '-' + mode + '-s3', 'curl', '-fsS', 'http://127.0.0.1:9000/health') is not None, 'restarted RustFS') wait_for(lambda: self.apis[mode].request('/api/test/ping') == {}, 'restarted ' + mode) def close(self): if self.server_thread.is_alive(): self.server.shutdown() self.server_thread.join(timeout=5) self.server.server_close() errors = [] for name in reversed(self.containers): try: run('docker', 'rm', '-f', '-v', name) except RuntimeError as error: errors.append(str(error)) if self.network_created: try: run('docker', 'network', 'rm', self.name) except RuntimeError as error: errors.append(str(error)) if errors: raise RuntimeError('Stand cleanup failed: ' + '; '.join(errors)) def fixture(stand, format): source = stand.apis['prod'] scenario = source.request('/api/scenario', {'name': 'TEST stage 8 ' + format})['ID'] filename = source.request('/api/files/upload', {'filename': 'stage8.pdf', 'data': base64.b64encode(MATERIAL).decode()})['filename'] source.request(f'/api/scenarios/{scenario}/introduction', {'introduction': {'text': 'TEST Introduction'}}, 'PUT') for code in ('p1', 'p2'): source.request(f'/api/scenarios/{scenario}/places', {'place': {'code': code, 'name': 'TEST Place ' + code, 'text': 'TEST Text ' + code, 'applications': [{'name': 'TEST Clue ' + code, 'image': filename}]}}) source.request(f'/api/scenarios/{scenario}/questions', {'question': {'code': 'q1', 'text': 'TEST Who?', 'answer': 'Butler'}}) start = (datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(hours=48)).isoformat() game = source.request('/api/games', {'name': 'TEST stage 8 ' + format, 'description': 'Disposable full flow', 'startAt': start, 'scenarioID': scenario, 'format': format, 'location': 'TEST venue'})['ID'] teams = [] for name, participation in [('Captain', 'confirmed'), ('Pending', 'pending'), ('Cancelled', 'cancelled')]: team = source.request('/api/teams', {'gameID': game, 'name': 'TEST ' + name, 'participantsCount': 2}) teams.append(team) if participation != 'pending': # Simulate issuing the email token; decisions themselves use the public API. token = secrets.token_hex(32) stand.sql('prod', f"UPDATE teams SET participation_token_hash=decode('{hashlib.sha256(token.encode()).hexdigest()}', 'hex') WHERE id={int(team['ID'])};") source.request('/api/participation/answer', {'token': token, 'answer': participation}) tickets = source.request(f'/api/games/{game}/tickets', {'count': 5})['tickets'] used = '/api' + urllib.parse.urlparse(tickets[0]['URL']).path source.request(used + '/accept', {}) result = {'game': game, 'scenario': scenario, 'format': format, 'teams': teams, 'tickets': tickets, 'startAt': start, 'filename': filename} (stand.directory / f'fixture-{game}.json').write_text(json.dumps(result)) return result def export_package(api, game): api.request(f'/api/games/{game}/export', {}) wait_for(lambda: api.game(game).get('export', {}).get('state') == 'ready', 'export ready') return api.request(f'/api/games/{game}/export/download', raw=True) def exercise(stand, data): source, venue = stand.apis['prod'], stand.apis['venue'] game, format = data['game'], data['format'] captain, pending, cancelled = data['teams'] package = data['package'] original = source.game(game) local = venue.game(game) assert local['format'] == format and local['status'] == 'draft' for old, new in zip(original['teams'], local['teams'], strict=True): for key in ('ID', 'password', 'participationStatus', 'participantsCount'): assert old[key] == new[key], 'Imported team changed: ' + key for ticket in data['tickets']: path = '/api' + urllib.parse.urlparse(ticket['URL']).path before, after = source.request(path)['ticket'], venue.request(path)['ticket'] assert before['ID'] == after['ID'] and before['status'] == after['status'] stand.current = 'venue' public = API(stand.public) assert public.story(captain)['game']['ID'] == game public.story({'ID': captain['ID'], 'password': 'wrong'}, expected=404) public.story(cancelled, expected=404) assert venue.request(f'/api/games/{game}/teams/qr.pdf', raw=True).startswith(b'%PDF-') source.request(f'/api/teams/{captain["ID"]}', {'name': 'TEST Changed after export'}, 'PUT') assert venue.game(game)['teams'][0]['name'] != 'TEST Changed after export' used, active, concurrent, after_finish, unused = ['/api' + urllib.parse.urlparse(t['URL']).path for t in data['tickets']] venue.request(used + '/accept', {}, expected=400) assert public.request(active)['ticket']['status'] == 'active' public.request(active + '/accept', {}, expected=401) venue.request(active + '/accept', {}) venue.request(active + '/accept', {}, expected=400) with ThreadPoolExecutor(max_workers=2) as pool: results = list(pool.map(lambda _: venue.request(concurrent + '/accept', {}, expected=(200, 400)), range(2))) assert sum('ticket' in result for result in results) == 1 rejected = next(result for result in results if 'ticket' not in result) assert rejected['code'] == 9 and rejected['message'] == 'билет уже использован' public.request('/api/tickets/invalid', expected=404) venue.request('/api/tickets/invalid/accept', {}, expected=404) local_ticket = venue.request(f'/api/games/{game}/tickets', {'count': 1})['tickets'][0] data['local_ticket'] = local_ticket assert local_ticket['URL'].startswith(stand.public + '/tickets/') assert public.request('/api' + urllib.parse.urlparse(local_ticket['URL']).path)['ticket']['status'] == 'active' new_team = venue.request('/api/teams', {'gameID': game, 'name': 'TEST Local team', 'participantsCount': 2}) data['new_team'] = new_team assert new_team['ID'] > max(t['ID'] for t in data['teams']) assert public.story(new_team)['game']['ID'] == game public.move(pending, 'p1', expected=400) assert next(t for t in venue.game(game)['teams'] if t['ID'] == pending['ID'])['participationStatus'] == 'pending' venue.request(f'/api/games/{game}/start', {}, 'PUT') public.move(cancelled, 'p1', expected=404) public.move({'ID': captain['ID'], 'password': 'wrong'}, 'p1', expected=404) public.move(pending, 'p1') public.move(new_team, 'p1') confirmed = next(t for t in venue.game(game)['teams'] if t['ID'] == pending['ID']) assert confirmed['participationStatus'] == 'confirmed' and confirmed['participationSource'] == 'first_action' public.move(captain, 'p1') view = public.story(captain) assert len(view['actions']) == 1 and view['story']['places'][0]['code'] == 'p1' assert not view['story'].get('conclusion') clue = next(t for t in venue.game(game)['teams'] if t['ID'] == captain['ID'])['applications'][0] assert OPENER.open(clue['image'], timeout=10).read() == MATERIAL if format == 'offline': assert not view['story'].get('questions') and not view['story'].get('introduction') assert not view['story']['places'][0]['applications'][0].get('image') else: assert view['story']['introduction']['text'] == 'TEST Introduction' assert not view['story']['questions'][0].get('answer') venue.request(f'/api/teams/{captain["ID"]}/applications', {'name': clue['name']}) assert not next(t for t in venue.game(game)['teams'] if t['ID'] == captain['ID'])['applications'] query = urllib.parse.urlencode({'password': captain['password']}) answers_path = f'/api/teams/{captain["ID"]}/answers' answers = {'password': captain['password'], 'answers': [{'questionCode': 'q1', 'answer': 'Butler'}]} public.request(answers_path, answers, expected=400 if format == 'offline' else 200) venue.request(f'/api/games/{game}/pause', {}, 'PUT') public.move(captain, 'p2', expected=400) assert len(public.story(captain)['actions']) == 1 stand.restart() assert venue.game(game)['status'] == 'pause' assert len(public.story(captain)['actions']) == 1 assert public.request(active)['ticket']['status'] == 'used' assert OPENER.open(clue['image'], timeout=10).read() == MATERIAL if format == 'online': assert public.request(answers_path + '?' + query)['answers'][0]['answer'] == 'Butler' assert venue.request('/api/games/import', package)['repeated'] assert venue.game(game)['status'] == 'pause' venue.request(f'/api/games/{game}/play', {}, 'PUT') public.move(captain, 'p2') assert len(public.story(captain)['actions']) == 2 venue.request(f'/api/games/{game}/finish', {}, 'PUT') public.move(captain, 'p1', expected=400) public.request(answers_path, answers, expected=400) venue.request(after_finish + '/accept', {}, expected=400) finished = public.story(captain) assert finished['game']['status'] == 'finish' and not finished.get('actions') and not finished['story'].get('places') venue.request(f'/api/teams/{captain["ID"]}/answers/q1/score', {'score': 5}, 'PUT') results = venue.request(f'/api/games/{game}/results') scored = next(t for t in results['teams'] if t['team']['ID'] == captain['ID']) assert scored['totalScore'] == 5 and scored['team']['actionsCount'] == 2 assert all(t['team']['ID'] != cancelled['ID'] for t in results['teams']) before_repeat = venue.game(game) assert venue.request('/api/games/import', package)['repeated'] assert venue.game(game) == before_repeat assert source.game(game)['status'] == 'draft' assert source.request(active)['ticket']['status'] == 'active' assert not next(t for t in source.game(game)['teams'] if t['ID'] == captain['ID'])['actionsCount'] print('PASS: ' + format + ' export/import, original links, tickets, teams, gameplay, restart, results and repeat', flush=True) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--apply', action='store_true', help='Also verify stage 10 applying venue results to prod') parser.add_argument('--results', action='store_true', help='Also verify stage 9 result packages before/after restart') parser.add_argument('--browser', action='store_true', help='Also run Firefox against the real APIs (Firefox required)') args = parser.parse_args() args.results = args.results or args.apply with tempfile.TemporaryDirectory(prefix='ed029-flow-') as directory: stand = Stand(directory) try: stand.start() pg_address = json.loads(run('docker', 'inspect', stand.name + '-venue-pg'))[0]['NetworkSettings']['Networks'][stand.name]['IPAddress'] run('env', f'TEST_DATABASE_URL=host={pg_address} user=test password=test-only dbname=test sslmode=disable', 'go', 'test', './internal/services/game_service', './internal/services/gameplay_service', './internal/services/mcp_service', './internal/repos/tickets_repo', './internal/app', cwd=ROOT / 'evening_detective_server') print('PASS: gameplay/lifecycle/MCP/app tests with isolated PostgreSQL', flush=True) prod_address = json.loads(run('docker', 'inspect', stand.name + '-prod-pg'))[0]['NetworkSettings']['Networks'][stand.name]['IPAddress'] run('env', f'TEST_DATABASE_URL=host={prod_address} user=test password=test-only dbname=test sslmode=disable', 'go', 'test', '-race', './internal/services/game_export_service', './internal/services/game_results_package_service', './internal/repos/game_returns_repo', '-count=1', cwd=ROOT / 'evening_detective_server') print('PASS: PostgreSQL/race result return regressions and real HTTP transport', flush=True) fixtures = [fixture(stand, format) for format in ('offline', 'online')] for data in fixtures: package = data['package'] = export_package(stand.apis['prod'], data['game']) (stand.directory / f'game-{data["game"]}.zip').write_bytes(package) imported = stand.apis['venue'].request('/api/games/import', package) assert imported['gameID'] == data['game'] and not imported['repeated'] for data in fixtures: exercise(stand, data) if args.results: source_before = [stand.apis['prod'].game(data['game']) for data in fixtures] venue_before = [stand.apis['venue'].game(data['game']) for data in fixtures] for attempt in range(2): if attempt: stand.restart() for data in fixtures: with zipfile.ZipFile(stand.directory / f'game-{data["game"]}.zip') as archive: manifest = json.loads(archive.read('manifest.json')) result = run('env', f'TEST_DATABASE_URL=host={pg_address} user=test password=test-only dbname=test sslmode=disable', f'ED029_RESULTS_GAME_ID={data["game"]}', f'ED029_RESULTS_PACKAGE_ID={manifest["package_id"]}', f'ED029_RESULTS_DIR={stand.directory}', 'go', 'test', './internal/services/game_results_package_service', '-run', 'TestVenueFlowResults', '-count=1', '-v', cwd=ROOT / 'evening_detective_server') assert '--- PASS: TestVenueFlowResults' in result with zipfile.ZipFile(stand.directory / f'results-{data["game"]}.zip') as archive: result_manifest = json.loads(archive.read('manifest.json')) payload = archive.read('results.json') assert set(archive.namelist()) == {'manifest.json', 'results.json'} assert result_manifest['source_sha256'] == hashlib.sha256(data['package']).hexdigest() assert result_manifest['sha256'] == hashlib.sha256(payload).hexdigest() assert b'game-import-' not in payload assert [stand.apis['prod'].game(data['game']) for data in fixtures] == source_before assert [stand.apis['venue'].game(data['game']) for data in fixtures] == venue_before print('PASS: stage 9 real result packages, fresh grading, concurrent snapshots, restart and unchanged games', flush=True) if args.apply: prod_pg = json.loads(run('docker', 'inspect', stand.name + '-prod-pg'))[0]['NetworkSettings']['Networks'][stand.name]['IPAddress'] prod_s3 = json.loads(run('docker', 'inspect', stand.name + '-prod-s3'))[0]['NetworkSettings']['Networks'][stand.name]['IPAddress'] for attempt in range(2): if attempt: stand.restart('prod') for data in fixtures: result = run('env', f'NO_PROXY={prod_s3}', f'no_proxy={prod_s3}', f'TEST_DATABASE_URL=host={prod_pg} user=test password=test-only dbname=test sslmode=disable', f'ED029_RETURN_GAME_ID={data["game"]}', f'ED029_RETURN_SOURCE={stand.public}', f'ED029_RETURN_S3=http://{prod_s3}:9000', f'ED029_RESULTS_DIR={stand.directory}', 'go', 'test', './internal/services/game_results_package_service', '-run', 'TestProdFlowReturns', '-count=1', '-v', cwd=ROOT / 'evening_detective_server') assert '--- PASS: TestProdFlowReturns' in result source = stand.apis['prod'] returned = source.game(data['game']) assert returned['status'] == 'finish' and not returned.get('export') assert returned['name'] == 'TEST Online title after export' assert returned['description'] == 'TEST Online description' assert source.story(data['teams'][0])['game']['status'] == 'finish' new_team = next(t for t in returned['teams'] if t['ID'] == data['new_team']['ID']) assert new_team['password'] == data['new_team']['password'] assert source.story(data['new_team'])['game']['status'] == 'finish' assert source.request('/api' + urllib.parse.urlparse(data['local_ticket']['URL']).path)['ticket']['status'] == 'active' assert source.request('/api' + urllib.parse.urlparse(data['tickets'][1]['URL']).path)['ticket']['status'] == 'used' scored = next(t for t in source.request(f'/api/games/{data["game"]}/results')['teams'] if t['team']['ID'] == data['teams'][0]['ID']) assert scored['totalScore'] == 0 and scored['team']['actionsCount'] == 2 print('PASS: stage 10 atomic prod returns, correction, original/venue links, preserved presentation, cleared export and restart repeats', flush=True) finally: stand.close() if args.browser: from flow_browser import check_browser_flow with tempfile.TemporaryDirectory(prefix='ed029-browser-flow-') as directory: stand = Stand(directory) try: stand.start() check_browser_flow(stand, [fixture(stand, format) for format in ('online', 'offline')]) finally: stand.close() print('PASS: stage 8 automated checks; phone/camera/no-WAN acceptance remains manual') if __name__ == '__main__': main()