Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_pass_state_instances_instead_of_names(self):
state_A = State('A')
state_B = State('B')
states = [state_A, state_B]
m = Machine(states=states, initial=state_A)
assert m.state == 'A'
m.add_transition('advance', state_A, state_B)
m.advance()
assert m.state == 'B'
state_B2 = State('B', on_enter='this_passes')
with self.assertRaises(ValueError):
m.add_transition('advance2', state_A, state_B2)
m2 = Machine(states=states, initial=state_A.name)
assert m.initial == m2.initial
with self.assertRaises(ValueError):
Machine(states=states, initial=State('A'))
def test_init_machine_with_hella_arguments(self):
states = [
State('State1'),
'State2',
{
'name': 'State3',
'on_enter': 'hello_world'
}
]
transitions = [
{'trigger': 'advance',
'source': 'State2',
'dest': 'State3'
}
]
s = Stuff()
m = Machine(model=s, states=states, transitions=transitions, initial='State2')
s.advance()
self.assertEqual(s.message, 'Hello World!')
def test_state_callable_callbacks(self):
class Model:
def __init__(self):
self.exit_A_called = False
self.exit_B_called = False
def on_enter_A(self, event):
pass
def on_enter_B(self, event):
pass
states = [State(name='A', on_enter='on_enter_A', on_exit='tests.test_core.on_exit_A'),
State(name='B', on_enter='on_enter_B', on_exit=on_exit_B),
State(name='C', on_enter='tests.test_core.AAAA')]
model = Model()
machine = Machine(model, states=states, send_event=True, initial='A')
state_a = machine.get_state('A')
state_b = machine.get_state('B')
self.assertEqual(len(state_a.on_enter), 1)
self.assertEqual(len(state_a.on_exit), 1)
self.assertEqual(len(state_b.on_enter), 1)
self.assertEqual(len(state_b.on_exit), 1)
model.to_B()
self.assertTrue(model.exit_A_called)
model.to_A()
self.assertTrue(model.exit_B_called)
with self.assertRaises(AttributeError):
def test_state_callbacks(self):
class Model:
def on_enter_A(self):
pass
def on_exit_A(self):
pass
def on_enter_B(self):
pass
def on_exit_B(self):
pass
states = [State(name='A', on_enter='on_enter_A', on_exit='on_exit_A'),
State(name='B', on_enter='on_enter_B', on_exit='on_exit_B')]
machine = Machine(Model(), states=states)
state_a = machine.get_state('A')
state_b = machine.get_state('B')
self.assertEqual(len(state_a.on_enter), 1)
self.assertEqual(len(state_a.on_exit), 1)
self.assertEqual(len(state_b.on_enter), 1)
self.assertEqual(len(state_b.on_exit), 1)
def test_pass_state_instances_instead_of_names(self):
state_A = State('A')
state_B = State('B')
states = [state_A, state_B]
m = Machine(states=states, initial=state_A)
assert m.state == 'A'
m.add_transition('advance', state_A, state_B)
m.advance()
assert m.state == 'B'
state_B2 = State('B', on_enter='this_passes')
with self.assertRaises(ValueError):
m.add_transition('advance2', state_A, state_B2)
m2 = Machine(states=states, initial=state_A.name)
assert m.initial == m2.initial
with self.assertRaises(ValueError):
Machine(states=states, initial=State('A'))
def test_pass_state_instances_instead_of_names(self):
state_A = State('A')
state_B = State('B')
states = [state_A, state_B]
m = Machine(states=states, initial=state_A)
assert m.state == 'A'
m.add_transition('advance', state_A, state_B)
m.advance()
assert m.state == 'B'
state_B2 = State('B', on_enter='this_passes')
with self.assertRaises(ValueError):
m.add_transition('advance2', state_A, state_B2)
m2 = Machine(states=states, initial=state_A.name)
assert m.initial == m2.initial
with self.assertRaises(ValueError):
Machine(states=states, initial=State('A'))
def test_pass_state_instances_instead_of_names(self):
state_A = State('A')
state_B = State('B')
states = [state_A, state_B]
m = Machine(states=states, initial=state_A)
assert m.state == 'A'
m.add_transition('advance', state_A, state_B)
m.advance()
assert m.state == 'B'
state_B2 = State('B', on_enter='this_passes')
with self.assertRaises(ValueError):
m.add_transition('advance2', state_A, state_B2)
m2 = Machine(states=states, initial=state_A.name)
assert m.initial == m2.initial
with self.assertRaises(ValueError):
Machine(states=states, initial=State('A'))
def test_auto_transitions(self):
states = ['A', {'name': 'B'}, State(name='C')]
m = Machine('self', states, initial='A', auto_transitions=True)
m.to_B()
self.assertEqual(m.state, 'B')
m.to_C()
self.assertEqual(m.state, 'C')
m.to_A()
self.assertEqual(m.state, 'A')
# Should fail if auto transitions is off...
m = Machine('self', states, initial='A', auto_transitions=False)
with self.assertRaises(AttributeError):
m.to_C()
from loopchain import configure as conf
from loopchain.blockchain import UnrecordedBlock, InvalidUnconfirmedBlock
from loopchain.blockchain.blocks import Block
from loopchain.peer import status_code
from loopchain.protos import loopchain_pb2
from loopchain.statemachine import statemachine
from loopchain.utils import loggers
@statemachine.StateMachine("Channel State Machine")
class ChannelStateMachine(object):
states = ['InitComponents',
State(name='Consensus',
ignore_invalid_triggers=True,
on_enter='_consensus_on_enter'),
State(name='BlockHeightSync',
ignore_invalid_triggers=True,
on_enter='_blockheightsync_on_enter'),
'EvaluateNetwork',
State(name='BlockSync',
ignore_invalid_triggers=True,
on_enter='_blocksync_on_enter',
on_exit='_blocksync_on_exit'),
State(name='SubscribeNetwork',
ignore_invalid_triggers=True,
on_enter='_subscribe_network_on_enter',
on_exit='_subscribe_network_on_exit'),
State(name='Watch',
ignore_invalid_triggers=True,
on_enter='_watch_on_enter',
on_exit='_watch_on_exit'),
State(name='Vote',
import time
import transitions
from panoptes.utils.logger import has_logger
@has_logger
class PanState(transitions.State):
""" Base class for PANOPTES transitions """
def __init__(self, *args, **kwargs):
name = kwargs.get('name', self.__class__)
self.panoptes = kwargs.get('panoptes', None)
super().__init__(name=name, on_enter=['execute'])
self._sleep_delay = 3 # seconds
def main(self, event_data):
assert self.panoptes is not None
msg = "Must implement `main` method inside class {}. Exiting".format(self.name)
self.panoptes.logger.warning(msg)