Files
agents/tests/venue/flow.py
T
2026-10-11 12:08:36 +07:00

434 lines
25 KiB
Python

"""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.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=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.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('/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))
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=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):
for suffix in ('venue-app', 'venue-pg', 'venue-s3'):
run('docker', 'stop', self.name + '-' + suffix)
for suffix in ('venue-pg', 'venue-s3', 'venue-app'):
run('docker', 'start', self.name + '-' + suffix)
if suffix == 'venue-pg':
wait_for(lambda: self.sql('venue', 'SELECT 1;') == '1', 'restarted PostgreSQL')
elif suffix == 'venue-s3':
wait_for(lambda: run('docker', 'exec', self.name + '-venue-s3', 'curl', '-fsS', 'http://127.0.0.1:9000/health') is not None, 'restarted RustFS')
wait_for(lambda: self.apis['venue'].request('/api/test/ping') == {}, 'restarted venue')
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]
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})
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('--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()
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)
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)
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()