Commits

Ian George  committed fb23017

Added tests for state change actions we are now passing args to state change actions through execute

  • Participants
  • Parent commits 580afbd

Comments (0)

Files changed (3)

File statemachine/fsm.py

     """
     Represents each individual state of the machine. 
 
-    exit_states: a list of strings representing allowed transitions
-    from this state
+    exit_states: 
+        a list of strings representing allowed transitions from this state
 
-    entry_action: a function to run on entry into the state. Args:
-    exited_state, model
+    entry_action(exited_state, model): 
+        a function to run on entry into the state. 
 
-    exit action: a function to run on exit from the state, must return
-    FSM_TransitionNotAllowed if conditions for the transition are not
-    met. Args: target_state, model
+    exit action(target_state, model): 
+        a function to run on exit from the state, must return
+        FSM_TransitionNotAllowed if conditions for the transition are not
+        met. 
+
     """
     name = ""
     exit_states = []
     entry_action = None
     exit_action = None
 
-    def __init__(self, name, exit_states, entry_action=None, exit_action=None, condition=None):
+    def __init__(self, name, exit_states, entry_action=None, exit_action=None):
         self.name = name
         self.exit_states = exit_states
         self.entry_action = entry_action
         self.exit_action = exit_action
 
-    def exit(self, target_state, model):
+    def exit(self, target_state, *args, **kwargs):
+        """ 
+        Checks states and exits if possible, if not possible it raises FSM_TransitionNotAllowed 
+        """
         if not target_state in self.exit_states:
             return FSM_TransitionNotAllowed("Target state not permitted from this state")
 
         if self.exit_action:
-            self.exit_action(target_state, model)
+            self.exit_action(target_state, *args, **kwargs)
 
         return True
 
-    def enter(self, exited_state, model):
+    def enter(self, exited_state, *args, **kwargs):
+        """
+        Runs an entry action if it is set
+        """
         if self.entry_action:
-            return self.entry_action(exited_state)
+            return self.entry_action(exited_state, *args, **kwargs)
         
 class FSM(object):
     """
         self.dbg = None
 
     def verify(self):
+        """
+        Best only called on fully formed machines, checks for unreachable states
+        """
         bad_states = []
         state_names = set(self.states.keys())
         for name, state in self.states.items():
             raise FSM_VerificationError(bad_states)
 
     def add(self, fsm_state):
+        """
+        Adds a FSM_State to the machine
+        """
         self.states[fsm_state.name] = fsm_state
 
-    def execute(self, new_state, model):
+    def execute(self, new_state, *args, **kwargs):
+        """
+        Transitions the machine to its new state.
+        """
         if not new_state in self.states:
             raise FSM_StateDoesNotExist(new_state)
 
         #TODO: this needs to be transactional, shouldn't exit unless it
         #can enter
 
-        exiting_state.exit(entering_state, model)
+        exiting_state.exit(entering_state, *args, **kwargs)
 
         if exiting_state.exit_action:
-            exiting_state.exit_action(new_state, model)
+            exiting_state.exit_action(new_state, *args, **kwargs)
         
-        entering_state.enter(entering_state, model)
+        entering_state.enter(entering_state, *args, **kwargs)
 
         self.state = new_state
 

File statemachine/models.py

 
 
 ##################################################
-# Get settings or defaults from django settings
+# Models
 ##################################################
 class StateMachineHistory(models.Model):
     """
     state = models.CharField(max_length=25, default="start")    
     state_history = generic.GenericRelation(StateMachineHistory, content_type_field='machine_type', object_id_field='machine_id')
     _current_state = ''
+    save_model = True
 
     def __init__(self, *args, **kwargs):
         super(StateMachineBase, self).__init__(*args, **kwargs)
 
         self._current_state = self.state
 
-    def save(self):
+    def save(self, *args, **kwargs):
         if self.state == self._current_state:
-            super(StateMachineBase, self).save()
+            super(StateMachineBase, self).save(*args, **kwargs)
         else:
             raise StateMachineNotAllowed("To change state, use the transition method")
         
-    def transition(self, transition_to, user, notes='', save_model=True):
+    def transition(self, transition_to, user, notes='', *args, **kwargs):
         #transition this model
         from_state = self.state
         self.state = self.__machine__.execute(transition_to, self)
 
         # we can allow save as a separate operation
-        if save_model:
+        if self.save_model:
             super(StateMachineBase, self).save()
         
         #log transition - not sure if we should do this here if we're not saving

File statemachine/tests.py

         """ Checking that history is being saved for each transition """
         self.assertEqual(self.fsm.state, "start")
         self.fsm.transition("step1a", None)
-        self.fsm.transition("step2", None)
+        self.fsm.transition("step2", None , "Some notes!")
         self.fsm.transition("end", None)
 
-        for item in self.fsm.state_history.all():
-            print "%s: %s - %s" % (item.change_date, item.from_state, item.to_state)
+        self.assertEqual(self.fsm.state_history.count(), 3)
+
+        self.assertEqual(self.fsm.state_history.all()[1].notes, "Some notes!")
+
+    def test_entry_action(self):
+        """ Tests an action set to run on entry to a state """
+        def entry_func(entered, row):
+            row.field1 = 500
+
+        self.fsm.__machine__.states["step3"].entry_action = entry_func
+        self.fsm.transition("step1a", None)
+        self.fsm.transition("step2", None)
+        self.fsm.transition("step3", None)
+        
+        self.assertEqual(self.fsm.field1, 500)
+
+    def test_exit_action(self):
+        """ Tests an action set to run on entry to a state """
+        def exit_func(entered, row):
+            row.field1 = 10
+            row.field2 = "Chickens"
+
+        self.fsm.__machine__.states["start"].exit_action = exit_func
+        self.fsm.transition("step1a", None)
+
+        self.assertEqual(self.fsm.field1, 10)
+        self.assertEqual(self.fsm.field2, "Chickens")