define with list of dictionaries define with list of dictionaries define with list of lists make sure parameters passed by trigger events can be handled state does not advance but increaselevel still runs an invalid transition shouldn t execute the callback first pass positional and keyword args directly to the callback now wrap arguments in an eventdata instance first pass positional and keyword args directly to the condition now wrap arguments in an eventdata instance should fail if auto transitions is off include initial state in loop do not loop transitions test userdetermined sequence and trigger name via init argument alter initial state partial state machine without the initial state exception is triggered by default set default value on machine level exception is suppressed so this passes set for some states but not others state value overrides machine behaviour afterstatechange should have been called with eventdata define with list of dictionaries define with list of dictionaries queue should contain tob inner tob toc inner toc since tob is currently executed it should still be in the list toc should be gone test whether models can be removed outside event queue state does not advance but increaselevel still runs an invalid transition shouldn t execute the callback for backwards compatibility model should return a model instance rather than a list self stuff machine should have totransitions to every state no transition has been assigned to eventdata yet valueerror in finalize should be suppressed but mock should have been called anyway deliberately passing wrong arguments type ignore define with list of dictionaries define with list of dictionaries type list union list str dict str str define with list of lists make sure parameters passed by trigger events can be handled state does not advance but increase_level still runs an invalid transition shouldn t execute the callback first pass positional and keyword args directly to the callback now wrap arguments in an eventdata instance first pass positional and keyword args directly to the condition now wrap arguments in an eventdata instance type list union str dict str str state should fail if auto transitions is off include initial state in loop do not loop transitions test user determined sequence and trigger name via init argument alter initial state partial state machine without the initial state exception is triggered by default set default value on machine level exception is suppressed so this passes set for some states but not others passes because exception suppressed for d state value overrides machine behaviour after_state_change should have been called with eventdata define with list of dictionaries define with list of dictionaries queue should contain to_b inner to_b to_c inner to_c since to_b is currently executed it should still be in the list to_c should be gone test whether models can be removed outside event queue state does not advance but increase_level still runs an invalid transition shouldn t execute the callback for backwards compatibility model should return a model instance rather than a list self stuff machine should have to transitions to every state no transition has been assigned to eventdata yet valueerror in finalize should be suppressed but mock should have been called anyway deliberately passing wrong arguments type ignore
try:
    from builtins import object
except ImportError:
    pass

import sys
from typing import TYPE_CHECKING
from functools import partial
from unittest import TestCase, skipIf
import weakref

from transitions import Machine, MachineError, State, EventData
from transitions.core import listify, _prep_ordered_arg

from .utils import InheritedStuff
from .utils import Stuff, DummyModel


try:
    from unittest.mock import MagicMock
except ImportError:
    from mock import MagicMock  

if TYPE_CHECKING:
    from typing import List, Union, Dict, Callable


def on_exit_A(event):
    event.model.exit_A_called = True


def on_exit_B(event):
    event.model.exit_B_called = True


class TestTransitions(TestCase):

    def setUp(self):
        self.stuff = Stuff()
        self.machine_cls = Machine

    def tearDown(self):
        pass

    def test_init_machine_with_hella_arguments(self):
        states = [
            State('State1'),
            'State2',
            {
                'name': 'State3',
                'on_enter': 'hello_world'
            }
        ]
        transitions = [
            {'trigger': 'advance',
                'source': 'State2',
                'dest': 'State3'
             }
        ]
        s = Stuff()
        m = s.machine_cls(model=s, states=states, transitions=transitions, initial='State2')
        s.advance()
        self.assertEqual(s.message, 'Hello World!')

    def test_listify(self):
        self.assertEqual(listify(4), [4])
        self.assertEqual(listify(None), [])
        self.assertEqual(listify((4, 5)), (4, 5))
        self.assertEqual(listify([1, 3]), [1, 3])

        class Foo:
            pass
        obj = Foo()
        proxy = weakref.proxy(obj)
        del obj
        self.assertEqual(listify(proxy), [proxy])

    def test_weakproxy_model(self):
        d = DummyModel()
        pr = weakref.proxy(d)
        self.machine_cls(pr, states=['A', 'B'], transitions=[['go', 'A', 'B']], initial='A')
        pr.go()
        self.assertTrue(pr.is_B())

    def test_property_initial(self):
        states = ['A', 'B', 'C', 'D']
        
        transitions = [
            {'trigger': 'walk', 'source': 'A', 'dest': 'B'},
            {'trigger': 'run', 'source': 'B', 'dest': 'C'},
            {'trigger': 'sprint', 'source': 'C', 'dest': 'D'}
        ]
        m = self.stuff.machine_cls(states=states, transitions=transitions, initial='A')
        self.assertEqual(m.initial, 'A')
        m = self.stuff.machine_cls(states=states, transitions=transitions, initial='C')
        self.assertEqual(m.initial, 'C')
        m = self.stuff.machine_cls(states=states, transitions=transitions)
        self.assertEqual(m.initial, 'initial')

    def test_transition_definitions(self):
        states = ['A', 'B', 'C', 'D']
        
        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')
        
        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_add_states(self):
        s = self.stuff
        s.machine.add_state('X')
        s.machine.add_state('Y')
        s.machine.add_state('Z')
        event = s.machine.events['to_{0}'.format(s.state)]
        self.assertEqual(1, len(event.transitions['X']))

    def test_transitioning(self):
        s = self.stuff
        s.machine.add_transition('advance', 'A', 'B')
        s.machine.add_transition('advance', 'B', 'C')
        s.machine.add_transition('advance', 'C', 'D')
        s.advance()
        self.assertEqual(s.state, 'B')
        self.assertFalse(s.is_A())
        self.assertTrue(s.is_B())
        s.advance()
        self.assertEqual(s.state, 'C')

    def test_pass_state_instances_instead_of_names(self):
        state_A = State('A')
        state_B = State('B')
        states = [state_A, state_B]
        m = Machine(states=states, initial=state_A)
        assert m.state == 'A'
        m.add_transition('advance', state_A, state_B)
        m.advance()
        assert m.state == 'B'
        state_B2 = State('B', on_enter='this_passes')
        with self.assertRaises(ValueError):
            m.add_transition('advance2', state_A, state_B2)
        m2 = Machine(states=states, initial=state_A.name)
        assert m.initial == m2.initial
        with self.assertRaises(ValueError):
            Machine(states=states, initial=State('A'))

    def test_conditions(self):
        s = self.stuff
        s.machine.add_transition('advance', 'A', 'B', conditions='this_passes')
        s.machine.add_transition('advance', 'B', 'C', unless=['this_fails'])
        s.machine.add_transition('advance', 'C', 'D', unless=['this_fails',
                                                              'this_passes'])
        s.advance()
        self.assertEqual(s.state, 'B')
        s.advance()
        self.assertEqual(s.state, 'C')
        s.advance()
        self.assertEqual(s.state, 'C')

    def test_uncallable_callbacks(self):
        s = self.stuff
        s.machine.add_transition('advance', 'A', 'B', conditions=['property_that_fails', 'is_false'])
        
        s.machine.add_transition('advance', 'A', 'C', before=['property_that_fails', 'is_false'])
        s.advance(level='MaximumSpeed')
        self.assertTrue(s.is_C())

    def test_conditions_with_partial(self):
        def check(result):
            return result

        s = self.stuff
        s.machine.add_transition('advance', 'A', 'B',
                                 conditions=partial(check, True))
        s.machine.add_transition('advance', 'B', 'C',
                                 unless=[partial(check, False)])
        s.machine.add_transition('advance', 'C', 'D',
                                 unless=[partial(check, False), partial(check, True)])
        s.advance()
        self.assertEqual(s.state, 'B')
        s.advance()
        self.assertEqual(s.state, 'C')
        s.advance()
        self.assertEqual(s.state, 'C')

    def test_multiple_add_transitions_from_state(self):
        s = self.stuff
        s.machine.add_transition(
            'advance', 'A', 'B', conditions=['this_fails'])
        s.machine.add_transition('advance', 'A', 'C')
        s.advance()
        self.assertEqual(s.state, 'C')

    def test_use_machine_as_model(self):
        states = ['A', 'B', 'C', 'D']
        m = Machine(states=states, initial='A')
        m.add_transition('move', 'A', 'B')
        m.add_transition('move_to_C', 'B', 'C')
        m.move()
        self.assertEqual(m.state, 'B')

    def test_state_change_listeners(self):
        s = self.stuff
        s.machine.add_transition('advance', 'A', 'B')
        s.machine.add_transition('reverse', 'B', 'A')
        s.machine.on_enter_B('hello_world')
        s.machine.on_exit_B('goodbye')
        s.advance()
        self.assertEqual(s.state, 'B')
        self.assertEqual(s.message, 'Hello World!')
        s.reverse()
        self.assertEqual(s.state, 'A')
        self.assertTrue(s.message is not None and s.message.startswith('So long'))

    def test_before_after_callback_addition(self):
        m = Machine(Stuff(), states=['A', 'B', 'C'], initial='A')
        m.add_transition('move', 'A', 'B')
        trans = m.events['move'].transitions['A'][0]
        trans.add_callback('after', 'increase_level')
        m.model.move()
        self.assertEqual(m.model.level, 2)

    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_prepare(self):
        m = Machine(Stuff(), states=['A', 'B', 'C'], initial='A')
        m.add_transition('move', 'A', 'B', prepare='increase_level')
        m.add_transition('move', 'B', 'C', prepare='increase_level')
        m.add_transition('move', 'C', 'A', prepare='increase_level', conditions='this_fails')
        m.add_transition('dont_move', 'A', 'C', prepare='increase_level')

        m.prepare_move('increase_level')

        m.model.move()
        self.assertEqual(m.model.state, 'B')
        self.assertEqual(m.model.level, 3)

        m.model.move()
        self.assertEqual(m.model.state, 'C')
        self.assertEqual(m.model.level, 5)

        
        m.model.move()
        self.assertEqual(m.model.state, 'C')
        self.assertEqual(m.model.level, 7)

        
        try:
            m.model.dont_move()
        except MachineError as e:
            self.assertTrue("Can't trigger event" in str(e))

        self.assertEqual(m.model.state, 'C')
        self.assertEqual(m.model.level, 7)

    def test_state_model_change_listeners(self):
        s = self.stuff
        s.machine.add_transition('go_e', 'A', 'E')
        s.machine.add_transition('go_f', 'E', 'F')
        s.machine.on_enter_F('hello_F')
        s.go_e()
        self.assertEqual(s.state, 'E')
        self.assertEqual(s.message, 'I am E!')
        s.go_f()
        self.assertEqual(s.state, 'F')
        self.assertEqual(s.exit_message, 'E go home...')
        self.assertIn('I am F!', s.message or "")
        self.assertIn('Hello F!', s.message or "")

    def test_inheritance(self):
        states = ['A', 'B', 'C', 'D', 'E']
        s = InheritedStuff(states=states, initial='A')
        s.add_transition('advance', 'A', 'B', conditions='this_passes')
        s.add_transition('advance', 'B', 'C')
        s.add_transition('advance', 'C', 'D')
        s.advance()
        self.assertEqual(s.state, 'B')
        self.assertFalse(s.is_A())
        self.assertTrue(s.is_B())
        s.advance()
        self.assertEqual(s.state, 'C')

        class NewMachine(Machine):
            def __init__(self, *args, **kwargs):
                super(NewMachine, self).__init__(*args, **kwargs)

        n = NewMachine(states=states, transitions=[['advance', 'A', 'B']], initial='A')
        self.assertTrue(n.is_A())
        n.advance()
        self.assertTrue(n.is_B())
        with self.assertRaises(ValueError):
            NewMachine(state=['A', 'B'])

    def test_send_event_data_callbacks(self):
        states = ['A', 'B', 'C', 'D', 'E']
        s = Stuff()
        
        m = Machine(model=s, states=states, initial='A', send_event=False,
                    auto_transitions=True)
        m.add_transition(
            trigger='advance', source='A', dest='B', before='set_message')
        s.advance(message='Hallo. My name is Inigo Montoya.')
        self.assertTrue(s.message is not None and s.message.startswith('Hallo.'))
        s.to_A()
        s.advance('Test as positional argument')
        self.assertTrue(s.message is not None and s.message.startswith('Test as'))
        
        m.send_event = True
        m.add_transition(
            trigger='advance', source='B', dest='C', before='extract_message')
        s.advance(message='You killed my father. Prepare to die.')
        self.assertTrue(s.message is not None and s.message.startswith('You'))

    def test_send_event_data_conditions(self):
        states = ['A', 'B', 'C', 'D']
        s = Stuff()
        
        m = Machine(model=s, states=states, initial='A', send_event=False)
        m.add_transition(
            trigger='advance', source='A', dest='B',
            conditions='this_fails_by_default')
        s.advance(boolean=True)
        self.assertEqual(s.state, 'B')
        
        m.send_event = True
        m.add_transition(
            trigger='advance', source='B', dest='C',
            conditions='extract_boolean')
        s.advance(boolean=False)
        self.assertEqual(s.state, 'B')

    def test_auto_transitions(self):
        states = ['A', {'name': 'B'}, State(name='C')]  
        m = Machine(states=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')
        
        m = Machine(states=states, initial='A', auto_transitions=False)
        with self.assertRaises(AttributeError):
            m.to_C()

    def test_ordered_transitions(self):
        states = ['beginning', 'middle', 'end']
        m = Machine(states=states)
        m.add_ordered_transitions()
        self.assertEqual(m.state, 'initial')
        m.next_state()
        self.assertEqual(m.state, 'beginning')
        m.next_state()
        m.next_state()
        self.assertEqual(m.state, 'end')
        m.next_state()
        self.assertEqual(m.state, 'initial')

        
        m = Machine(states=states)
        m.add_ordered_transitions(loop_includes_initial=False)
        m.to_end()
        m.next_state()
        self.assertEqual(m.state, 'beginning')

        
        m = Machine(states=states)
        m.add_ordered_transitions(loop=False)
        m.to_end()
        with self.assertRaises(MachineError):
            m.next_state()

        
        m = Machine(states=states, initial='beginning')
        m.add_ordered_transitions(['end', 'beginning'], trigger='advance')
        m.advance()
        self.assertEqual(m.state, 'end')
        m.advance()
        self.assertEqual(m.state, 'beginning')

        
        m = Machine(states=states, initial='beginning', ordered_transitions=True)
        m.next_state()
        self.assertEqual(m.state, 'middle')

        
        m = Machine(states=states, initial='middle', ordered_transitions=True)
        m.next_state()
        self.assertEqual(m.state, 'end')
        m.next_state()
        self.assertEqual(m.state, 'beginning')

        
        m = Machine(states=states, initial='beginning')
        m.add_ordered_transitions(['middle', 'end'])
        self.assertEqual(m.state, 'beginning')
        with self.assertRaises(MachineError):
            m.next_state()
        m.to_middle()
        for s in ('end', 'middle', 'end'):
            m.next_state()
            self.assertEqual(m.state, s)

    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_ignore_invalid_triggers(self):
        a_state = State('A')
        transitions = [['a_to_b', 'A', 'B']]
        
        b_state = State('B')
        m1 = Machine(states=[a_state, b_state], transitions=transitions,
                     initial='B')
        with self.assertRaises(MachineError):
            m1.a_to_b()
        
        m2 = Machine(states=[a_state, b_state], transitions=transitions,
                     initial='B', ignore_invalid_triggers=True)
        m2.a_to_b()
        
        b_state = State('B', ignore_invalid_triggers=True)
        m3 = Machine(states=[a_state, b_state], transitions=transitions,
                     initial='B')
        m3.a_to_b()
        
        new_states = ['C', 'D']
        m1.add_states(new_states, ignore_invalid_triggers=True)
        m1.to_D()
        m1.a_to_b()  
        m1.to_B()
        with self.assertRaises(MachineError):
            m1.a_to_b()
        
        m3 = Machine(states=[a_state, b_state], transitions=transitions,
                     initial='B', ignore_invalid_triggers=False)
        m3.a_to_b()

    def test_string_callbacks(self):

        m = Machine(states=['A', 'B'],
                    before_state_change='before_state_change',
                    after_state_change='after_state_change', send_event=True,
                    initial='A', auto_transitions=True)

        m.before_state_change = MagicMock()
        m.after_state_change = MagicMock()

        m.to_B()

        self.assertTrue(m.before_state_change[0].called)
        self.assertTrue(m.after_state_change[0].called)

        
        event_data = m.after_state_change[0].call_args[0][0]
        self.assertIsInstance(event_data, EventData)
        self.assertTrue(event_data.result)

    def test_function_callbacks(self):
        before_state_change = MagicMock()
        after_state_change = MagicMock()

        m = Machine(states=['A', 'B'],
                    before_state_change=before_state_change,
                    after_state_change=after_state_change, send_event=True,
                    initial='A', auto_transitions=True)
        self.assertEqual(before_state_change, m.before_state_change[0])
        self.assertEqual(after_state_change, m.after_state_change[0])
        m.to_B()
        self.assertTrue(before_state_change.called)
        self.assertTrue(after_state_change.called)

    def test_state_callbacks(self):

        class Model:
            def on_enter_A(self):
                pass

            def on_exit_A(self):
                pass

            def on_enter_B(self):
                pass

            def on_exit_B(self):
                pass

        states = [State(name='A', on_enter='on_enter_A', on_exit='on_exit_A'),
                  State(name='B', on_enter='on_enter_B', on_exit='on_exit_B')]

        machine = Machine(Model(), states=states)
        state_a = machine.get_state('A')
        state_b = machine.get_state('B')
        self.assertEqual(len(state_a.on_enter), 1)
        self.assertEqual(len(state_a.on_exit), 1)
        self.assertEqual(len(state_b.on_enter), 1)
        self.assertEqual(len(state_b.on_exit), 1)

    def test_state_callable_callbacks(self):

        class Model:

            def __init__(self):
                self.exit_A_called = False
                self.exit_B_called = False

            def on_enter_A(self, event):
                pass

            def on_enter_B(self, event):
                pass

        states = [State(name='A', on_enter='on_enter_A', on_exit='tests.test_core.on_exit_A'),
                  State(name='B', on_enter='on_enter_B', on_exit=on_exit_B),
                  State(name='C', on_enter='tests.test_core.AAAA')]

        model = Model()
        machine = Machine(model, states=states, send_event=True, initial='A')
        state_a = machine.get_state('A')
        state_b = machine.get_state('B')
        self.assertEqual(len(state_a.on_enter), 1)
        self.assertEqual(len(state_a.on_exit), 1)
        self.assertEqual(len(state_b.on_enter), 1)
        self.assertEqual(len(state_b.on_exit), 1)
        model.to_B()
        self.assertTrue(model.exit_A_called)
        model.to_A()
        self.assertTrue(model.exit_B_called)
        with self.assertRaises(AttributeError):
            model.to_C()

    def test_pickle(self):
        import sys
        if sys.version_info < (3, 4):
            import dill as pickle
        else:
            import pickle

        states = ['A', 'B', 'C', 'D']
        
        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()
        dump = pickle.dumps(m)
        self.assertIsNotNone(dump)
        m2 = pickle.loads(dump)
        self.assertEqual(m.state, m2.state)
        m2.run()

    def test_pickle_model(self):
        import sys
        if sys.version_info < (3, 4):
            import dill as pickle
        else:
            import pickle

        self.stuff.to_B()
        dump = pickle.dumps(self.stuff)
        self.assertIsNotNone(dump)
        model2 = pickle.loads(dump)
        self.assertEqual(self.stuff.state, model2.state)
        model2.to_F()

    def test_queued(self):
        states = ['A', 'B', 'C', 'D']
        

        def change_state(machine):
            self.assertEqual(machine.state, 'A')
            if machine.has_queue:
                machine.run(machine=machine)
                self.assertEqual(machine.state, 'A')
            else:
                with self.assertRaises(MachineError):
                    machine.run(machine=machine)

        transitions = [
            {'trigger': 'walk', 'source': 'A', 'dest': 'B', 'before': change_state},
            {'trigger': 'run', 'source': 'B', 'dest': 'C'},
            {'trigger': 'sprint', 'source': 'C', 'dest': 'D'}
        ]

        m = Machine(states=states, transitions=transitions, initial='A')
        m.walk(machine=m)
        self.assertEqual(m.state, 'B')
        m = Machine(states=states, transitions=transitions, initial='A', queued=True)
        m.walk(machine=m)
        self.assertEqual(m.state, 'C')

    def test_queued_errors(self):
        def before_change(machine):
            if machine.has_queue:
                machine.to_A(machine)
            machine._queued = False

        def after_change(machine):
            machine.to_C(machine)

        states = ['A', 'B', 'C']
        transitions = [{'trigger': 'do', 'source': '*', 'dest': 'C',
                        'before': partial(self.stuff.this_raises, ValueError)}]
        m = Machine(states=states, transitions=transitions, queued=True,
                    before_state_change=before_change, after_state_change=after_change)
        with self.assertRaises(MachineError):
            m.to_B(machine=m)

        with self.assertRaises(ValueError):
            m.do(machine=m)

    def test_queued_remove(self):
        m = self.machine_cls(model=None, states=['A', 'B', 'C'], initial='A', queued=True)
        assert_equal = self.assertEqual

        class BaseModel:
            def on_enter_A(self):
                pass

            def on_enter_B(self):
                pass

            def on_enter_C(self):
                pass

        class SubModel(BaseModel):
            def __init__(self):
                self.inner = BaseModel()

            def on_enter_A(self):
                self.to_B()
                self.inner.to_B()

            def on_enter_B(self):
                self.to_C()
                self.inner.to_C()
                
                assert_equal(4, len(m._transition_queue))
                m.remove_model(self)
                
                assert_equal(3, len(m._transition_queue))

            def on_enter_C(self):
                raise RuntimeError("Event was not cancelled")
        model = SubModel()
        m.add_model([model, model.inner])
        model.to_A()
        
        m.remove_model(model.inner)
        self.assertTrue(model.inner.is_C())

    def test___getattr___and_identify_callback(self):
        m = self.machine_cls(Stuff(), states=['A', 'B', 'C'], initial='A')
        m.add_transition('move', 'A', 'B')
        m.add_transition('move', 'B', 'C')

        callback = m.__getattr__('before_move')
        self.assertTrue(callable(callback))

        with self.assertRaises(AttributeError):
            m.__getattr__('before_no_such_transition')

        with self.assertRaises(AttributeError):
            m.__getattr__('before_no_such_transition')

        with self.assertRaises(AttributeError):
            m.__getattr__('__no_such_method__')

        with self.assertRaises(AttributeError):
            m.__getattr__('')

        type, target = m._identify_callback('on_exit_foobar')
        self.assertEqual(type, 'on_exit')
        self.assertEqual(target, 'foobar')

        type, target = m._identify_callback('on_exitfoobar')
        self.assertEqual(type, None)
        self.assertEqual(target, None)

        type, target = m._identify_callback('notacallback_foobar')
        self.assertEqual(type, None)
        self.assertEqual(target, None)

        type, target = m._identify_callback('totallyinvalid')
        self.assertEqual(type, None)
        self.assertEqual(target, None)

        type, target = m._identify_callback('before__foobar')
        self.assertEqual(type, 'before')
        self.assertEqual(target, '_foobar')

        type, target = m._identify_callback('before__this__user__likes__underscores___')
        self.assertEqual(type, 'before')
        self.assertEqual(target, '_this__user__likes__underscores___')

        type, target = m._identify_callback('before_stuff')
        self.assertEqual(type, 'before')
        self.assertEqual(target, 'stuff')

        type, target = m._identify_callback('before_trailing_underscore_')
        self.assertEqual(type, 'before')
        self.assertEqual(target, 'trailing_underscore_')

        type, target = m._identify_callback('before_')
        self.assertIs(type, None)
        self.assertIs(target, None)

        type, target = m._identify_callback('__')
        self.assertIs(type, None)
        self.assertIs(target, None)

        type, target = m._identify_callback('')
        self.assertIs(type, None)
        self.assertIs(target, None)

    def test_state_and_transition_with_underscore(self):
        m = Machine(Stuff(), states=['_A_', '_B_', '_C_'], initial='_A_')
        m.add_transition('_move_', '_A_', '_B_', prepare='increase_level')
        m.add_transition('_after_', '_B_', '_C_', prepare='increase_level')
        m.add_transition('_on_exit_', '_C_', '_A_', prepare='increase_level', conditions='this_fails')

        m.model._move_()
        self.assertEqual(m.model.state, '_B_')
        self.assertEqual(m.model.level, 2)

        m.model._after_()
        self.assertEqual(m.model.state, '_C_')
        self.assertEqual(m.model.level, 3)

        
        m.model._on_exit_()
        self.assertEqual(m.model.state, '_C_')
        self.assertEqual(m.model.level, 4)

    def test_callback_identification(self):
        m = Machine(Stuff(), states=['A', 'B', 'C', 'D', 'E', 'F'], initial='A')
        m.add_transition('transition', 'A', 'B', before='increase_level')
        m.add_transition('after', 'B', 'C', before='increase_level')
        m.add_transition('on_exit_A', 'C', 'D', before='increase_level', conditions='this_fails')
        m.add_transition('check', 'C', 'E', before='increase_level')
        m.add_transition('prepare', 'E', 'F', before='increase_level')
        m.add_transition('before', 'F', 'A', before='increase_level')

        m.before_transition('increase_level')
        m.before_after('increase_level')
        m.before_on_exit_A('increase_level')
        m.after_check('increase_level')
        m.before_prepare('increase_level')
        m.before_before('increase_level')

        m.model.transition()
        self.assertEqual(m.model.state, 'B')
        self.assertEqual(m.model.level, 3)

        m.model.after()
        self.assertEqual(m.model.state, 'C')
        self.assertEqual(m.model.level, 5)

        m.model.on_exit_A()
        self.assertEqual(m.model.state, 'C')
        self.assertEqual(m.model.level, 5)

        m.model.check()
        self.assertEqual(m.model.state, 'E')
        self.assertEqual(m.model.level, 7)

        m.model.prepare()
        self.assertEqual(m.model.state, 'F')
        self.assertEqual(m.model.level, 9)

        m.model.before()
        self.assertEqual(m.model.state, 'A')
        self.assertEqual(m.model.level, 11)

        
        with self.assertRaises(MachineError):
            m.model.on_exit_A()

    def test_process_trigger(self):
        m = Machine(states=['raw', 'processed'], initial='raw')
        m.add_transition('process', 'raw', 'processed')

        m.process()
        self.assertEqual(m.state, 'processed')

    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.assertTrue(isinstance(m.model, list) and 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])
        
        
        self.assertNotIsInstance(m.model, list)

    def test_dispatch(self):
        s1, s2 = Stuff(), Stuff()
        states = ['A', 'B', 'C']
        m = Machine(model=s1, states=states, ignore_invalid_triggers=True,
                    initial=states[0], transitions=[['go', 'A', 'B'], ['go', 'B', 'C']])
        m.add_model(s2, initial='B')
        m.dispatch('go')
        self.assertEqual(s1.state, 'B')
        self.assertEqual(s2.state, 'C')

    def test_remove_model(self):
        m = self.machine_cls()
        self.assertIn(m, m.models)
        m.remove_model(m)
        self.assertNotIn(m, m.models)

    def test_string_trigger(self):
        def return_value(value):
            return value

        class Model:
            def trigger(self, value):
                return value

        self.stuff.machine.add_transition('do', '*', 'C')
        self.stuff.trigger('do')
        self.assertTrue(self.stuff.is_C())
        self.stuff.machine.add_transition('maybe', 'C', 'A', conditions=return_value)
        self.assertFalse(self.stuff.trigger('maybe', value=False))
        self.assertTrue(self.stuff.trigger('maybe', value=True))
        self.assertTrue(self.stuff.is_A())
        with self.assertRaises(AttributeError):
            self.stuff.trigger('not_available')
        with self.assertRaises(MachineError):
            self.stuff.trigger('maybe')

        model = Model()
        m = Machine(model=model)
        self.assertEqual(model.trigger(5), 5)
        self.stuff.machine.add_transition('do_raise_keyerror', '*', 'C',
                                          before=partial(self.stuff.this_raises, KeyError))
        with self.assertRaises(KeyError):
            self.stuff.trigger('do_raise_keyerror')

        self.stuff.machine.get_model_state(self.stuff).ignore_invalid_triggers = True
        self.stuff.trigger('should_not_raise_anything')
        self.stuff.trigger('to_A')
        self.assertTrue(self.stuff.is_A())
        self.stuff.machine.ignore_invalid_triggers = True
        self.stuff.trigger('should_not_raise_anything')

    def test_get_triggers(self):
        states = ['A', 'B', 'C']
        transitions = [['a2b', 'A', 'B'],
                       ['a2c', 'A', 'C'],
                       ['c2b', 'C', 'B']]
        machine = Machine(states=states, transitions=transitions, initial='A', auto_transitions=False)
        self.assertEqual(len(machine.get_triggers('A')), 2)
        self.assertEqual(len(machine.get_triggers('B')), 0)
        self.assertEqual(len(machine.get_triggers('C')), 1)
        
        m = self.stuff.machine
        self.assertEqual(len(m.get_triggers('B')), len(m.states))
        trigger_name = m.get_triggers('B')
        trigger_state = m.get_triggers(m.states['B'])
        self.assertEqual(trigger_name, trigger_state)

    def test_skip_override(self):
        local_mock = MagicMock()

        class Model(object):

            def go(self):
                local_mock()
        model = Model()
        transitions = [['go', 'A', 'B'], ['advance', 'A', 'B']]
        m = self.stuff.machine_cls(model=model, states=['A', 'B'], transitions=transitions, initial='A')
        model.go()
        self.assertEqual(model.state, 'A')
        self.assertTrue(local_mock.called)
        model.advance()
        self.assertEqual(model.state, 'B')
        model.to_A()
        model.trigger('go')
        self.assertEqual(model.state, 'B')

    @skipIf(sys.version_info < (3, ),
            "String-checking disabled on PY-2 because is different")
    def test_repr(self):
        def a_condition(event_data):
            self.assertRegex(
                str(event_data.transition.conditions),
                r"\[<Condition\(<function TestTransitions.test_repr.<locals>"
                r".a_condition at [^>]+>\)@\d+>\]")
            return True

        
        def check_prepare_repr(event_data):
            self.assertRegex(
                str(event_data),
                r"<EventData\(<Event\('do_strcheck'\)@\d+>, "
                r"<State\('A'\)@\d+>, "
                r"None\)@\d+>")

        def check_before_repr(event_data):
            self.assertRegex(
                str(event_data),
                r"<EventData\(<Event\('do_strcheck'\)@\d+>, "
                r"<State\('A'\)@\d+>, "
                r"<Transition\('A', 'B'\)@\d+>\)@\d+>")
            m.checked = True

        m = Machine(states=['A', 'B'],
                    prepare_event=check_prepare_repr,
                    before_state_change=check_before_repr, send_event=True,
                    initial='A')
        m.add_transition('do_strcheck', 'A', 'B', conditions=a_condition)

        self.assertTrue(m.do_strcheck())
        self.assertIn('checked', vars(m))

    def test_machine_prepare(self):

        global_mock = MagicMock()
        local_mock = MagicMock()

        def global_callback():
            global_mock()

        def local_callback():
            local_mock()

        def always_fails():
            return False

        transitions = [
            {'trigger': 'go', 'source': 'A', 'dest': 'B', 'conditions': always_fails, 'prepare': local_callback},
            {'trigger': 'go', 'source': 'A', 'dest': 'B', 'conditions': always_fails, 'prepare': local_callback},
            {'trigger': 'go', 'source': 'A', 'dest': 'B', 'conditions': always_fails, 'prepare': local_callback},
            {'trigger': 'go', 'source': 'A', 'dest': 'B', 'conditions': always_fails, 'prepare': local_callback},
            {'trigger': 'go', 'source': 'A', 'dest': 'B', 'prepare': local_callback},

        ]
        m = Machine(states=['A', 'B'], transitions=transitions,
                    prepare_event=global_callback, initial='A')

        m.go()
        self.assertEqual(global_mock.call_count, 1)
        self.assertEqual(local_mock.call_count, len(transitions))

    def test_machine_finalize(self):

        finalize_mock = MagicMock()

        def always_fails(event_data):
            return False

        transitions = [
            {'trigger': 'go', 'source': 'A', 'dest': 'B'},
            {'trigger': 'planA', 'source': 'B', 'dest': 'A', 'conditions': always_fails},
            {'trigger': 'planB', 'source': 'B', 'dest': 'A',
             'conditions': partial(self.stuff.this_raises, RuntimeError)}
        ]
        m = self.stuff.machine_cls(states=['A', 'B'], transitions=transitions,
                                   finalize_event=finalize_mock, initial='A', send_event=True)

        m.go()
        self.assertEqual(finalize_mock.call_count, 1)
        m.planA()

        event_data = finalize_mock.call_args[0][0]
        self.assertIsInstance(event_data, EventData)
        self.assertEqual(finalize_mock.call_count, 2)
        self.assertFalse(event_data.result)
        with self.assertRaises(RuntimeError):
            m.planB()

        m.finalize_event.append(partial(self.stuff.this_raises, ValueError))
        
        
        with self.assertRaises(RuntimeError):
            m.planB()
        self.assertEqual(4, finalize_mock.call_count)

    def test_machine_finalize_exception(self):

        def finalize_callback(event):
            self.assertIsInstance(event.error, ZeroDivisionError)

        m = self.stuff.machine_cls(states=['A', 'B'], send_event=True, initial='A',
                                   before_state_change=partial(self.stuff.this_raises, ZeroDivisionError),
                                   finalize_event=finalize_callback)

        with self.assertRaises(ZeroDivisionError):
            m.to_B()

    def test_prep_ordered_arg(self):
        self.assertTrue(len(_prep_ordered_arg(3, None)) == 3)
        self.assertTrue(all(a is None for a in _prep_ordered_arg(3, None)))
        with self.assertRaises(ValueError):
            
            _prep_ordered_arg(3, [None, None])  

    def test_ordered_transition_callback(self):
        class Model:
            def __init__(self):
                self.flag = False

            def make_true(self):
                self.flag = True

        model = Model()
        states = ['beginning', 'middle', 'end']
        transits = [None, None, 'make_true']
        m = Machine(model, states, initial='beginning')
        m.add_ordered_transitions(before=transits)
        model.next_state()
        self.assertFalse(model.flag)
        model.next_state()
        model.next_state()
        self.assertTrue(model.flag)

    def test_ordered_transition_condition(self):
        class Model:
            def __init__(self):
                self.blocker = False

            def check_blocker(self):
                return self.blocker

        model = Model()
        states = ['beginning', 'middle', 'end']
        m = Machine(model, states, initial='beginning')
        m.add_ordered_transitions(conditions=[None, None, 'check_blocker'])
        model.to_end()
        self.assertFalse(model.next_state())
        model.blocker = True
        self.assertTrue(model.next_state())

    def test_get_transitions(self):
        states = ['A', 'B', 'C', 'D']
        m = self.machine_cls(states=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')])
        self.assertEqual(
            sorted([(t.source, t.dest)
                    for t in m.get_transitions(dest='D')]),
            [('A', 'D'), ('A', 'D'), ('B', 'D'), ('C', 'D')])
        self.assertEqual(
            [(t.source, t.dest)
             for t in m.get_transitions(source=m.states['A'], dest=m.states['D'])],
            [('A', 'D'), ('A', 'D')])
        self.assertEqual(
            sorted([(t.source, t.dest)
                    for t in m.get_transitions(dest=m.states['D'])]),
            [('A', 'D'), ('A', 'D'), ('B', 'D'), ('C', 'D')])

    def test_remove_transition(self):
        self.stuff.machine.add_transition('go', ['A', 'B', 'C'], 'D')
        self.stuff.machine.add_transition('walk', 'A', 'B')
        self.stuff.go()
        self.assertEqual(self.stuff.state, 'D')
        self.stuff.to_A()
        self.stuff.machine.remove_transition('go', source='A')
        with self.assertRaises(MachineError):
            self.stuff.go()
        self.stuff.machine.add_transition('go', 'A', 'D')
        self.stuff.walk()
        self.stuff.go()
        self.assertEqual(self.stuff.state, 'D')
        self.stuff.to_C()
        self.stuff.machine.remove_transition('go', dest='D')
        self.assertFalse(hasattr(self.stuff, 'go'))

    def test_reflexive_transition(self):
        self.stuff.machine.add_transition('reflex', ['A', 'B'], '=', after='increase_level')
        self.assertEqual(self.stuff.state, 'A')
        self.stuff.reflex()
        self.assertEqual(self.stuff.state, 'A')
        self.assertEqual(self.stuff.level, 2)
        self.stuff.to_B()
        self.assertEqual(self.stuff.state, 'B')
        self.stuff.reflex()
        self.assertEqual(self.stuff.state, 'B')
        self.assertEqual(self.stuff.level, 3)
        self.stuff.to_C()
        with self.assertRaises(MachineError):
            self.stuff.reflex()
        self.assertEqual(self.stuff.level, 3)

    def test_internal_transition(self):
        m = Machine(Stuff(), states=['A', 'B'], initial='A')
        m.add_transition('move', 'A', None, prepare='increase_level')
        m.model.move()
        self.assertEqual(m.model.state, 'A')
        self.assertEqual(m.model.level, 2)

    def test_dynamic_model_state_attribute(self):
        class Model:
            def __init__(self):
                self.status = None
                self.state = 'some_value'

        m = self.machine_cls(Model(), states=['A', 'B'], initial='A', model_attribute='status')
        self.assertEqual(m.model.status, 'A')
        self.assertEqual(m.model.state, 'some_value')

        m.add_transition('move', 'A', 'B')
        m.model.move()

        self.assertEqual(m.model.status, 'B')
        self.assertEqual(m.model.state, 'some_value')

    def test_multiple_machines_per_model(self):
        class Model:
            def __init__(self):
                self.car_state = None
                self.driver_state = None

        instance = Model()
        machine_a = Machine(instance, states=['A', 'B'], initial='A', model_attribute='car_state')
        machine_a.add_transition('accelerate_car', 'A', 'B')
        machine_b = Machine(instance, states=['A', 'B'], initial='B', model_attribute='driver_state')
        machine_b.add_transition('driving', 'B', 'A')

        assert instance.car_state == 'A'
        assert instance.driver_state == 'B'
        assert instance.is_car_state_A()
        assert instance.is_driver_state_B()

        instance.accelerate_car()
        assert instance.car_state == 'B'
        assert instance.driver_state == 'B'
        assert not instance.is_car_state_A()
        assert instance.is_car_state_B()

        instance.driving()
        assert instance.driver_state == 'A'
        assert instance.car_state == 'B'
        assert instance.is_driver_state_A()
        assert not instance.is_driver_state_B()
        assert instance.to_driver_state_B()
        assert instance.driver_state == 'B'

    def test_initial_not_registered(self):
        m1 = self.machine_cls(states=['A', 'B'], initial=self.machine_cls.state_cls('C'))
        self.assertTrue(m1.is_C())
        self.assertTrue('C' in m1.states)

    def test_trigger_name_cannot_be_equal_to_model_attribute(self):
        m = self.machine_cls(states=['A', 'B'])

        with self.assertRaises(ValueError):
            m.add_transition(m.model_attribute, "A", "B")

    def test_new_state_in_enter_callback(self):

        machine = self.machine_cls(states=['A', 'B'], initial='A')

        def on_enter_B():
            state = self.machine_cls.state_cls(name='C')
            machine.add_state(state)
            machine.to_C()

        machine.on_enter_B(on_enter_B)
        machine.to_B()

    def test_on_exception_callback(self):
        mock = MagicMock()

        def on_exception(event_data):
            self.assertIsInstance(event_data.error, (ValueError, MachineError))
            mock()

        m = self.machine_cls(states=['A', 'B'], initial='A', transitions=[['go', 'A', 'B']], send_event=True,
                             after_state_change=partial(self.stuff.this_raises, ValueError))
        with self.assertRaises(ValueError):
            m.to_B()
        self.assertTrue(m.is_B())
        with self.assertRaises(MachineError):
            m.go()

        m.on_exception.append(on_exception)
        m.to_B()
        m.go()
        self.assertTrue(mock.called)
        self.assertEqual(2, mock.call_count)

    def test_may_transition(self):
        states = ['A', 'B', 'C']
        d = DummyModel()
        m = Machine(model=d, states=states, initial='A', auto_transitions=False)
        m.add_transition('walk', 'A', 'B')
        m.add_transition('stop', 'B', 'C')
        m.add_transition('wait', 'B', None)
        assert d.may_walk()
        assert not d.may_stop()
        assert not d.may_wait()

        d.walk()
        assert not d.may_walk()
        assert d.may_stop()
        assert d.may_wait()

    def test_may_transition_for_autogenerated_triggers(self):
        states = ['A', 'B', 'C']
        m = Machine(states=states, initial='A')
        assert m.may_to_A()
        m.to_A()
        assert m.to_B()
        m.to_B()
        assert m.may_to_C()
        m.to_C()

    def test_may_transition_with_conditions(self):
        states = ['A', 'B', 'C']
        d = DummyModel()
        m = Machine(model=d, states=states, initial='A', auto_transitions=False)
        m.add_transition('walk', 'A', 'B', conditions=[lambda: False])
        m.add_transition('stop', 'B', 'C')
        m.add_transition('run', 'A', 'C')
        assert not d.may_walk()
        assert not d.may_stop()
        assert d.may_run()
        d.run()
        assert not d.may_run()

    def test_may_transition_with_auto_transitions(self):
        states = ['A', 'B', 'C']
        d = DummyModel()
        self.machine_cls(model=d, states=states, initial='A')
        assert d.may_to_A()
        assert d.may_to_B()
        assert d.may_to_C()

    def test_machine_may_transitions(self):
        states = ['A', 'B', 'C']
        m = self.machine_cls(states=states, initial='A', auto_transitions=False)
        m.add_transition('walk', 'A', 'B', conditions=[lambda: False])
        m.add_transition('stop', 'B', 'C')
        m.add_transition('run', 'A', 'C')
        m.add_transition('reset', 'C', 'A')

        assert not m.may_walk()
        assert not m.may_stop()
        assert m.may_run()
        m.run()
        assert not m.may_run()
        assert not m.may_stop()
        assert not m.may_walk()

    def test_may_transition_with_invalid_state(self):
        states = ['A', 'B', 'C']
        d = DummyModel()
        m = self.machine_cls(model=d, states=states, initial='A', auto_transitions=False)
        m.add_transition('walk', 'A', 'UNKNOWN')
        assert not d.may_walk()