<s> __author__ = '<STR_LIT>' <EOL> from pyon . ion . process import IonProcessThread <EOL> from pyon . ion . endpoint import ProcessRPCServer <EOL> from gevent . event import AsyncResult , Event <EOL> from gevent . coros import Semaphore <EOL> from gevent . timeout import Timeout <EOL> from pyon . util . unit_test import PyonTestCase <EOL> from pyon . util . int_test import IonIntegrationTestCase <EOL> from pyon . util . context import LocalContextMixin <EOL> from pyon . core . exception import IonException , NotFound , ContainerError , Timeout as IonTimeout <EOL> from pyon . util . async import spawn <EOL> from mock import sentinel , Mock , MagicMock , ANY , patch <EOL> from nose . plugins . attrib import attr <EOL> from pyon . net . endpoint import RPCClient <EOL> from pyon . ion . service import BaseService <EOL> from interface . objects import ProcessStateEnum <EOL> import time <EOL> import os <EOL> import unittest <EOL> @ attr ( '<STR_LIT>' , group = '<STR_LIT>' ) <EOL> class ProcessTest ( PyonTestCase ) : <EOL> class ExpectedFailure ( StandardError ) : <EOL> pass <EOL> def _make_service ( self ) : <EOL> """<STR_LIT>""" <EOL> svc = LocalContextMixin ( ) <EOL> svc . id = "<STR_LIT>" <EOL> svc . name = "<STR_LIT>" <EOL> svc . container = Mock ( ) <EOL> svc . container . context = LocalContextMixin ( ) <EOL> return svc <EOL> def test_spawn_proc_with_no_listeners ( self ) : <EOL> p = IonProcessThread ( name = sentinel . name , listeners = [ ] ) <EOL> p . start ( ) <EOL> p . get_ready_event ( ) . wait ( timeout = <NUM_LIT:5> ) <EOL> self . assertEquals ( len ( p . thread_manager . children ) , <NUM_LIT:1> ) <EOL> p . _notify_stop ( ) <EOL> self . assertTrue ( p . thread_manager . children [ <NUM_LIT:0> ] . proc . dead ) <EOL> p . stop ( ) <EOL> def test_spawn_proc_with_one_listener ( self ) : <EOL> mocklistener = Mock ( spec = ProcessRPCServer ) <EOL> p = IonProcessThread ( name = sentinel . name , listeners = [ mocklistener ] ) <EOL> readyev = Event ( ) <EOL> readyev . set ( ) <EOL> mocklistener . get_ready_event . return_value = readyev <EOL> p . start ( ) <EOL> p . get_ready_event ( ) . wait ( timeout = <NUM_LIT:5> ) <EOL> p . start_listeners ( ) <EOL> self . assertEquals ( len ( p . thread_manager . children ) , <NUM_LIT:2> ) <EOL> mocklistener . listen . assert_called_once_with ( thread_name = ANY ) <EOL> self . assertEqual ( mocklistener . routing_call , p . _routing_call ) <EOL> p . _notify_stop ( ) <EOL> mocklistener . close . assert_called_once_with ( ) <EOL> p . stop ( ) <EOL> @ unittest . skip ( "<STR_LIT>" ) <EOL> def test_spawn_with_listener_failure ( self ) : <EOL> mocklistener = Mock ( spec = ProcessRPCServer ) <EOL> mocklistener . listen . side_effect = self . ExpectedFailure <EOL> readyev = Event ( ) <EOL> readyev . set ( ) <EOL> mocklistener . get_ready_event . return_value = readyev <EOL> p = IonProcessThread ( name = sentinel . name , listeners = [ mocklistener ] ) <EOL> p . start ( ) <EOL> p . get_ready_event ( ) . wait ( timeout = <NUM_LIT:5> ) <EOL> p . start_listeners ( ) <EOL> self . assertTrue ( p . proc . dead ) <EOL> self . assertIsInstance ( p . proc . exception , self . ExpectedFailure ) <EOL> self . assertRaises ( self . ExpectedFailure , p . _notify_stop ) <EOL> self . assertTrue ( p . thread_manager . children [ - <NUM_LIT:1> ] . proc . dead ) <EOL> p . stop ( ) <EOL> def test__routing_call ( self ) : <EOL> svc = self . _make_service ( ) <EOL> p = IonProcessThread ( name = sentinel . name , listeners = [ ] , service = svc ) <EOL> p . start ( ) <EOL> p . get_ready_event ( ) . wait ( timeout = <NUM_LIT:5> ) <EOL> ar = AsyncResult ( ) <EOL> p . _routing_call ( ar . set , None , value = sentinel . callarg ) <EOL> v = ar . get ( timeout = <NUM_LIT:5> ) <EOL> self . assertEquals ( v , sentinel . callarg ) <EOL> p . _notify_stop ( ) <EOL> p . stop ( ) <EOL> def test_competing__routing_call ( self ) : <EOL> svc = self . _make_service ( ) <EOL> p = IonProcessThread ( name = sentinel . name , listeners = [ ] , service = svc ) <EOL> p . start ( ) <EOL> p . get_ready_event ( ) . wait ( timeout = <NUM_LIT:5> ) <EOL> sem = Semaphore ( ) <EOL> def thecall ( ar = None ) : <EOL> semres = sem . acquire ( blocking = False ) <EOL> if not semres : <EOL> raise StandardError ( "<STR_LIT>" ) <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> sem . release ( ) <EOL> ar . set ( True ) <EOL> ar1 = AsyncResult ( ) <EOL> ar2 = AsyncResult ( ) <EOL> ar3 = AsyncResult ( ) <EOL> p . _routing_call ( thecall , None , ar = ar3 ) <EOL> p . _routing_call ( thecall , None , ar = ar1 ) <EOL> p . _routing_call ( thecall , None , ar = ar2 ) <EOL> ar1 . get ( timeout = <NUM_LIT:5> ) <EOL> ar2 . get ( timeout = <NUM_LIT:5> ) <EOL> ar3 . get ( timeout = <NUM_LIT:5> ) <EOL> p . _notify_stop ( ) <EOL> p . stop ( ) <EOL> def test_known_error ( self ) : <EOL> svc = self . _make_service ( ) <EOL> p = IonProcessThread ( name = sentinel . name , listeners = [ ] , service = svc ) <EOL> p . start ( ) <EOL> p . get_ready_event ( ) . wait ( timeout = <NUM_LIT:5> ) <EOL> self . addCleanup ( p . stop ) <EOL> def proc_call ( ) : <EOL> raise NotFound ( "<STR_LIT>" ) <EOL> def client_call ( p = None , ar = None ) : <EOL> try : <EOL> ca = p . _routing_call ( proc_call , None ) <EOL> ca . get ( timeout = <NUM_LIT:5> ) <EOL> except IonException as e : <EOL> ar . set ( e ) <EOL> ar = AsyncResult ( ) <EOL> gl_call = spawn ( client_call , p = p , ar = ar ) <EOL> e = ar . get ( timeout = <NUM_LIT:5> ) <EOL> self . assertIsInstance ( e , NotFound ) <EOL> def test_unknown_error ( self ) : <EOL> svc = self . _make_service ( ) <EOL> p = IonProcessThread ( name = sentinel . name , listeners = [ ] , service = svc ) <EOL> p . start ( ) <EOL> p . get_ready_event ( ) . wait ( timeout = <NUM_LIT:5> ) <EOL> self . addCleanup ( p . stop ) <EOL> def proc_call ( ) : <EOL> raise self . ExpectedError ( "<STR_LIT>" ) <EOL> def client_call ( p = None , ar = None ) : <EOL> try : <EOL> ca = p . _routing_call ( proc_call , None ) <EOL> ca . get ( timeout = <NUM_LIT:5> ) <EOL> except IonException as e : <EOL> ar . set ( e ) <EOL> ar = AsyncResult ( ) <EOL> gl_call = spawn ( client_call , p = p , ar = ar ) <EOL> e = ar . get ( timeout = <NUM_LIT:5> ) <EOL> self . assertIsInstance ( e , ContainerError ) <EOL> self . assertEquals ( len ( p . _errors ) , <NUM_LIT:1> ) <EOL> def test_has_pending_call ( self ) : <EOL> svc = self . _make_service ( ) <EOL> p = IonProcessThread ( name = sentinel . name , listeners = [ ] , service = svc ) <EOL> ar = p . _routing_call ( sentinel . call , MagicMock ( ) ) <EOL> self . assertTrue ( p . has_pending_call ( ar ) ) <EOL> def test_has_pending_call_with_no_call ( self ) : <EOL> svc = self . _make_service ( ) <EOL> p = IonProcessThread ( name = sentinel . name , listeners = [ ] , service = svc ) <EOL> ar = p . _routing_call ( sentinel . call , MagicMock ( ) ) <EOL> p . _ctrl_queue . get ( ) <EOL> self . assertFalse ( p . has_pending_call ( ar ) ) <EOL> def test__cancel_pending_call ( self ) : <EOL> svc = self . _make_service ( ) <EOL> p = IonProcessThread ( name = sentinel . name , listeners = [ ] , service = svc ) <EOL> ar = p . _routing_call ( sentinel . call , MagicMock ( ) ) <EOL> val = p . _cancel_pending_call ( ar ) <EOL> self . assertTrue ( val ) <EOL> self . assertTrue ( ar . ready ( ) ) <EOL> def test__cancel_pending_call_with_no_call ( self ) : <EOL> svc = self . _make_service ( ) <EOL> p = IonProcessThread ( name = sentinel . name , listeners = [ ] , service = svc ) <EOL> ar = p . _routing_call ( sentinel . call , MagicMock ( ) ) <EOL> p . _ctrl_queue . get ( ) <EOL> val = p . _cancel_pending_call ( ar ) <EOL> self . assertFalse ( val ) <EOL> def test__interrupt_control_thread ( self ) : <EOL> svc = self . _make_service ( ) <EOL> p = IonProcessThread ( name = sentinel . name , listeners = [ ] , service = svc ) <EOL> p . start ( ) <EOL> p . get_ready_event ( ) . wait ( timeout = <NUM_LIT:5> ) <EOL> self . addCleanup ( p . stop ) <EOL> waitar = AsyncResult ( ) <EOL> callar = AsyncResult ( ) <EOL> def spin ( inar , outar ) : <EOL> outar . set ( True ) <EOL> inar . wait ( ) <EOL> ar = p . _routing_call ( spin , MagicMock ( ) , callar , waitar ) <EOL> waitar . get ( timeout = <NUM_LIT:2> ) <EOL> p . _interrupt_control_thread ( ) <EOL> self . assertFalse ( ar . ready ( ) ) <EOL> ar2 = p . _routing_call ( callar . set , MagicMock ( ) , sentinel . val ) <EOL> ar2 . get ( timeout = <NUM_LIT:2> ) <EOL> self . assertTrue ( callar . ready ( ) ) <EOL> self . assertEquals ( callar . get ( ) , sentinel . val ) <EOL> def test__control_flow_cancelled_call ( self ) : <EOL> svc = self . _make_service ( ) <EOL> p = IonProcessThread ( name = sentinel . name , listeners = [ ] , service = svc ) <EOL> p . start ( ) <EOL> p . get_ready_event ( ) . wait ( timeout = <NUM_LIT:5> ) <EOL> self . addCleanup ( p . stop ) <EOL> waitar = AsyncResult ( ) <EOL> callar = AsyncResult ( ) <EOL> def spin ( inar , outar ) : <EOL> outar . set ( True ) <EOL> inar . wait ( ) <EOL> ar = p . _routing_call ( spin , MagicMock ( ) , callar , waitar ) <EOL> futurear = AsyncResult ( ) <EOL> ar2 = p . _routing_call ( futurear . set , MagicMock ( ) , sentinel . val ) <EOL> waitar . get ( timeout = <NUM_LIT:2> ) <EOL> p . cancel_or_abort_call ( ar2 ) <EOL> callar . set ( ) <EOL> ar . get ( timeout = <NUM_LIT:2> ) <EOL> self . assertRaises ( Timeout , futurear . get , timeout = <NUM_LIT:2> ) <EOL> self . assertTrue ( ar2 . ready ( ) ) <EOL> def test__control_flow_expired_call ( self ) : <EOL> svc = self . _make_service ( ) <EOL> p = IonProcessThread ( name = sentinel . name , listeners = [ ] , service = svc ) <EOL> p . start ( ) <EOL> p . get_ready_event ( ) . wait ( timeout = <NUM_LIT:5> ) <EOL> self . addCleanup ( p . stop ) <EOL> def make_call ( call , ctx , val ) : <EOL> ar = p . _routing_call ( call , ctx , val ) <EOL> return ar . get ( timeout = <NUM_LIT:10> ) <EOL> ctx = { '<STR_LIT>' : <NUM_LIT:0> } <EOL> futurear = AsyncResult ( ) <EOL> with patch ( '<STR_LIT>' ) as gcm : <EOL> waitar = AsyncResult ( ) <EOL> gcm . getcurrent ( ) . kill . side_effect = lambda * a , ** k : waitar . set ( ) <EOL> ar = p . _routing_call ( futurear . set , ctx , sentinel . val ) <EOL> waitar . get ( timeout = <NUM_LIT:10> ) <EOL> self . assertFalse ( futurear . ready ( ) ) <EOL> self . assertFalse ( ar . ready ( ) ) <EOL> self . assertEquals ( gcm . getcurrent ( ) . kill . call_count , <NUM_LIT:1> ) <EOL> self . assertIsInstance ( gcm . getcurrent ( ) . kill . call_args [ <NUM_LIT:1> ] [ '<STR_LIT>' ] , IonTimeout ) <EOL> futurear2 = AsyncResult ( ) <EOL> ar2 = p . _routing_call ( futurear2 . set , MagicMock ( ) , sentinel . val2 ) <EOL> ar2 . get ( timeout = <NUM_LIT:2> ) <EOL> def test_heartbeat_no_listeners ( self ) : <EOL> svc = self . _make_service ( ) <EOL> p = IonProcessThread ( name = sentinel . name , listeners = [ ] , service = svc ) <EOL> p . start ( ) <EOL> p . get_ready_event ( ) . wait ( timeout = <NUM_LIT:5> ) <EOL> self . addCleanup ( p . stop ) <EOL> hb = p . heartbeat ( ) <EOL> self . assertEquals ( ( True , True , True ) , hb ) <EOL> self . assertEquals ( <NUM_LIT:0> , p . _heartbeat_count ) <EOL> self . assertIsNone ( p . _heartbeat_op ) <EOL> def test_heartbeat_with_listeners ( self ) : <EOL> mocklistener = Mock ( spec = ProcessRPCServer ) <EOL> svc = self . _make_service ( ) <EOL> p = IonProcessThread ( name = sentinel . name , listeners = [ mocklistener ] , service = svc ) <EOL> readyev = Event ( ) <EOL> readyev . set ( ) <EOL> mocklistener . get_ready_event . return_value = readyev <EOL> def fake_listen ( evout , evin ) : <EOL> evout . set ( True ) <EOL> evin . wait ( ) <EOL> listenoutev = AsyncResult ( ) <EOL> listeninev = Event ( ) <EOL> mocklistener . listen = lambda * a , ** kw : fake_listen ( listenoutev , listeninev ) <EOL> p . start ( ) <EOL> p . get_ready_event ( ) . wait ( timeout = <NUM_LIT:5> ) <EOL> p . start_listeners ( ) <EOL> listenoutev . wait ( timeout = <NUM_LIT:5> ) <EOL> self . addCleanup ( listeninev . set ) <EOL> self . addCleanup ( p . stop ) <EOL> hb = p . heartbeat ( ) <EOL> self . assertEquals ( ( True , True , True ) , hb ) <EOL> self . assertEquals ( <NUM_LIT:0> , p . _heartbeat_count ) <EOL> self . assertIsNone ( p . _heartbeat_op ) <EOL> def test_heartbeat_listener_dead ( self ) : <EOL> mocklistener = Mock ( spec = ProcessRPCServer ) <EOL> svc = self . _make_service ( ) <EOL> p = IonProcessThread ( name = sentinel . name , listeners = [ mocklistener ] , service = svc ) <EOL> readyev = Event ( ) <EOL> readyev . set ( ) <EOL> mocklistener . get_ready_event . return_value = readyev <EOL> def fake_listen ( evout , evin ) : <EOL> evout . set ( True ) <EOL> evin . wait ( ) <EOL> listenoutev = AsyncResult ( ) <EOL> listeninev = Event ( ) <EOL> p . start ( ) <EOL> p . get_ready_event ( ) . wait ( timeout = <NUM_LIT:5> ) <EOL> p . start_listeners ( ) <EOL> listenoutev . wait ( timeout = <NUM_LIT:5> ) <EOL> self . addCleanup ( listeninev . set ) <EOL> self . addCleanup ( p . stop ) <EOL> listeninev . set ( ) <EOL> p . thread_manager . children [ <NUM_LIT:1> ] . join ( timeout = <NUM_LIT:5> ) <EOL> hb = p . heartbeat ( ) <EOL> self . assertEquals ( ( False , True , True ) , hb ) <EOL> self . assertEquals ( <NUM_LIT:0> , p . _heartbeat_count ) <EOL> self . assertIsNone ( p . _heartbeat_op ) <EOL> def test_heartbeat_ctrl_thread_dead ( self ) : <EOL> svc = self . _make_service ( ) <EOL> p = IonProcessThread ( name = sentinel . name , listeners = [ ] , service = svc ) <EOL> p . start ( ) <EOL> p . get_ready_event ( ) . wait ( timeout = <NUM_LIT:5> ) <EOL> self . addCleanup ( p . stop ) <EOL> p . _ctrl_thread . stop ( ) <EOL> hb = p . heartbeat ( ) <EOL> self . assertEquals ( ( True , False , True ) , hb ) <EOL> self . assertEquals ( <NUM_LIT:0> , p . _heartbeat_count ) <EOL> self . assertIsNone ( p . _heartbeat_op ) <EOL> def test_heartbeat_with_current_op ( self ) : <EOL> svc = self . _make_service ( ) <EOL> p = IonProcessThread ( name = sentinel . name , listeners = [ ] , service = svc ) <EOL> p . start ( ) <EOL> p . get_ready_event ( ) . wait ( timeout = <NUM_LIT:5> ) <EOL> p . _ctrl_thread . ev_exit . set ( ) <EOL> def fake_op ( evout , evin ) : <EOL> evout . set ( True ) <EOL> evin . wait ( ) <EOL> listenoutev = AsyncResult ( ) <EOL> listeninev = Event ( ) <EOL> self . addCleanup ( listeninev . set ) <EOL> self . addCleanup ( p . stop ) <EOL> ar = p . _routing_call ( fake_op , None , listenoutev , listeninev ) <EOL> listenoutev . wait ( timeout = <NUM_LIT:5> ) <EOL> hb = p . heartbeat ( ) <EOL> self . assertEquals ( ( True , True , True ) , hb ) <EOL> self . assertEquals ( <NUM_LIT:1> , p . _heartbeat_count ) <EOL> self . assertEquals ( ar , p . _heartbeat_op ) <EOL> self . assertIsNotNone ( p . _heartbeat_time ) <EOL> self . assertIsNotNone ( p . _heartbeat_stack ) <EOL> self . assertIn ( "<STR_LIT>" , str ( p . _heartbeat_stack ) ) <EOL> def test_heartbeat_with_current_op_multiple_times ( self ) : <EOL> svc = self . _make_service ( ) <EOL> p = IonProcessThread ( name = sentinel . name , listeners = [ ] , service = svc ) <EOL> p . start ( ) <EOL> p . get_ready_event ( ) . wait ( timeout = <NUM_LIT:5> ) <EOL> p . _ctrl_thread . ev_exit . set ( ) <EOL> def fake_op ( evout , evin ) : <EOL> evout . set ( True ) <EOL> evin . wait ( ) <EOL> listenoutev = AsyncResult ( ) <EOL> listeninev = Event ( ) <EOL> self . addCleanup ( listeninev . set ) <EOL> self . addCleanup ( p . stop ) <EOL> ar = p . _routing_call ( fake_op , None , listenoutev , listeninev ) <EOL> listenoutev . wait ( timeout = <NUM_LIT:5> ) <EOL> for x in xrange ( <NUM_LIT:5> ) : <EOL> hb = p . heartbeat ( ) <EOL> self . assertEquals ( ( True , True , True ) , hb ) <EOL> self . assertEquals ( <NUM_LIT:5> , p . _heartbeat_count ) <EOL> self . assertEquals ( ar , p . _heartbeat_op ) <EOL> def test_heartbeat_current_op_over_limit ( self ) : <EOL> self . patch_cfg ( '<STR_LIT>' , { '<STR_LIT>' : { '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT:2> } } } ) <EOL> svc = self . _make_service ( ) <EOL> p = IonProcessThread ( name = sentinel . name , listeners = [ ] , service = svc ) <EOL> p . start ( ) <EOL> p . get_ready_event ( ) . wait ( timeout = <NUM_LIT:5> ) <EOL> p . _ctrl_thread . ev_exit . set ( ) <EOL> def fake_op ( evout , evin ) : <EOL> evout . set ( True ) <EOL> evin . wait ( ) <EOL> listenoutev = AsyncResult ( ) <EOL> listeninev = Event ( ) <EOL> self . addCleanup ( listeninev . set ) <EOL> self . addCleanup ( p . stop ) <EOL> ar = p . _routing_call ( fake_op , None , listenoutev , listeninev ) <EOL> listenoutev . wait ( timeout = <NUM_LIT:5> ) <EOL> for x in xrange ( <NUM_LIT:3> ) : <EOL> hb = p . heartbeat ( ) <EOL> self . assertEquals ( ( True , True , False ) , hb ) <EOL> class FakeService ( BaseService ) : <EOL> """<STR_LIT>""" <EOL> name = '<STR_LIT>' <EOL> dependencies = [ ] <EOL> def takes_too_long ( self , noticear = None ) : <EOL> if noticear is not None : <EOL> noticear . set ( True ) <EOL> ar = AsyncResult ( ) <EOL> ar . wait ( ) <EOL> @ attr ( '<STR_LIT>' , group = '<STR_LIT>' ) <EOL> @ unittest . skip ( "<STR_LIT>" ) <EOL> class TestProcessInt ( IonIntegrationTestCase ) : <EOL> def setUp ( self ) : <EOL> self . _start_container ( ) <EOL> self . pid = self . container . spawn_process ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . fsclient = RPCClient ( to_name = '<STR_LIT>' ) <EOL> @ unittest . skip ( "<STR_LIT>" ) <EOL> def test_timeout_with_messaging ( self ) : <EOL> with self . assertRaises ( IonTimeout ) as cm : <EOL> self . fsclient . request ( { } , op = '<STR_LIT>' , timeout = <NUM_LIT:5> ) <EOL> self . assertIn ( '<STR_LIT>' , cm . exception . message ) <EOL> @ unittest . skipIf ( os . getenv ( '<STR_LIT>' , False ) , "<STR_LIT>" ) <EOL> @ unittest . skip ( "<STR_LIT>" ) <EOL> def test_heartbeat_failure ( self ) : <EOL> self . patch_cfg ( '<STR_LIT>' , { '<STR_LIT>' : { '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT:2> , '<STR_LIT>' : <NUM_LIT:1.0> } } } ) <EOL> svc = self . container . proc_manager . procs [ self . pid ] <EOL> ip = svc . _process <EOL> stopar = AsyncResult ( ) <EOL> self . container . proc_manager . add_proc_state_changed_callback ( lambda * args : stopar . set ( args ) ) <EOL> noticear = AsyncResult ( ) <EOL> ar = ip . _routing_call ( svc . takes_too_long , None , noticear = noticear ) <EOL> noticear . get ( timeout = <NUM_LIT:10> ) <EOL> for x in xrange ( <NUM_LIT:2> ) : <EOL> ip . heartbeat ( ) <EOL> ip . _ctrl_thread . join ( timeout = <NUM_LIT:5> ) <EOL> stopargs = stopar . get ( timeout = <NUM_LIT:5> ) <EOL> self . assertEquals ( stopargs , ( svc , ProcessStateEnum . FAILED , self . container ) ) <EOL> self . assertEquals ( len ( self . container . proc_manager . procs ) , <NUM_LIT:0> ) </s>
