99 lines
4.6 KiB
Python
99 lines
4.6 KiB
Python
import unittest
|
|
import test_fleet
|
|
import tempfile
|
|
from pathlib import Path
|
|
from test_room import server
|
|
|
|
|
|
class ChainTests(unittest.TestCase):
|
|
setUp = test_fleet.FleetTests.setUp
|
|
tearDown = test_fleet.FleetTests.tearDown
|
|
poll = test_fleet.FleetTests.poll
|
|
ack = test_fleet.FleetTests.ack
|
|
|
|
def prepare(self):
|
|
self.fleet.operator('prepare_scenario', {'scenario':'cabinet_chain_test'})
|
|
self.ack('cabinet_01')
|
|
self.ack('cabinet_02')
|
|
|
|
def win(self):
|
|
source = self.fleet.rooms['cabinet_01']
|
|
self.poll('cabinet_01', [dict(kind='event', event=e, id=e, session=source.state['session']) for e in ['started','won']])
|
|
|
|
def test_wait_load_start_once(self):
|
|
self.prepare()
|
|
self.assertEqual(self.fleet.rooms['cabinet_02'].state['phase'], 'waiting')
|
|
self.assertFalse(self.poll('cabinet_02')['commands'])
|
|
with self.assertRaises(ValueError): self.fleet.operator('start', {'cabinet':'cabinet_02'})
|
|
self.win()
|
|
commands = self.poll('cabinet_02')['commands']
|
|
self.assertEqual([c['type'] for c in commands], ['prepare'])
|
|
self.assertEqual(commands[0]['game'], 'galaga')
|
|
self.win()
|
|
self.assertEqual(len(self.poll('cabinet_02')['commands']), 1)
|
|
self.ack('cabinet_02')
|
|
self.assertEqual([c['type'] for c in self.poll('cabinet_02')['commands']], ['start'])
|
|
self.ack('cabinet_02')
|
|
self.assertEqual(self.fleet.rooms['cabinet_02'].state['phase'], 'playing')
|
|
self.win()
|
|
self.assertFalse(self.poll('cabinet_02')['commands'])
|
|
|
|
def test_loss_does_not_activate(self):
|
|
self.prepare()
|
|
source=self.fleet.rooms['cabinet_01']
|
|
self.poll('cabinet_01',[dict(kind='event',event=e,id=e,session=source.state['session']) for e in ['started','lost']])
|
|
self.assertEqual(self.fleet.rooms['cabinet_02'].state['phase'],'waiting')
|
|
|
|
def test_reset_invalidates_chain_and_old_win(self):
|
|
self.prepare()
|
|
old=self.fleet.rooms['cabinet_01'].state['session']
|
|
self.fleet.operator('reset', {'cabinet':'cabinet_01'})
|
|
self.poll('cabinet_01',[dict(kind='event',event='won',id='late',session=old)])
|
|
self.assertEqual(self.fleet.rooms['cabinet_02'].state['phase'],'needs_prepare')
|
|
self.assertFalse(self.poll('cabinet_02')['commands'])
|
|
|
|
def test_target_reconnect_receives_activation(self):
|
|
self.prepare()
|
|
self.now+=5
|
|
self.win()
|
|
self.assertEqual(self.fleet.rooms['cabinet_02'].state['phase'],'waiting')
|
|
self.assertEqual(self.poll('cabinet_02')['commands'][0]['type'],'prepare')
|
|
|
|
def test_failed_load_does_not_start(self):
|
|
self.prepare()
|
|
self.win()
|
|
command=self.poll('cabinet_02')['commands'][0]
|
|
self.poll('cabinet_02',[dict(kind='ack',id=command['id'],session=command['session'],ok=False)])
|
|
self.assertEqual(self.fleet.rooms['cabinet_02'].state['phase'],'needs_prepare')
|
|
self.assertFalse(self.poll('cabinet_02')['commands'])
|
|
|
|
def test_cycles_rejected_before_dispatch(self):
|
|
self.catalog['scenarios']['cabinet_chain_test']['assignments'][0]['activate_after_win']='cabinet_02'
|
|
with self.assertRaisesRegex(ValueError,'cycle'):
|
|
self.fleet.operator('prepare_scenario', {'scenario':'cabinet_chain_test'})
|
|
self.assertFalse(self.poll('cabinet_01')['commands'])
|
|
self.assertFalse(self.poll('cabinet_02')['commands'])
|
|
|
|
def test_restart_preserves_gate_and_deduplicates_auto_start(self):
|
|
self.prepare()
|
|
self.win()
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
path=Path(temp)/'chain.db'
|
|
fleet=server.Fleet(path,self.catalog,lambda:self.now)
|
|
for key,room in fleet.rooms.items():
|
|
room.state=self.fleet.rooms[key].state.copy()
|
|
room.save()
|
|
room.db.close()
|
|
fleet=server.Fleet(path,self.catalog,lambda:self.now)
|
|
try:
|
|
target=fleet.rooms['cabinet_02']
|
|
command=next(c for c in target.state['commands'] if c['type']=='prepare')
|
|
report=dict(kind='ack',id=command['id'],session=command['session'],ok=True)
|
|
payload=dict(cabinet='cabinet_02',protocol=2,boot='cabinet_02',reports=[report,report])
|
|
fleet.poll(payload)
|
|
self.assertEqual(sum(c['type']=='start' for c in target.state['commands']),1)
|
|
fleet.poll(payload)
|
|
self.assertEqual(sum(c['type']=='start' for c in target.state['commands']),1)
|
|
finally:
|
|
for room in fleet.rooms.values(): room.db.close()
|