<s> __author__ = '<STR_LIT>' <EOL> from pyon . core . thread import PyonThreadManager , PyonThread <EOL> from pyon . core . exception import ContainerError <EOL> from pyon . util . int_test import IonIntegrationTestCase <EOL> from pyon . util . unit_test import PyonTestCase <EOL> from unittest import SkipTest <EOL> from nose . plugins . attrib import attr <EOL> import time <EOL> @ attr ( '<STR_LIT>' , group = '<STR_LIT>' ) <EOL> class ProcessTest ( PyonTestCase ) : <EOL> def setUp ( self ) : <EOL> self . counter = <NUM_LIT:0> <EOL> def increment ( self , amount = <NUM_LIT:1> ) : <EOL> self . counter += amount <EOL> def test_proc ( self ) : <EOL> self . counter = <NUM_LIT:0> <EOL> proc = PyonThread ( self . increment , <NUM_LIT:2> ) <EOL> proc . start ( ) <EOL> self . assertEqual ( self . counter , <NUM_LIT:0> ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> proc . join ( ) <EOL> self . assertGreaterEqual ( self . counter , <NUM_LIT:2> ) <EOL> def test_supervisor ( self ) : <EOL> self . counter = <NUM_LIT:0> <EOL> sup = PyonThreadManager ( ) <EOL> sup . start ( ) <EOL> proc = sup . spawn ( self . increment , amount = <NUM_LIT:2> ) <EOL> self . assertEqual ( self . counter , <NUM_LIT:0> ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> sup . join_children ( ) <EOL> self . assertGreaterEqual ( self . counter , <NUM_LIT:2> ) <EOL> def test_supervisor_shutdown ( self ) : <EOL> """<STR_LIT>""" <EOL> sup = PyonThreadManager ( ) <EOL> sup . start ( ) <EOL> import gevent <EOL> self . assertIs ( time . sleep , gevent . hub . sleep ) <EOL> proc_sleep_secs , proc_count = <NUM_LIT> , <NUM_LIT:5> <EOL> [ sup . spawn ( time . sleep , seconds = proc_sleep_secs ) for i in xrange ( <NUM_LIT:5> ) ] <EOL> elapsed = sup . shutdown ( <NUM_LIT:2> * proc_sleep_secs ) <EOL> self . assertLess ( elapsed - proc_sleep_secs , <NUM_LIT> ) <EOL> self . assertLess ( elapsed , <NUM_LIT> + proc_sleep_secs * <NUM_LIT:3> ) <EOL> wait_secs = <NUM_LIT> <EOL> [ sup . spawn ( time . sleep , seconds = proc_sleep_secs ) for i in xrange ( <NUM_LIT:5> ) ] <EOL> elapsed = sup . shutdown ( wait_secs ) <EOL> self . assertLess ( elapsed - wait_secs , <NUM_LIT> ) <EOL> self . assertLess ( elapsed , <NUM_LIT> + proc_sleep_secs ) <EOL> [ sup . spawn ( time . sleep , seconds = proc_sleep_secs ) for i in xrange ( <NUM_LIT:5> ) ] <EOL> elapsed = sup . shutdown ( ) <EOL> self . assertLess ( elapsed - proc_sleep_secs , <NUM_LIT> ) <EOL> def test_ensure_ready ( self ) : <EOL> sup = PyonThreadManager ( ) <EOL> sup . start ( ) <EOL> proc = sup . spawn ( self . increment , amount = <NUM_LIT:5> ) <EOL> sup . ensure_ready ( proc ) <EOL> self . assertEqual ( self . counter , <NUM_LIT:5> ) <EOL> def test_ensure_ready_failed_proc ( self ) : <EOL> def failboat ( ) : <EOL> self . increment ( <NUM_LIT:5> , <NUM_LIT:1> ) <EOL> sup = PyonThreadManager ( ) <EOL> sup . start ( ) <EOL> proc = sup . spawn ( failboat ) <EOL> self . assertRaises ( ContainerError , sup . ensure_ready , proc ) </s>
