Verstack Local
f73167de85
Package Apple Silicon VM exports and cached runtime switching with local cabinet controls and conagent provisioning. Preserve experimental SPIKE 2 emulation. Improve preparation concurrency, Ghidra checkpoints, signature importing, sound indexing and client-side spectrograms. Include regression tests and validation notes.
167 lines
8.9 KiB
Python
167 lines
8.9 KiB
Python
"""Real broker/session regressions. Run inside the emulator dependency image."""
|
|
import asyncio
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import sys
|
|
import time
|
|
import unittest
|
|
from aiohttp.test_utils import TestClient, TestServer
|
|
import server
|
|
|
|
|
|
class Sessions(unittest.IsolatedAsyncioTestCase):
|
|
async def asyncSetUp(self):
|
|
self.directory = tempfile.TemporaryDirectory()
|
|
self.previous = server.SESSIONS
|
|
server.SESSIONS = Path(self.directory.name)
|
|
self.client = TestClient(TestServer(server.application()))
|
|
await self.client.start_server()
|
|
|
|
async def asyncTearDown(self):
|
|
await self.client.close()
|
|
server.SESSIONS = self.previous
|
|
self.directory.cleanup()
|
|
|
|
async def create(self):
|
|
response = await self.client.post('/emulator/sessions', json={'mode':'diagnostic'}, headers={'X-Verstack-Client':'1'})
|
|
self.assertEqual(response.status, 200, await response.text())
|
|
session = await response.json()
|
|
for _ in range(50):
|
|
result = await self.get(session)
|
|
if 'machine' in result:
|
|
return session
|
|
await asyncio.sleep(.05)
|
|
self.fail('Machine broker did not start')
|
|
|
|
def headers(self, session):
|
|
return {'Authorization':'Bearer ' + session['token'], 'X-Verstack-Client':'1'}
|
|
|
|
async def get(self, session):
|
|
response = await self.client.get('/emulator/sessions/' + session['id'], headers=self.headers(session))
|
|
self.assertEqual(response.status, 200)
|
|
return await response.json()
|
|
|
|
async def test_isolation_stream_controls_and_cleanup(self):
|
|
a, b = await self.create(), await self.create()
|
|
denied = await self.client.get('/emulator/sessions/' + b['id'], headers=self.headers(a))
|
|
self.assertEqual(denied.status, 404)
|
|
full = await self.client.post('/emulator/sessions', json={'mode':'diagnostic'}, headers={'X-Verstack-Client':'1'})
|
|
self.assertEqual(full.status, 409)
|
|
path = '/emulator/sessions/' + a['id']
|
|
pressed = await self.client.post(path + '/event', json={'name':'left_flipper','state':True,'sequence':1}, headers=self.headers(a))
|
|
self.assertEqual(pressed.status, 200, await pressed.text())
|
|
self.assertEqual((await self.get(a))['held'], ['left_flipper'])
|
|
self.assertEqual((await self.get(b))['held'], [])
|
|
stale = await self.client.post(path + '/event', json={'name':'left_flipper','state':False,'sequence':1}, headers=self.headers(a))
|
|
self.assertEqual(stale.status, 400)
|
|
response = await self.client.get(path + '/media', headers=self.headers(a))
|
|
types = set()
|
|
async with asyncio.timeout(5):
|
|
while types != {b'A', b'V'}:
|
|
header = await response.content.readexactly(5)
|
|
payload = await response.content.readexactly(int.from_bytes(header[1:], 'big'))
|
|
types.add(header[:1])
|
|
if header[:1] == b'V':
|
|
self.assertTrue(payload.startswith(b'\xff\xd8'))
|
|
else:
|
|
self.assertEqual(len(payload) % 4, 0)
|
|
self.assertTrue(any(payload))
|
|
response.close()
|
|
for _ in range(50):
|
|
if not (await self.get(a))['held']:
|
|
break
|
|
await asyncio.sleep(.1)
|
|
self.assertEqual((await self.get(a))['held'], [])
|
|
instance = self.client.server.app['sessions'][a['id']]
|
|
stopped = await self.client.post(path + '/stop', json={}, headers=self.headers(a))
|
|
self.assertEqual(stopped.status, 200)
|
|
self.assertIsNotNone(instance.process.returncode)
|
|
self.assertFalse(instance.directory.exists())
|
|
self.assertEqual((await self.get(b))['status'], 'diagnostic')
|
|
|
|
async def test_boundary_and_invalid_inputs(self):
|
|
response = await self.client.post('/emulator/sessions', json={'mode':'diagnostic'})
|
|
self.assertEqual(response.status, 403)
|
|
response = await self.client.post('/emulator/sessions', json={'mode':'diagnostic'}, headers={'Origin':'http://elsewhere.invalid','X-Verstack-Client':'1'})
|
|
self.assertEqual(response.status, 403)
|
|
a = await self.create()
|
|
response = await self.client.post('/emulator/sessions/' + a['id'] + '/event', json={'name':'arbitrary','sequence':1}, headers=self.headers(a))
|
|
self.assertEqual(response.status, 400)
|
|
|
|
async def test_every_table_switch_reaches_broker(self):
|
|
session = await self.create()
|
|
for sequence, (name, *_rest) in enumerate(server.TABLE['PLAYFIELD_SWITCHES'], 1):
|
|
response = await self.client.post('/emulator/sessions/' + session['id'] + '/event',
|
|
json={'name':name, 'sequence':sequence}, headers=self.headers(session))
|
|
self.assertEqual(response.status, 200, name + ': ' + await response.text())
|
|
|
|
async def test_held_switch_expires_without_heartbeat(self):
|
|
session = await self.create()
|
|
response = await self.client.post('/emulator/sessions/' + session['id'] + '/event',
|
|
json={'name':'left_flipper', 'state':True, 'sequence':1}, headers=self.headers(session))
|
|
self.assertEqual(response.status, 200)
|
|
await asyncio.sleep(4.2)
|
|
self.assertEqual((await self.get(session))['held'], [])
|
|
|
|
|
|
class Broker(unittest.TestCase):
|
|
def test_pulse_timer_does_not_release_a_later_hold(self):
|
|
sys.path.insert(0, str(server.ROOT / 'vendor/emulation/machine'))
|
|
from machine import MachineModel, TraceRecorder, load_config
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
config = load_config(workspace=Path(directory))
|
|
model = MachineModel(config, TraceRecorder(None, 100))
|
|
try:
|
|
model.event('left_flipper')
|
|
model.event('left_flipper', True)
|
|
time.sleep(.7)
|
|
# Node 8 input 25 is active-low in the donor's flipper mapping.
|
|
self.assertEqual(model.netbridge.model.input_states[8][3] & 2, 0)
|
|
finally:
|
|
model.release_all()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|
|
|
|
class RuntimeSelection(unittest.IsolatedAsyncioTestCase):
|
|
async def test_explicit_release_selection_and_missing_targets(self):
|
|
from unittest.mock import patch, AsyncMock
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root=Path(tmp)
|
|
for identity,arch,targets in [('spike2','armhf',{'game':'game'}),('spike3','aarch64',{'game':'game','boot-display':'spike3/bin/boot_display'})]:
|
|
folder=root/identity;folder.mkdir()
|
|
(folder/'runtime.json').write_text(json.dumps({'schema':2,'snapshot':identity,'release':{'repository':identity,'version':'2'},'architecture':arch,'game_root':'game','system_root':'system','targets':targets}))
|
|
with patch.object(server,'RUNTIME',root),patch.object(server.Session,'start',new=AsyncMock()):
|
|
client=TestClient(TestServer(server.application()));await client.start_server()
|
|
try:
|
|
info=await (await client.get('/emulator/info')).json()
|
|
self.assertEqual(len(info['runtimes']),2)
|
|
for identity in ('spike2','spike3'):
|
|
response=await client.post('/emulator/sessions',json={'mode':'game','runtime':identity},headers={'X-Verstack-Client':'1'})
|
|
self.assertEqual(response.status,200,await response.text())
|
|
self.assertEqual((await response.json())['runtime'],identity)
|
|
for body in ({'mode':'game','runtime':'../outside'},{'mode':'boot-display','runtime':'spike2'}):
|
|
response=await client.post('/emulator/sessions',json=body,headers={'X-Verstack-Client':'1'})
|
|
self.assertEqual(response.status,400)
|
|
finally:await client.close()
|
|
|
|
class StartupStatus(unittest.TestCase):
|
|
def test_no_frame_and_failure_are_explicit(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
session = server.Session('game')
|
|
session.directory = Path(directory)
|
|
session.created = time.monotonic() - 65
|
|
self.assertIn('No game frame after 60 seconds', session.record()['startup_note'])
|
|
(session.directory / 'console.log').write_text('failed to initialize bridge: Invalid argument\n')
|
|
self.assertIn('cabinet bridge initialization', session.record()['startup_note'])
|
|
game_log = session.directory / 'emulation/work/capsule/writable/dump/debug_log.txt'
|
|
game_log.parent.mkdir(parents=True)
|
|
game_log.write_text('FATAL: NVRAM migration failed')
|
|
self.assertIn('FATAL: NVRAM migration failed', session.log_tail())
|
|
session.rendered = True
|
|
self.assertEqual(session.record()['startup_note'], '')
|
|
(session.directory / 'console.log').write_text("Failed to open device '/dev/ttymxc1': No such file or directory")
|
|
self.assertIn('SPIKE 2 node-bus', session.record()['startup_note'])
|