Allow for raising on duplicate transition registration
Instead of always by default overwriting an existing transition from a state to another state on a given event make it possible to avoid that overwriting (and raise a duplication exception instead). Change-Id: I7f5c1abf3b8737b71033250f7754595fbb28589a
This commit is contained in:
parent
725f0aa4b0
commit
2276f53661
automaton
@ -188,8 +188,17 @@ class FiniteMachine(object):
|
||||
raise excp.Duplicate("State '%s' reaction to event '%s'"
|
||||
" already defined" % (state, event))
|
||||
|
||||
def add_transition(self, start, end, event):
|
||||
"""Adds an allowed transition from start -> end for the given event."""
|
||||
def add_transition(self, start, end, event, replace=False):
|
||||
"""Adds an allowed transition from start -> end for the given event.
|
||||
|
||||
:param start: starting state
|
||||
:param end: ending state
|
||||
:param event: event that causes start state to
|
||||
transition to end state
|
||||
:param replace: replace existing event instead of raising a
|
||||
:py:class:`~automaton.exceptions.Duplicate` exception
|
||||
when the transition already exists.
|
||||
"""
|
||||
if self.frozen:
|
||||
raise excp.FrozenMachine()
|
||||
if start not in self._states:
|
||||
@ -204,9 +213,22 @@ class FiniteMachine(object):
|
||||
raise excp.InvalidState("Can not add a transition on event '%s'"
|
||||
" that starts in the terminal state '%s'"
|
||||
% (event, start))
|
||||
self._transitions[start][event] = _Jump(end,
|
||||
self._states[end]['on_enter'],
|
||||
self._states[start]['on_exit'])
|
||||
if event in self._transitions[start] and not replace:
|
||||
target = self._transitions[start][event]
|
||||
if target.name != end:
|
||||
raise excp.Duplicate("Cannot add transition from"
|
||||
" '%(start_state)s' to '%(end_state)s'"
|
||||
" on event '%(event)s' because a"
|
||||
" transition from '%(start_state)s'"
|
||||
" to '%(existing_end_state)s' on"
|
||||
" event '%(event)s' already exists."
|
||||
% {'existing_end_state': target.name,
|
||||
'end_state': end, 'event': event,
|
||||
'start_state': start})
|
||||
else:
|
||||
target = _Jump(end, self._states[end]['on_enter'],
|
||||
self._states[start]['on_exit'])
|
||||
self._transitions[start][event] = target
|
||||
|
||||
def _pre_process_event(self, event):
|
||||
current = self._current
|
||||
|
@ -75,6 +75,21 @@ class FSMTest(testcase.TestCase):
|
||||
m = self._create_fsm('unknown')
|
||||
self.assertRaises(excp.Duplicate, m.add_state, 'unknown')
|
||||
|
||||
def test_duplicate_transition(self):
|
||||
m = self.jumper
|
||||
m.add_state('side_ways')
|
||||
self.assertRaises(excp.Duplicate,
|
||||
m.add_transition, 'up', 'side_ways', 'fall')
|
||||
|
||||
def test_duplicate_transition_replace(self):
|
||||
m = self.jumper
|
||||
m.add_state('side_ways')
|
||||
m.add_transition('up', 'side_ways', 'fall', replace=True)
|
||||
|
||||
def test_duplicate_transition_same_transition(self):
|
||||
m = self.jumper
|
||||
m.add_transition('up', 'down', 'fall')
|
||||
|
||||
def test_duplicate_reaction(self):
|
||||
self.assertRaises(
|
||||
# Currently duplicate reactions are not allowed...
|
||||
|
Loading…
x
Reference in New Issue
Block a user