Commits

Ian George committed 453059c

Massive refactor, machine is now a field rather than an abstract model

  • Participants
  • Parent commits 34424db

Comments (0)

Files changed (5)

File statemachine/fields.py

+from copy import deepcopy
+from django.db import models
+from fsm import FSM_State, FSM
+
+class FSM_StateField(models.Field):
+    __metaclass__ = models.SubfieldBase
+
+    def __init__(self, machine, *args, **kwargs):
+        """
+        machine must be a class, not an instance
+        """
+        self.__base_machine = machine
+        kwargs['max_length'] = 50
+        kwargs['default'] = "start"
+        super(FSM_StateField, self).__init__(self, *args, **kwargs)
+
+    def db_type(self, connection):
+        return "char(50)"
+
+    def to_python(self, value):
+        self._machine = deepcopy(self.__base_machine)
+        self._machine.set_initial_state(value)
+        return self._machine
+
+    def get_db_prep_value(self, value, connection, prepared=False):
+        return self._machine.state

File statemachine/fsm.py

+from django.conf import settings
+
+"""
+Some Exceptions
+"""
 class FSM_Exception(Exception): pass
 class FSM_TransitionNotAllowed(FSM_Exception): pass
 class FSM_StateDoesNotExist(FSM_Exception): pass
+class FSM_NotAllowed(FSM_Exception): pass
+
 class FSM_VerificationError(FSM_Exception):
     def __init__(self, message, states=None):
         self.message = message
         else:
             return self.message
 
+"""
+Django settings
+"""
+try:
+    STATE_MACHINE_KEEP_HISTORY = settings.STATE_MACHINE_KEEP_HISTORY
+except AttributeError:
+    STATE_MACHINE_KEEP_HISTORY = True
+try:
+    STATE_MACHINE_DEFAULT = settings.STATE_MACHINE_DEFAULT
+except AttributeError:
+    STATE_MACHINE_DEFAULT = "start"
+
+
 class FSM_State(object):
     """
     Represents each individual state of the machine. 
         self.entry_action = entry_action
         self.exit_action = exit_action
 
+
     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, *args, **kwargs)
 
     def __init__(self, initial_state, verify_on_execute=True):
         self.states = {}
         self.add(initial_state)
-        self.state = initial_state.name
+        self.__state = initial_state.name
         self.dbg = None
         self.verify_on_execute = verify_on_execute
+
+    def getstate(self):
+        return self.__state
+
+    def setstate(self, value):
+        raise FSM_NotAllowed("State is read only, please use execute() to change state")
+
+    state = property(getstate, setstate)
+
+    def set_initial_state(self, state):
+        self.__state = state
         
     def verify(self):
         """
             raise FSM_VerificationError("Invalid exit state(s)", bad_states)
 
     def get_django_choices(self):
-        return tuple([(name,name) for name in self.states.keys()])
+        state_choices = []
+        if self.__state:
+            state_choices.append((self.state,self.state))
+            state_choices.extend([(name,name) for name in self.states[self.state].exit_states])
+
+        return state_choices
 
     def add(self, fsm_state):
         """
         """
         self.states[fsm_state.name] = fsm_state
 
-    def execute(self, new_state, *args, **kwargs):
+    def change(self, new_state, *args, **kwargs):
         """
         Transitions the machine to its new state.
         """
         #can enter
 
         exiting_state.exit(entering_state, *args, **kwargs)
-
-        if exiting_state.exit_action:
-            exiting_state.exit_action(new_state, *args, **kwargs)
-        
         entering_state.enter(entering_state, *args, **kwargs)
 
-        self.state = new_state
+        self.__state = new_state
 
         return self.state

File statemachine/models.py

 from django.conf import settings
 
 from fsm import FSM_Exception, FSM
-
-
-##################################################
-# Get settings or defaults from django settings
-##################################################
-try:
-    STATE_MACHINE_KEEP_HISTORY = settings.STATE_MACHINE_KEEP_HISTORY
-except AttributeError:
-    STATE_MACHINE_KEEP_HISTORY = True
-try:
-    STATE_MACHINE_DEFAULT = settings.STATE_MACHINE_DEFAULT
-except AttributeError:
-    STATE_MACHINE_DEFAULT = "start"
-    
-class StateMachineNotAllowed(FSM_Exception): pass
-class StateMachineMissing(FSM_Exception): pass
-
+from fields import FSM_StateField
+from widgets import FSM_StateChoice
 
 ##################################################
 # Models
     #optional notes
     notes = models.CharField(max_length=200, null=True, blank=True)
 
-class StateMachineBase(models.Model):
+class StateMachineWithHistory(models.Model):
     """
     Abstract base class for a model requiring a state machine field
+    Class needs a FSM_StateField called status
     """
-    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)
+        super(StateMachineWithHistory, self).__init__(*args, **kwargs)
+        self._opening_state = self.status.state
 
-        # if we don't have a machine to work with, raise an exception
-        if self.__machine_class__:
-            self.__machine__ = deepcopy(self.__machine_class__)
-        else:
-            raise StateMachineMissing("Instance of fsm.FSM required for this model to function (set __machine_class__ on the model)")
+    def save(self, user, notes='', *args, **kwargs):
+        super(StateMachineWithHistory, self).save(*args, **kwargs)
+        #log transition - not sure if we should do this here if we're not saving
+        s = StateMachineHistory()
+        s.machine = self
+        s.from_state = self._opening_state
+        s.to_state = self.status.state
+        s.user = user
+        s.notes = notes
+        s.save()
 
-        self.__machine__.state = self.state
-        self._current_state = self.state
-
-        self._meta.get_field_by_name("state")[0]._choices = self.__machine__.get_django_choices()
-        super(StateMachineBase, self).__init__(*args, **kwargs)
-
-
-    def save(self, *args, **kwargs):
-        if self.state == self._current_state:
-            super(StateMachineBase, self).save(*args, **kwargs)
-        else:
-            raise StateMachineNotAllowed("To change state, use the transition method")
-        
-    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 self.save_model:
-            super(StateMachineBase, self).save()
-        
-        #log transition - not sure if we should do this here if we're not saving
-        if STATE_MACHINE_KEEP_HISTORY:
-            s = StateMachineHistory()
-            s.machine = self
-            s.from_state = from_state
-            s.to_state = self.state
-            s.user = user
-            s.notes = notes
-            s.save()
-
-        #reset the current state 
-        self.__current_state = self.state
 
     class Meta:
         abstract = True

File statemachine/tests.py

 from django.test import TestCase
 from django.test.client import Client
 
-from models import StateMachineBase, StateMachineNotAllowed
-from fsm import FSM, FSM_State, FSM_TransitionNotAllowed, FSM_VerificationError
-
-def can_fsm_end(state_from, state_to, input=None):
-    if state_to == "end" and state_from == "step3":
-        return True
+from models import StateMachineWithHistory
+from fsm import FSM, FSM_State, FSM_NotAllowed, FSM_TransitionNotAllowed, FSM_VerificationError
+from fields import FSM_StateField
 
 testfsm = FSM(FSM_State("start" ,["step1a", "step1b", "step1c"]))
 testfsm.add(FSM_State("step1a", ["step2",]))
 testfsm.add(FSM_State("step3", ["end",]))
 testfsm.add(FSM_State("end", []))
 
-class TestStateMachine(StateMachineBase):
-    __machine_class__ = testfsm
+class TestStateMachine(StateMachineWithHistory):
+    status = FSM_StateField(testfsm)
     field1 = models.IntegerField()
     field2 = models.CharField(max_length=25)
 
 class StateMachineTests(TestCase):
-    def setUp(self):
-        self.fsm = TestStateMachine()
-        self.fsm.field1 = 100
-        self.fsm.field2 = "Test string"
-        self.fsm.save()
+    def txest_verify(self):
+        """ Should raise a FSM_VerificationError """
+        testmachine = TestStateMachine(field1=100, field2="LALALALALA")
+        testmachine.status.add(FSM_State("random", ["doesntexist", "alsodoesnexist"]))
+        with self.assertRaises(FSM_VerificationError):
+            testmachine.status.verify()
 
-    def test_verify(self):
-        """ Should raise a FSM_VerificationError """
-        
-        self.fsm.__machine__.add(FSM_State("random", ["doesntexist", "alsodoesnexist"]))
-        with self.assertRaises(FSM_VerificationError):
-            self.fsm.__machine__.verify()
-            
+    def txest_save_fail(self):
+        """Should raise a StateMachineNotAllowed, messing with the state outside of the machine."""
+        testmachine = TestStateMachine(field1=100, field2="LALALALALA")
+        testmachine.save(None)
+        with self.assertRaises(FSM_NotAllowed):
+            testmachine.status.state = "Testing"    
+            testmachine.save(None)
 
-    def test_save_fail(self):
-        """Should raise a StateMachineNotAllowed, messing with the state outside of the machine."""
-        
-        self.fsm = TestStateMachine(field1=100, field2="LALALALALA")
-        self.fsm.state = "Testing"    
-        with self.assertRaises(StateMachineNotAllowed):
-            self.fsm.save()
+    def txest_full_transition(self):
+        """ End to end smoke test of the transitions """
+        testmachine = TestStateMachine(field1=100, field2="LALALALALA")
+        testmachine.save(None)
+        testmachine.status.change("step1a")
+        testmachine.save(None)
+        testmachine.status.change("step2")
+        testmachine.save(None)
+        #reload to check state is saved and reinstated
+        self.assertEqual(testmachine.status.state, "step2")
+        fsm_id = testmachine.id
+        self.assertEqual(testmachine.status.state, "step2")
+        testmachine = TestStateMachine.objects.get(pk=fsm_id)
+        testmachine.status.change("step3")
+        testmachine.save(None)
+        testmachine.status.change("end")
+        testmachine.save(None)
 
-    def test_full_transition(self):
-        """ End to end smoke test of the transitions """
-        self.fsm.transition("step1a", None)
-        self.fsm.transition("step2", None)
-        #reload to check state is saved and reinstated
-        self.assertEqual(self.fsm.state, "step2")
-        fsm_id = self.fsm.id
-        self.assertEqual(self.fsm.state, "step2")
-        self.fsm = TestStateMachine.objects.get(pk=fsm_id)
-        self.fsm.transition("step3", None)
-        self.fsm.transition("end", None)
-
-    def test_invalid_transition(self):
+    def txest_invalid_transition(self):
         """ Testing invalid transitions """
-
+        testmachine = TestStateMachine(field1=100, field2="LALALALALA")
+        testmachine.save(None)
         #check we are starting at "start"
-        self.assertEqual(self.fsm.state, "start")
+        self.assertEqual(testmachine.status.state, "start")
         
         with self.assertRaises(FSM_TransitionNotAllowed):
-            self.fsm.transition("step3", None)
+            testmachine.status.change("step3")
 
         #check we are still at "start" after a failed transition
-        self.assertEqual(self.fsm.state, "start")
+        self.assertEqual(testmachine.status.state, "start")
 
-    def test_transition_history(self):
+    def txest_transition_history(self):
         """ 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 , "Some notes!")
-        self.fsm.transition("end", None)
+        testmachine = TestStateMachine(field1=100, field2="LALALALALA")
+        self.assertEqual(testmachine.status.state, "start")
+        testmachine.status.change("step1a")
+        testmachine.save(None)
+        testmachine.status.change("step2")
+        testmachine.save(None, "Some notes!")
+        testmachine.status.change("end")
+        testmachine.save(None)
 
         #reload so we get everything from the db
-        fsm_id = self.fsm.id
-        self.fsm = TestStateMachine.objects.get(pk=fsm_id)
+        fsm_id = testmachine.id
+        testmachine = TestStateMachine.objects.get(pk=fsm_id)
 
-        self.assertEqual(self.fsm.state_history.count(), 3)
-        self.assertEqual(self.fsm.state_history.all()[1].notes, "Some notes!")
+        self.assertEqual(testmachine.state_history.count(), 3)
+        self.assertEqual(testmachine.state_history.all()[1].notes, "Some notes!")
 
-    def test_entry_action(self):
+    def txest_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)
+        testmachine = TestStateMachine(field1=100, field2="LALALALALA")
+        testmachine.save(None)
+        testmachine.status.states["step3"].entry_action = entry_func
+        testmachine.status.change("step1a")
+        testmachine.save(None)
+        testmachine.status.change("step2")
+        testmachine.save(None)
+        testmachine.status.change("step3", testmachine)
+        testmachine.save(None)
         
-        self.assertEqual(self.fsm.field1, 500)
+        self.assertEqual(testmachine.field1, 500)
 
     def test_exit_action(self):
         """ Tests an action set to run on entry to a state """
             row.field1 = 10
             row.field2 = "Chickens"
 
-        self.fsm.__machine__.states["start"].exit_action = exit_func
-        self.fsm.transition("step1a", None)
+        testmachine = TestStateMachine(field1=100, field2="LALALALALA")
+        testmachine.save(None)
+        testmachine.status.states["start"].exit_action = exit_func
+        testmachine.status.change("step1a", testmachine)
 
-        self.assertEqual(self.fsm.field1, 10)
-        self.assertEqual(self.fsm.field2, "Chickens")
+        self.assertEqual(testmachine.field1, 10)
+        self.assertEqual(testmachine.field2, "Chickens")

File statemachine/widgets.py

+from django import forms
+
+class FSM_StateChoice(forms.Select):
+    pass
+        
+