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_state_tags(self):
@add_state_features(Tags, Timeout)
class CustomMachine(self.machine_cls):
pass
self.states[0] = {'name': 'A', 'tags': ['new', 'polling'], 'timeout': 5, 'on_enter': 'say_hello',
'on_exit': 'say_goodbye', 'on_timeout': 'do_something'}
m = CustomMachine(states=self.states, transitions=self.transitions, initial='A', show_state_attributes=True)
g = m.get_graph(show_roi=True)
def test_state_tags(self):
@add_state_features(Tags, Timeout)
class CustomMachine(self.machine_cls):
pass
self.states[0] = {'name': 'A', 'tags': ['new', 'polling'], 'timeout': 5, 'on_enter': 'say_hello',
'on_exit': 'say_goodbye', 'on_timeout': 'do_something'}
m = CustomMachine(states=self.states, transitions=self.transitions, initial='A', show_state_attributes=True,
use_pygraphviz=self.use_pygraphviz)
g = m.get_graph(show_roi=True)
def __init__(self, *args, **kwargs):
"""
Args:
**kwargs: If kwargs contains `tags`, assign them to the attribute.
"""
self.tags = kwargs.pop('tags', [])
super(Tags, self).__init__(*args, **kwargs)
"""
def __init__(self, *args, **kwargs):
"""
Args:
**kwargs: If kwargs contains `tags`, assign them to the attribute.
"""
self.tags = kwargs.pop('tags', [])
super(Tags, self).__init__(*args, **kwargs)
def __getattr__(self, item):
if item.startswith('is_'):
return item[3:] in self.tags
return super(Tags, self).__getattribute__(item)
class Error(Tags):
""" This mix in builds upon tag and should be used INSTEAD of Tags if final states that have
not been tagged with 'accepted' should throw an `MachineError`.
"""
def __init__(self, *args, **kwargs):
"""
Args:
**kwargs: If kwargs contains the keywork `accepted` add the 'accepted' tag to a tag list
which will be forwarded to the Tags constructor.
"""
tags = kwargs.get('tags', [])
accepted = kwargs.pop('accepted', False)
if accepted:
tags.append('accepted')
kwargs['tags'] = tags
super(Error, self).__init__(*args, **kwargs)
self.logger.debug(f"Checking {state_module}")
on_enter_method = getattr(state_module, 'on_enter')
setattr(self, f'on_enter_{state}', on_enter_method)
self.logger.trace(f"Added `on_enter` method from {state_module} {on_enter_method}")
if state_info is None:
state_info = dict()
# Add horizon if state requires.
with suppress(KeyError):
self._horizon_lookup[state] = state_info['horizon']
del state_info['horizon']
self.logger.debug(f"Creating {state=} with {state_info=}")
state_machine = MachineState(name=state, **state_info)
# Add default callbacks.
state_machine.add_callback('enter', '_update_status')
state_machine.add_callback('enter', f'on_enter_{state}')
except Exception as e:
raise error.InvalidConfig(f"Can't load state modules: {state}\t{e!r}")
return state_machine