80 lines
4.9 KiB
Python
80 lines
4.9 KiB
Python
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
import urllib.request
|
|
import urllib.error
|
|
import zipfile
|
|
from unittest.mock import patch
|
|
from bundles.server import ExportServer
|
|
|
|
|
|
class ExportAPI(unittest.TestCase):
|
|
def test_quick_export_and_component_plan_end_to_end(self):
|
|
self.check_export('aarch64')
|
|
|
|
def test_spike2_quick_export_and_component_plan_end_to_end(self):
|
|
self.check_export('armhf')
|
|
|
|
def check_export(self, architecture):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root=Path(temporary);runtime=root/'runtimes/sd';(runtime/'partition-02/lib').mkdir(parents=True)
|
|
(runtime/'partition-02/lib/runtime').write_text('SD libraries')
|
|
(runtime/'game').mkdir();(runtime/'game/game').write_text('exact game')
|
|
(runtime/'runtime.json').write_text(json.dumps(dict(schema=2,snapshot='sd',system_snapshot='sd',
|
|
architecture=architecture,system_root='partition-02',game_root='game',targets={'game':'game'},
|
|
release=dict(repository='Fixture',version='1.0.0'))))
|
|
server=ExportServer(('127.0.0.1',0),root/'exports',root/'runtimes')
|
|
thread=threading.Thread(target=server.serve_forever);thread.start()
|
|
base='http://127.0.0.1:'+str(server.server_port)
|
|
def get(path):return json.load(urllib.request.urlopen(base+'/emulator/exports'+path))
|
|
try:
|
|
self.assertEqual(get('/catalog')[0]['runtime_label'],'Fixture 1.0.0')
|
|
self.assertTrue(get('/catalog')[0]['exportable'])
|
|
if architecture=='armhf':
|
|
standard=urllib.request.Request(base+'/emulator/exports',
|
|
data=json.dumps(dict(game_id='sd',mode='standard',catalog_url=base)).encode(),
|
|
headers={'Content-Type':'application/json','X-Verstack-Client':'1'})
|
|
with patch('bundles.jobs.spawn'):
|
|
accepted=json.load(urllib.request.urlopen(standard))
|
|
self.assertEqual(accepted['status'],'queued')
|
|
self.assertEqual(get('/'+accepted['id'])['runtime_id'],'sd')
|
|
req=urllib.request.Request(base+'/emulator/exports',data=json.dumps(dict(game_id='sd',mode='quick',catalog_url=base)).encode(),headers={'Content-Type':'application/json','X-Verstack-Client':'1'})
|
|
job=json.load(urllib.request.urlopen(req));deadline=time.monotonic()+20
|
|
while time.monotonic()<deadline:
|
|
status=get('/'+job['id'])
|
|
if status['status'] in ('complete','failed'):break
|
|
time.sleep(.1)
|
|
self.assertEqual(status['status'],'complete',status)
|
|
archive=root/'download.zip'
|
|
archive.write_bytes(urllib.request.urlopen(base+'/emulator/exports/'+job['id']+'/download').read())
|
|
with zipfile.ZipFile(archive) as zipped:
|
|
self.assertEqual(json.loads(zipped.read('connection.json'))['game_id'],'sd')
|
|
self.assertIn('export-tools/bundles/vendor/pokemon_emulator/emulation/dashboard/server.py',zipped.namelist())
|
|
self.assertIn('Launch-universal.command',zipped.namelist())
|
|
self.assertIn('export-tools/bundles/spike2/worker/runner.py',zipped.namelist())
|
|
self.assertIn('export-tools/bundles/spike2/launch-spike2.py',zipped.namelist())
|
|
failed=root/'exports/jobs'/('f'*32);failed.mkdir()
|
|
(failed/'job.json').write_text(json.dumps(dict(id='f'*32,status='failed',created=time.time(),
|
|
request=dict(game_id='sd',shared_runtime_id=None,mode='components'),
|
|
runtime_id='sd',error='Transient preparation failure')))
|
|
before=set((root/'exports/jobs').iterdir())
|
|
with self.assertRaises(urllib.error.HTTPError) as error:
|
|
get('/plan?game_id=sd')
|
|
self.assertEqual(error.exception.code,500)
|
|
self.assertEqual(set((root/'exports/jobs').iterdir()),before)
|
|
plan=get('/plan?game_id=sd&retry=1')
|
|
while time.monotonic()<deadline:
|
|
if plan.get('status')!='preparing':break
|
|
time.sleep(.1)
|
|
plan=get('/plan?game_id=sd')
|
|
self.assertEqual(plan['selection_reason'],'own-sd')
|
|
self.assertEqual(len(set((root/'exports/jobs').iterdir())-before),1)
|
|
for kind,text in [('runtime','SD libraries'),('game','exact game')]:
|
|
component=root/(kind+'.zip');component.write_bytes(urllib.request.urlopen(base+plan[kind+'_component']['url']).read())
|
|
with zipfile.ZipFile(component) as zipped:
|
|
self.assertIn(text.encode(),[zipped.read(n) for n in zipped.namelist()])
|
|
finally:server.shutdown();thread.join();server.server_close()
|