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_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()
def test_before_after_transition_listeners(self):
m = Machine(Stuff(), states=['A', 'B', 'C'], initial='A')
m.add_transition('move', 'A', 'B')
m.add_transition('move', 'B', 'C')
m.before_move('increase_level')
m.model.move()
self.assertEqual(m.model.level, 2)
m.model.move()
self.assertEqual(m.model.level, 3)
def test_get_transitions(self):
states = ['A', 'B', 'C', 'D']
m = Machine('self', states, initial='a', auto_transitions=False)
m.add_transition('go', ['A', 'B', 'C'], 'D')
m.add_transition('run', 'A', 'D')
self.assertEqual(
{(t.source, t.dest) for t in m.get_transitions('go')},
{('A', 'D'), ('B', 'D'), ('C', 'D')})
self.assertEqual(
[(t.source, t.dest)
for t in m.get_transitions(source='A', dest='D')],
[('A', 'D'), ('A', 'D')])
def test_ordered_transition_error(self):
m = Machine(states=['A'], initial='A')
with self.assertRaises(ValueError):
m.add_ordered_transitions()
m.add_state('B')
m.add_ordered_transitions()
m.add_state('C')
with self.assertRaises(ValueError):
m.add_ordered_transitions(['C'])
def test_transition_definitions(self):
states = ['A', 'B', 'C', 'D']
# Define with list of dictionaries
transitions = [
{'trigger': 'walk', 'source': 'A', 'dest': 'B'},
{'trigger': 'run', 'source': 'B', 'dest': 'C'},
{'trigger': 'sprint', 'source': 'C', 'dest': 'D'}
]
m = Machine(states=states, transitions=transitions, initial='A')
m.walk()
self.assertEqual(m.state, 'B')
# Define with list of lists
transitions = [
['walk', 'A', 'B'],
['run', 'B', 'C'],
['sprint', 'C', 'D']
]
m = Machine(states=states, transitions=transitions, initial='A')
m.to_C()
m.sprint()
self.assertEqual(m.state, 'D')
def test_multiple_models(self):
s1, s2 = Stuff(), Stuff()
states = ['A', 'B', 'C']
m = Machine(model=[s1, s2], states=states,
initial=states[0])
self.assertEqual(len(m.models), 2)
self.assertEqual(len(m.model), 2)
m.add_transition('advance', 'A', 'B')
s1.advance()
self.assertEqual(s1.state, 'B')
self.assertEqual(s2.state, 'A')
m = Machine(model=s1, states=states,
initial=states[0])
# for backwards compatibility model should return a model instance
# rather than a list
self.assertNotIsInstance(m.model, list)
# self._imgdw.local_dir = local_dir
self._imgdw = image_download
# self._imgdw.state = ImageDownload.DOWNLOAD_UNKNOWN # voltha_pb2
self._download_task_cls = download_task_cls
self._timeout_delay = timeout_delay
self._current_task = None
self._task_deferred = None
self._ret_deferred = None
self._timeout_dc = None # DelayedCall
self._advertise_events = advertise_events
self.reactor = clock if clock is not None else reactor
self.log.debug("ImageDownloadeSTM", image_download=self._imgdw)
self.machine = Machine(model=self, states=states,
transitions=transitions,
initial=initial_state,
queued=True,
name='{}-{}'.format(self.__class__.__name__, self._imgdw.id))
# @property
def __init__(self, config: DictConfig, init_state: str, states: list, transitions: (list, dict), name='pipeline'):
self.config = config
self.finite_state_machine = Machine(model=self, transitions=transitions, states=states, initial=init_state)
self.total_test_samples = 0
self.total_train_samples = 0
self.recorder = Recorder(flush_by_split_status=False)
self.status_collector = StatusCollector()
Basic.__init__(self, name=name, status=StatusWithSingleInfo(obj=self))
for state_name in self.STATE_LIST:
if not hasattr(self, 'on_enter_{}'.format(state_name)):
raise AssertionError('{} method is missed'.format('on_enter_{}'.format(state_name)))
if not hasattr(self, 'on_exit_{}'.format(state_name)):
raise AssertionError('{} method is missed'.format('on_exit_{}'.format(state_name)))
def __init__(self):
self.args = self.parse_arguments()
self.tag_type = RfidHid.TAG_T5577 if self.args.t5577 else RfidHid.TAG_EM4305
self.rfid = self.connect(self.args.usb_vid, self.args.usb_pid)
self.machine = Machine(
model=self, states=self.states, initial='start', send_event=True)
# Initialize Device
self.machine.add_transition(
trigger='next', source='start', dest='init', after='initialize', conditions=['is_init'])
self.machine.add_transition(
trigger='next', source='init', dest='exit', after='exit')
# Read Tag
self.machine.add_transition(
trigger='next', source='start', dest='read', after='read', conditions=['is_read'])
self.machine.add_transition(
trigger='next', source='read', dest='print', after='print', conditions=['is_read'])
self.machine.add_transition(
trigger='next', source='print', dest='exit', after='exit', conditions=['is_read'], unless=['is_loop'])
self.machine.add_transition(