<s> from calvin . utilities import calvinuuid <EOL> from calvin . runtime . north import storage <EOL> from calvin . runtime . north import appmanager <EOL> from calvin . runtime . south . plugins . async import threads <EOL> from calvin . utilities . calvin_callback import CalvinCB <EOL> from calvin . tests import TestNode , TestActor , TestPort <EOL> import Queue <EOL> import pytest <EOL> import time <EOL> try : <EOL> import pytest . inlineCallbacks <EOL> except ImportError : <EOL> pytest . inlineCallbacks = lambda * args : False <EOL> @ pytest . mark . interactive <EOL> class TestSetFlushAndGet ( object ) : <EOL> @ pytest . mark . slow <EOL> @ pytest . inlineCallbacks <EOL> def test_late_start ( self ) : <EOL> self . q = Queue . Queue ( ) <EOL> def cb ( key , value ) : <EOL> self . q . put ( { "<STR_LIT:key>" : key , "<STR_LIT:value>" : value } ) <EOL> def started_cb ( started ) : <EOL> self . q . put ( started ) <EOL> self . storage = storage . Storage ( ) <EOL> self . storage . set ( "<STR_LIT:test>" , "<STR_LIT:1>" , <NUM_LIT:1> , None ) <EOL> self . storage . set ( "<STR_LIT:test>" , "<STR_LIT:2>" , <NUM_LIT:2> , None ) <EOL> self . storage . set ( "<STR_LIT:test>" , "<STR_LIT:3>" , <NUM_LIT:3> , None ) <EOL> assert "<STR_LIT>" in self . storage . localstore <EOL> assert "<STR_LIT>" in self . storage . localstore <EOL> assert "<STR_LIT>" in self . storage . localstore <EOL> yield threads . defer_to_thread ( self . storage . start , CalvinCB ( started_cb ) ) <EOL> yield threads . defer_to_thread ( time . sleep , <NUM_LIT:2> ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value <EOL> assert "<STR_LIT>" not in self . storage . localstore <EOL> assert "<STR_LIT>" not in self . storage . localstore <EOL> assert "<STR_LIT>" not in self . storage . localstore <EOL> yield threads . defer_to_thread ( self . storage . get , "<STR_LIT:test>" , "<STR_LIT:3>" , CalvinCB ( func = cb ) ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value [ "<STR_LIT:value>" ] == <NUM_LIT:3> <EOL> yield threads . defer_to_thread ( self . storage . stop ) <EOL> @ pytest . mark . interactive <EOL> @ pytest . mark . slow <EOL> class TestStorageStarted ( object ) : <EOL> @ pytest . inlineCallbacks <EOL> def setup_class ( self ) : <EOL> self . storage = storage . Storage ( ) <EOL> yield threads . defer_to_thread ( self . storage . start ) <EOL> yield threads . defer_to_thread ( time . sleep , <NUM_LIT:2> ) <EOL> @ pytest . inlineCallbacks <EOL> def teardown_class ( self ) : <EOL> yield threads . defer_to_thread ( self . storage . stop ) <EOL> yield threads . defer_to_thread ( time . sleep , <NUM_LIT:2> ) <EOL> @ pytest . inlineCallbacks <EOL> def test_node_functions ( self ) : <EOL> self . q = Queue . Queue ( ) <EOL> def cb ( key , value ) : <EOL> self . q . put ( { "<STR_LIT:key>" : key , "<STR_LIT:value>" : value } ) <EOL> yield threads . defer_to_thread ( time . sleep , <NUM_LIT:2> ) <EOL> node = TestNode ( "<STR_LIT>" ) <EOL> yield threads . defer_to_thread ( self . storage . add_node , node , cb = CalvinCB ( cb ) ) <EOL> yield threads . defer_to_thread ( time . sleep , <NUM_LIT:2> ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value [ "<STR_LIT:value>" ] is True <EOL> yield threads . defer_to_thread ( self . storage . get_node , node . id , cb = CalvinCB ( cb ) ) <EOL> yield threads . defer_to_thread ( time . sleep , <NUM_LIT:2> ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value [ "<STR_LIT:value>" ] == { '<STR_LIT>' : node . uri } <EOL> yield threads . defer_to_thread ( self . storage . delete_node , node , cb = CalvinCB ( cb ) ) <EOL> yield threads . defer_to_thread ( time . sleep , <NUM_LIT:2> ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value [ "<STR_LIT:value>" ] is True <EOL> yield threads . defer_to_thread ( self . storage . get_node , node . id , cb = CalvinCB ( cb ) ) <EOL> yield threads . defer_to_thread ( time . sleep , <NUM_LIT:2> ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value [ "<STR_LIT:value>" ] is None <EOL> @ pytest . inlineCallbacks <EOL> def test_application_functions ( self ) : <EOL> self . q = Queue . Queue ( ) <EOL> def cb ( key , value ) : <EOL> self . q . put ( { "<STR_LIT:key>" : key , "<STR_LIT:value>" : value } ) <EOL> yield threads . defer_to_thread ( time . sleep , <NUM_LIT:2> ) <EOL> application = appmanager . Application ( calvinuuid . uuid ( <EOL> '<STR_LIT>' ) , "<STR_LIT>" , [ calvinuuid . uuid ( '<STR_LIT>' ) , calvinuuid . uuid ( '<STR_LIT>' ) ] ) <EOL> yield threads . defer_to_thread ( self . storage . add_application , application , cb = CalvinCB ( cb ) ) <EOL> yield threads . defer_to_thread ( time . sleep , <NUM_LIT:2> ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value [ "<STR_LIT:value>" ] is True <EOL> yield threads . defer_to_thread ( self . storage . get_application , application . id , cb = CalvinCB ( cb ) ) <EOL> yield threads . defer_to_thread ( time . sleep , <NUM_LIT:2> ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value [ "<STR_LIT:value>" ] [ "<STR_LIT:name>" ] == application . name <EOL> yield threads . defer_to_thread ( self . storage . delete_application , application . id , cb = CalvinCB ( cb ) ) <EOL> yield threads . defer_to_thread ( time . sleep , <NUM_LIT:2> ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value [ "<STR_LIT:value>" ] is True <EOL> yield threads . defer_to_thread ( self . storage . get_application , application . id , cb = CalvinCB ( cb ) ) <EOL> yield threads . defer_to_thread ( time . sleep , <NUM_LIT:2> ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value [ "<STR_LIT:value>" ] is None <EOL> @ pytest . inlineCallbacks <EOL> def test_actor_functions ( self ) : <EOL> self . q = Queue . Queue ( ) <EOL> def cb ( key , value ) : <EOL> self . q . put ( { "<STR_LIT:key>" : key , "<STR_LIT:value>" : value } ) <EOL> yield threads . defer_to_thread ( time . sleep , <NUM_LIT:2> ) <EOL> port1 = TestPort ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> port2 = TestPort ( "<STR_LIT>" , "<STR_LIT>" , ) <EOL> port1 . peers = [ ( "<STR_LIT>" , port2 . id ) ] <EOL> port2 . peer = ( "<STR_LIT>" , port1 . id ) <EOL> actor = TestActor ( "<STR_LIT>" , "<STR_LIT>" , { } , { port1 . name : port1 } ) <EOL> yield threads . defer_to_thread ( self . storage . add_actor , actor , calvinuuid . uuid ( "<STR_LIT>" ) , cb = CalvinCB ( cb ) ) <EOL> yield threads . defer_to_thread ( time . sleep , <NUM_LIT:2> ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value [ "<STR_LIT:value>" ] is True <EOL> yield threads . defer_to_thread ( self . storage . get_actor , actor . id , cb = CalvinCB ( cb ) ) <EOL> yield threads . defer_to_thread ( time . sleep , <NUM_LIT:2> ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value [ "<STR_LIT:value>" ] [ "<STR_LIT:name>" ] == actor . name <EOL> yield threads . defer_to_thread ( self . storage . delete_actor , actor . id , cb = CalvinCB ( cb ) ) <EOL> yield threads . defer_to_thread ( time . sleep , <NUM_LIT:2> ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value [ "<STR_LIT:value>" ] is True <EOL> yield threads . defer_to_thread ( self . storage . get_actor , actor . id , cb = CalvinCB ( cb ) ) <EOL> yield threads . defer_to_thread ( time . sleep , <NUM_LIT:2> ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value [ "<STR_LIT:value>" ] is None <EOL> @ pytest . mark . interactive <EOL> class TestStorageNotStarted ( object ) : <EOL> def setup_class ( self ) : <EOL> self . storage = storage . Storage ( ) <EOL> def teardown_class ( self ) : <EOL> pass <EOL> def test_node_functions ( self ) : <EOL> self . q = Queue . Queue ( ) <EOL> def cb ( key , value ) : <EOL> self . q . put ( { "<STR_LIT:key>" : key , "<STR_LIT:value>" : value } ) <EOL> node = TestNode ( "<STR_LIT>" ) <EOL> self . storage . add_node ( node ) <EOL> value = self . storage . get_node ( node_id = node . id , cb = CalvinCB ( func = cb ) ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value [ "<STR_LIT:key>" ] == node . id and value [ "<STR_LIT:value>" ] == { '<STR_LIT>' : node . uri } <EOL> self . storage . delete_node ( node , cb = CalvinCB ( func = cb ) ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value <EOL> assert node . id not in self . storage . localstore <EOL> def test_application_functions ( self ) : <EOL> self . q = Queue . Queue ( ) <EOL> def cb ( key , value ) : <EOL> self . q . put ( { "<STR_LIT:key>" : key , "<STR_LIT:value>" : value } ) <EOL> application = appmanager . Application ( calvinuuid . uuid ( <EOL> '<STR_LIT>' ) , "<STR_LIT>" , [ calvinuuid . uuid ( '<STR_LIT>' ) , calvinuuid . uuid ( '<STR_LIT>' ) ] ) <EOL> self . storage . add_application ( application ) <EOL> value = self . storage . get_application ( <EOL> application . id , cb = CalvinCB ( func = cb ) ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value [ "<STR_LIT:key>" ] == application . id and value [ <EOL> "<STR_LIT:value>" ] [ "<STR_LIT:name>" ] == application . name <EOL> self . storage . delete_application ( application . id , cb = CalvinCB ( func = cb ) ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value <EOL> assert application . id not in self . storage . localstore <EOL> def test_actor_and_port_functions ( self ) : <EOL> self . q = Queue . Queue ( ) <EOL> def cb ( key , value ) : <EOL> self . q . put ( { "<STR_LIT:key>" : key , "<STR_LIT:value>" : value } ) <EOL> port1 = TestPort ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> port2 = TestPort ( "<STR_LIT>" , "<STR_LIT>" , ) <EOL> port1 . peers = [ ( "<STR_LIT>" , port2 . id ) ] <EOL> port2 . peer = ( "<STR_LIT>" , port1 . id ) <EOL> actor1 = TestActor ( "<STR_LIT>" , "<STR_LIT>" , { } , { port1 . name : port1 } ) <EOL> actor2 = TestActor ( "<STR_LIT>" , "<STR_LIT>" , { port2 . name : port2 } , { } ) <EOL> self . storage . add_actor ( actor1 , calvinuuid . uuid ( "<STR_LIT>" ) ) <EOL> value = self . storage . get_actor ( actor1 . id , cb = CalvinCB ( func = cb ) ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value [ "<STR_LIT:key>" ] == actor1 . id and value [ <EOL> "<STR_LIT:value>" ] [ "<STR_LIT:name>" ] == actor1 . name <EOL> assert value [ "<STR_LIT:value>" ] [ "<STR_LIT:name>" ] == actor1 . name <EOL> assert value [ "<STR_LIT:value>" ] [ "<STR_LIT:type>" ] == actor1 . _type <EOL> assert value [ "<STR_LIT:value>" ] [ "<STR_LIT>" ] == [ ] <EOL> assert value [ "<STR_LIT:value>" ] [ "<STR_LIT>" ] [ <NUM_LIT:0> ] [ "<STR_LIT:id>" ] == port1 . id <EOL> value = self . storage . get_port ( port1 . id , cb = CalvinCB ( func = cb ) ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value [ "<STR_LIT:key>" ] == port1 . id <EOL> assert value [ "<STR_LIT:value>" ] [ "<STR_LIT:name>" ] == port1 . name <EOL> assert value [ "<STR_LIT:value>" ] [ "<STR_LIT>" ] == port1 . direction <EOL> assert value [ "<STR_LIT:value>" ] [ "<STR_LIT>" ] == [ [ "<STR_LIT>" , port2 . id ] ] <EOL> self . storage . add_actor ( actor2 , calvinuuid . uuid ( "<STR_LIT>" ) ) <EOL> value = self . storage . get_actor ( actor2 . id , cb = CalvinCB ( func = cb ) ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value [ "<STR_LIT:key>" ] == actor2 . id <EOL> assert value [ "<STR_LIT:value>" ] [ "<STR_LIT:name>" ] == actor2 . name <EOL> assert value [ "<STR_LIT:value>" ] [ "<STR_LIT:type>" ] == actor2 . _type <EOL> assert value [ "<STR_LIT:value>" ] [ "<STR_LIT>" ] [ <NUM_LIT:0> ] [ "<STR_LIT:id>" ] == port2 . id <EOL> assert value [ "<STR_LIT:value>" ] [ "<STR_LIT>" ] == [ ] <EOL> value = self . storage . get_port ( port2 . id , cb = CalvinCB ( func = cb ) ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value [ "<STR_LIT:key>" ] == port2 . id <EOL> assert value [ "<STR_LIT:value>" ] [ "<STR_LIT:name>" ] == port2 . name <EOL> assert value [ "<STR_LIT:value>" ] [ "<STR_LIT>" ] == port2 . direction <EOL> assert value [ "<STR_LIT:value>" ] [ "<STR_LIT>" ] == [ "<STR_LIT>" , port1 . id ] <EOL> self . storage . delete_actor ( actor1 . id , cb = CalvinCB ( func = cb ) ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value <EOL> assert actor1 . id not in self . storage . localstore <EOL> self . storage . delete_port ( port1 . id , cb = CalvinCB ( func = cb ) ) <EOL> value = self . q . get ( timeout = <NUM_LIT> ) <EOL> assert value <EOL> assert port1 . id not in self . storage . localstore </s>
