give thread some time to start give thread some time to start if forward has not been locked it is still running we have to wait to be sure it is done thread will release lock before transition is finished go to non initial state b pickle stuff model check if machines of stuff and stuff2 are truly separated give thread some time to start both objects should be in different states and also not share locks stuff should not be locked and execute fast stuff2 should be locked and take about 1 second to be executed this test has been used to quantify the changes made in locking in version 0 5 0 see https github comtyarkonitransitionsissues167 for the results def testperformanceself import timeit states a b c transitions go a b go b c go c a m1 machinefactory getpredefined m2 machinefactory getpredefinedlockedtrue def testm1 m1 m1statesstates transitionstransitions initial a m1 gettriggers a def testm2 m2 m2statesstates transitionstransitions initial a m2 gettriggers a t1 timeit timeittestm1 number20000 t2 timeit timeittestm2 number20000 self assertalmostequalt2t1 1 delta0 5 there are a lot of internal enterexits but the key is that the outermost are in the expected order since there are a lot of internal enterexits we don t actually know how deep in the stack to look for these should be able to correct when https github comtyarkonitransitionsissues167 same as testlockedtransition but with lockedhierarchicalmachine give thread some time to start if forward has not been locked it is still running we have to wait to be sure it is done go to non initial state b pickle stuff model check if machines of stuff and stuff2 are truly separated give thread some time to start both objects should be in different states and also not share locks stuff should not be locked and execute fast stuff2 should be locked and take about 1 second to be executed type ignore type type lockedmachine give thread some time to start give thread some time to start if forward has not been locked it is still running we have to wait to be sure it is done checking takes 1s and returns false thread will release lock before transition is finished go to non initial state b pickle stuff model check if machines of stuff and stuff2 are truly separated give thread some time to start both objects should be in different states and also not share locks stuff should not be locked and execute fast stuff2 should be locked and take about 1 second to be executed was 3 before was 72 before this test has been used to quantify the changes made in locking in version 0 5 0 see https github com tyarkoni transitions issues 167 for the results def test_performance self import timeit states a b c transitions go a b go b c go c a m1 machinefactory get_predefined m2 machinefactory get_predefined locked true def test_m1 m1 m1 states states transitions transitions initial a m1 get_triggers a def test_m2 m2 m2 states states transitions transitions initial a m2 get_triggers a t1 timeit timeit test_m1 number 20000 t2 timeit timeit test_m2 number 20000 self assertalmostequal t2 t1 1 delta 0 5 type list tuple any str type type lockedmachine there are a lot of internal enter exits but the key is that the outermost are in the expected order since there are a lot of internal enter exits we don t actually know how deep in the stack to look for these should be able to correct when https github com tyarkoni transitions issues 167 same as testlockedtransition but with lockedhierarchicalmachine type type lockedhierarchicalmachine give thread some time to start if forward has not been locked it is still running we have to wait to be sure it is done type ignore go to non initial state b pickle stuff model check if machines of stuff and stuff2 are truly separated give thread some time to start both objects should be in different states and also not share locks stuff should not be locked and execute fast stuff2 should be locked and take about 1 second to be executed
try:
    from builtins import object
except ImportError:
    pass

import time
from threading import Thread
import logging

from transitions.extensions import LockedHierarchicalMachine, LockedMachine
from .test_nesting import TestNestedTransitions
from .test_core import TestTransitions, TYPE_CHECKING
from .utils import Stuff, DummyModel, SomeContext

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

if TYPE_CHECKING:
    from typing import List, Type, Tuple, Any

logger = logging.getLogger(__name__)
logger.addHandler(logging.NullHandler())


def heavy_processing():
    time.sleep(1)


def heavy_checking():
    time.sleep(0.5)
    return False


class TestLockedTransitions(TestTransitions):

    def setUp(self):
        self.machine_cls = LockedMachine  
        self.stuff = Stuff(machine_cls=self.machine_cls)
        self.stuff.heavy_processing = heavy_processing
        self.stuff.machine.add_transition('forward', 'A', 'B', before='heavy_processing')

    def tearDown(self):
        pass

    def test_thread_access(self):
        thread = Thread(target=self.stuff.forward)
        thread.start()
        
        time.sleep(0.01)
        self.assertTrue(self.stuff.is_B())

    def test_parallel_access(self):
        thread = Thread(target=self.stuff.forward)
        thread.start()
        
        time.sleep(0.01)
        self.stuff.to_C()
        
        
        time.sleep(1)
        self.assertEqual(self.stuff.state, "C")

    def test_parallel_deep(self):
        self.stuff.machine.add_transition('deep', source='*', dest='C', after='to_D')
        thread = Thread(target=self.stuff.deep)
        thread.start()
        time.sleep(0.01)
        self.stuff.to_C()
        time.sleep(1)
        self.assertEqual(self.stuff.state, "C")

    def test_conditional_access(self):
        self.stuff.heavy_checking = heavy_checking  
        self.stuff.machine.add_transition('advance', 'A', 'B', conditions='heavy_checking')
        self.stuff.machine.add_transition('advance', 'A', 'D')
        t = Thread(target=self.stuff.advance)
        t.start()
        time.sleep(0.1)
        logger.info('Check if state transition done...')
        
        res = self.stuff.is_D()
        self.assertTrue(res)

    def test_pickle(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)
        stuff2 = pickle.loads(dump)
        self.assertTrue(stuff2.is_B())
        
        stuff2.to_A()
        self.stuff.to_C()
        self.assertTrue(stuff2.is_A())
        thread = Thread(target=stuff2.forward)
        thread.start()
        
        time.sleep(0.01)
        
        
        begin = time.time()
        
        self.assertTrue(self.stuff.is_C())
        fast = time.time()
        
        
        self.assertTrue(stuff2.is_B())
        blocked = time.time()
        self.assertAlmostEqual(fast - begin, 0, delta=0.1)
        self.assertAlmostEqual(blocked - begin, 1, delta=0.1)

    def test_context_managers(self):

        class CounterContext(object):
            def __init__(self):
                self.counter = 0
                self.level = 0
                self.max = 0
                super(CounterContext, self).__init__()

            def __enter__(self):
                self.counter += 1
                self.level += 1
                self.max = max(self.level, self.max)

            def __exit__(self, *exc):
                self.level -= 1

        M = LockedMachine
        c = CounterContext()
        m = M(states=['A', 'B', 'C', 'D'], transitions=[['reset', '*', 'A']], initial='A', machine_context=c)
        m.get_triggers('A')
        self.assertEqual(c.max, 1)  
        self.assertEqual(c.counter, 4)  

    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    


class TestMultipleContexts(TestTransitions):

    def setUp(self):
        self.event_list = []  

        self.s1 = DummyModel()

        self.c1 = SomeContext(event_list=self.event_list)
        self.c2 = SomeContext(event_list=self.event_list)
        self.c3 = SomeContext(event_list=self.event_list)
        self.c4 = SomeContext(event_list=self.event_list)

        self.machine_cls = LockedMachine  
        self.stuff = Stuff(machine_cls=self.machine_cls, extra_kwargs={
            'machine_context': [self.c1, self.c2]
        })
        self.stuff.machine.add_model(self.s1, model_context=[self.c3, self.c4])
        del self.event_list[:]

        self.stuff.machine.add_transition('forward', 'A', 'B')

    def tearDown(self):
        self.stuff.machine.remove_model(self.s1)

    def test_ordering(self):
        self.stuff.forward()
        
        self.assertEqual((self.c1, "enter"), self.event_list[0])
        self.assertEqual((self.c2, "enter"), self.event_list[1])
        self.assertEqual((self.c2, "exit"), self.event_list[-2])
        self.assertEqual((self.c1, "exit"), self.event_list[-1])

    def test_model_context(self):
        self.s1.forward()
        self.assertEqual((self.c1, "enter"), self.event_list[0])
        self.assertEqual((self.c2, "enter"), self.event_list[1])

        
        
        self.assertIn((self.c3, "enter"), self.event_list)
        self.assertIn((self.c4, "enter"), self.event_list)
        self.assertIn((self.c4, "exit"), self.event_list)
        self.assertIn((self.c3, "exit"), self.event_list)

        self.assertEqual((self.c2, "exit"), self.event_list[-2])
        self.assertEqual((self.c1, "exit"), self.event_list[-1])



class TestLockedHierarchicalTransitions(TestNestedTransitions, TestLockedTransitions):

    def setUp(self):
        states = ['A', 'B', {'name': 'C', 'children': ['1', '2', {'name': '3', 'children': ['a', 'b', 'c']}]},
                  'D', 'E', 'F']
        self.machine_cls = LockedHierarchicalMachine  
        self.state_cls = self.machine_cls.state_cls
        self.state_cls.separator = '_'
        self.stuff = Stuff(states, machine_cls=self.machine_cls)
        self.stuff.heavy_processing = heavy_processing
        self.stuff.machine.add_transition('forward', '*', 'B', before='heavy_processing')

    def test_parallel_access(self):
        thread = Thread(target=self.stuff.forward)
        thread.start()
        
        time.sleep(0.01)
        self.stuff.to_C()
        
        
        time.sleep(1)
        self.assertEqual(self.stuff.state, "C")

    def test_callbacks(self):

        class MachineModel(self.stuff.machine_cls):  
            def __init__(self):
                self.mock = MagicMock()
                super(MachineModel, self).__init__(self, states=['A', 'B', 'C'])

            def on_enter_A(self):
                self.mock()

        model = MachineModel()
        model.to_A()
        self.assertTrue(model.mock.called)

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

        states = ['A', 'B', {'name': 'C', 'children': ['1', '2', {'name': '3', 'children': ['a', 'b', 'c']}]},
                  'D', 'E', 'F']
        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')
        m.heavy_processing = heavy_processing
        m.add_transition('forward', 'A', 'B', before='heavy_processing')

        
        m.to_B()

        
        dump = pickle.dumps(m)
        self.assertIsNotNone(dump)
        m2 = pickle.loads(dump)
        self.assertTrue(m2.is_B())
        m2.to_C_3_a()
        m2.to_C_3_b()
        
        m2.to_A()
        m.to_C()
        self.assertTrue(m2.is_A())
        thread = Thread(target=m2.forward)
        thread.start()
        
        time.sleep(0.01)
        
        
        begin = time.time()
        
        self.assertTrue(m.is_C())
        fast = time.time()
        
        
        self.assertTrue(m2.is_B())
        blocked = time.time()
        self.assertAlmostEqual(fast - begin, 0, delta=0.1)
        self.assertAlmostEqual(blocked - begin, 1, delta=0.1)