146 lines
7.3 KiB
Python
146 lines
7.3 KiB
Python
import json
|
|
import tempfile
|
|
import sqlite3
|
|
from pathlib import Path
|
|
import threading
|
|
import urllib.request
|
|
import urllib.error
|
|
from http.server import HTTPServer
|
|
from test_room import ROOT, server
|
|
import unittest
|
|
|
|
|
|
class FleetTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.now = 100
|
|
self.catalog = json.loads((ROOT / 'tools/room_controller/catalog.json').read_text())
|
|
self.fleet = server.Fleet(':memory:', self.catalog, lambda: self.now)
|
|
for cabinet in self.fleet.rooms: self.poll(cabinet)
|
|
|
|
def tearDown(self):
|
|
for room in self.fleet.rooms.values(): room.db.close()
|
|
|
|
def poll(self, cabinet, reports=None):
|
|
return self.fleet.poll(dict(cabinet=cabinet, protocol=2, boot=cabinet, reports=reports or []))
|
|
|
|
def ack(self, cabinet):
|
|
commands = self.poll(cabinet)['commands']
|
|
self.poll(cabinet, [dict(kind='ack', id=c['id'], session=c['session'], ok=True) for c in commands])
|
|
|
|
def test_scenario_routes_games_and_events_to_correct_cabinets(self):
|
|
self.fleet.operator('prepare_scenario', {'scenario': 'two_cabinet_test'})
|
|
self.assertEqual(self.poll('cabinet_01')['commands'][0]['game'], 'pacman')
|
|
self.assertEqual(self.poll('cabinet_02')['commands'][0]['game'], 'galaga')
|
|
self.assertFalse(self.poll('cabinet_03')['commands'])
|
|
self.ack('cabinet_01')
|
|
self.assertEqual(self.fleet.rooms['cabinet_02'].state['phase'], 'preparing')
|
|
self.ack('cabinet_02')
|
|
self.assertEqual(self.poll('cabinet_02')['commands'][0]['action_id'], 'glitch_code')
|
|
self.assertFalse(self.poll('cabinet_01')['commands'])
|
|
|
|
def test_offline_assignment_prevents_any_scenario_dispatch(self):
|
|
self.now += 5
|
|
self.poll('cabinet_01')
|
|
with self.assertRaisesRegex(ValueError, 'offline'):
|
|
self.fleet.operator('prepare_scenario', {'scenario': 'two_cabinet_test'})
|
|
self.assertFalse(self.fleet.rooms['cabinet_01'].state['commands'])
|
|
|
|
def test_manual_assignment_does_not_inherit_reward(self):
|
|
self.fleet.operator('prepare', {'cabinet': 'cabinet_01', 'game': 'galaga'})
|
|
self.ack('cabinet_01')
|
|
room = self.fleet.rooms['cabinet_01']
|
|
self.poll('cabinet_01', [dict(kind='event', id=event, event=event, session=room.state['session']) for event in ['started', 'won']])
|
|
self.assertEqual(room.state['phase'], 'won')
|
|
self.assertFalse(self.poll('cabinet_01')['commands'])
|
|
|
|
def test_compatibility_rejects_missing_controls_and_accepts_alternatives(self):
|
|
with self.assertRaisesRegex(ValueError, 'hardware'):
|
|
self.fleet.operator('prepare', {'cabinet': 'cabinet_01', 'game': 'missile_command'})
|
|
cabinet = self.catalog['cabinets']['cabinet_03']
|
|
self.assertFalse(self.fleet.compatible('cabinet_03', 'tempest'))
|
|
cabinet['capabilities'].append('button_2')
|
|
self.assertTrue(self.fleet.compatible('cabinet_03', 'tempest'))
|
|
cabinet['capabilities'].remove('spinner')
|
|
self.assertFalse(self.fleet.compatible('cabinet_03', 'tempest'))
|
|
cabinet['capabilities'].append('trackball')
|
|
self.assertTrue(self.fleet.compatible('cabinet_03', 'tempest'))
|
|
|
|
def test_old_export_cannot_silently_load_wrong_game(self):
|
|
with self.assertRaisesRegex(ValueError, 'protocol 2'):
|
|
self.fleet.poll(dict(cabinet='cabinet_01', protocol=1, boot='old'))
|
|
|
|
def test_invalid_mapping_does_not_dispatch(self):
|
|
self.catalog['scenarios']['two_cabinet_test']['assignments'][1]['mappings']['unknown_event'] = ['coin_release']
|
|
with self.assertRaisesRegex(ValueError, 'not implemented'):
|
|
self.fleet.operator('prepare_scenario', {'scenario': 'two_cabinet_test'})
|
|
self.assertFalse(self.poll('cabinet_01')['commands'])
|
|
|
|
def test_output_timeout_does_not_report_a_reward(self):
|
|
room = self.fleet.rooms['cabinet_01']
|
|
room.command('output')
|
|
self.now += 16
|
|
room.expire()
|
|
self.assertEqual(room.state['reward'], 'not requested')
|
|
self.assertIn('unknown', room.state['last_output'])
|
|
|
|
def test_one_accepted_output_cannot_hide_another_failed_reward(self):
|
|
room = self.fleet.rooms['cabinet_01']
|
|
first, second = room.command('reward'), room.command('reward')
|
|
room.ack(dict(id=first['id'], ok=False))
|
|
room.ack(dict(id=second['id'], ok=True))
|
|
self.assertIn('failed', room.state['reward'])
|
|
|
|
def test_migration_and_independent_persistence(self):
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
path = str(Path(temp) / 'migration.db')
|
|
legacy = sqlite3.connect(path)
|
|
legacy.execute('CREATE TABLE room (id INTEGER PRIMARY KEY, value TEXT NOT NULL)')
|
|
old = dict(self.fleet.rooms['cabinet_01'].state, session='old-session', phase='won')
|
|
legacy.execute('INSERT INTO room VALUES (1,?)', (json.dumps(old),))
|
|
legacy.commit()
|
|
legacy.close()
|
|
fleet = server.Fleet(path, self.catalog)
|
|
try:
|
|
self.assertEqual(fleet.rooms['cabinet_01'].state['session'], 'old-session')
|
|
self.assertEqual(fleet.rooms['cabinet_02'].state['session'], '')
|
|
fleet.rooms['cabinet_02'].state['game'] = 'galaga'
|
|
fleet.rooms['cabinet_02'].save()
|
|
finally:
|
|
for room in fleet.rooms.values(): room.db.close()
|
|
restored = server.Fleet(path, self.catalog)
|
|
try:
|
|
self.assertEqual(restored.rooms['cabinet_01'].state['phase'], 'won')
|
|
self.assertEqual(restored.rooms['cabinet_02'].state['game'], 'galaga')
|
|
self.assertFalse(restored.rooms['cabinet_01'].online())
|
|
finally:
|
|
for room in restored.rooms.values(): room.db.close()
|
|
|
|
def test_http_browser_and_identity_bound_credentials(self):
|
|
service = HTTPServer(('127.0.0.1', 0), server.handler_for(self.fleet, 'operator-secret', {'cabinet_01': 'cabinet-one', 'cabinet_02': 'cabinet-two'}))
|
|
thread = threading.Thread(target=service.serve_forever, daemon=True)
|
|
thread.start()
|
|
base = 'http://127.0.0.1:' + str(service.server_port)
|
|
try:
|
|
with urllib.request.urlopen(base) as response:
|
|
self.assertIn(b'Room control', response.read())
|
|
def request(path, token, data, origin=base):
|
|
req = urllib.request.Request(base + path, data=json.dumps(data).encode(), headers={'Authorization': 'Bearer '+token, 'Origin': origin})
|
|
return urllib.request.urlopen(req)
|
|
with request('/operator/state', 'operator-secret', {}) as response:
|
|
self.assertEqual(len(json.load(response)['cabinets']), 3)
|
|
for path, token, data, code, origin in [
|
|
('/operator/state', 'cabinet-one', {}, 401, base),
|
|
('/cabinet/poll', 'cabinet-one', {'cabinet':'cabinet_02','protocol':2,'boot':'x'}, 401, base),
|
|
('/operator/state', 'operator-secret', {}, 400, 'http://unrelated.example'),
|
|
]:
|
|
with self.assertRaises(urllib.error.HTTPError) as caught:
|
|
request(path, token, data, origin)
|
|
self.assertEqual(caught.exception.code, code)
|
|
finally:
|
|
service.shutdown()
|
|
service.server_close()
|
|
thread.join()
|
|
|
|
|
|
if __name__ == '__main__': unittest.main()
|