<s> import random <EOL> import fixtures <EOL> import mock <EOL> import six <EOL> import testtools <EOL> from sahara import context <EOL> from sahara import exceptions as ex <EOL> rnd = random . Random ( ) <EOL> class ContextTest ( testtools . TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( ContextTest , self ) . setUp ( ) <EOL> self . useFixture ( fixtures . FakeLogger ( '<STR_LIT>' ) ) <EOL> ctx = context . Context ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , { } , <EOL> remote_semaphore = '<STR_LIT>' ) <EOL> context . set_ctx ( ctx ) <EOL> def _add_element ( self , lst , i ) : <EOL> context . sleep ( rnd . uniform ( <NUM_LIT:0> , <NUM_LIT:0.1> ) ) <EOL> lst . append ( i ) <EOL> def _raise_test_exc ( self , exc_msg ) : <EOL> raise TestException ( exc_msg ) <EOL> def test_thread_group_waits_threads ( self ) : <EOL> for _ in six . moves . range ( <NUM_LIT:5> ) : <EOL> lst = [ ] <EOL> with context . ThreadGroup ( ) as tg : <EOL> for i in six . moves . range ( <NUM_LIT> ) : <EOL> tg . spawn ( '<STR_LIT>' % i , self . _add_element , lst , i ) <EOL> self . assertEqual ( <NUM_LIT> , len ( lst ) ) <EOL> def test_thread_group_waits_threads_if_spawning_exception ( self ) : <EOL> lst = [ ] <EOL> with testtools . ExpectedException ( RuntimeError ) : <EOL> with context . ThreadGroup ( ) as tg : <EOL> for i in six . moves . range ( <NUM_LIT> ) : <EOL> tg . spawn ( '<STR_LIT>' % i , self . _add_element , lst , i ) <EOL> raise RuntimeError ( ) <EOL> self . assertEqual ( <NUM_LIT> , len ( lst ) ) <EOL> def test_thread_group_waits_threads_if_child_exception ( self ) : <EOL> lst = [ ] <EOL> with testtools . ExpectedException ( ex . ThreadException ) : <EOL> with context . ThreadGroup ( ) as tg : <EOL> tg . spawn ( '<STR_LIT>' , self . _raise_test_exc , '<STR_LIT>' ) <EOL> for i in six . moves . range ( <NUM_LIT> ) : <EOL> tg . spawn ( '<STR_LIT>' % i , self . _add_element , lst , i ) <EOL> self . assertEqual ( <NUM_LIT> , len ( lst ) ) <EOL> def test_thread_group_handles_spawning_exception ( self ) : <EOL> with testtools . ExpectedException ( TestException ) : <EOL> with context . ThreadGroup ( ) : <EOL> raise TestException ( ) <EOL> def test_thread_group_handles_child_exception ( self ) : <EOL> try : <EOL> with context . ThreadGroup ( ) as tg : <EOL> tg . spawn ( '<STR_LIT>' , self . _raise_test_exc , '<STR_LIT>' ) <EOL> except ex . ThreadException as te : <EOL> self . assertIn ( '<STR_LIT>' , six . text_type ( te ) ) <EOL> self . assertIn ( '<STR_LIT>' , six . text_type ( te ) ) <EOL> def test_thread_group_prefers_spawning_exception ( self ) : <EOL> with testtools . ExpectedException ( RuntimeError ) : <EOL> with context . ThreadGroup ( ) as tg : <EOL> tg . spawn ( '<STR_LIT>' , self . _raise_test_exc , '<STR_LIT>' ) <EOL> raise RuntimeError ( ) <EOL> def test_wrapper_does_not_set_exception ( self ) : <EOL> func = mock . MagicMock ( ) <EOL> tg = mock . MagicMock ( exc = None , failed_thread = None ) <EOL> context . _wrapper ( None , '<STR_LIT>' , tg , func ) <EOL> self . assertIsNone ( tg . exc ) <EOL> self . assertIsNone ( tg . failed_thread ) <EOL> def test_wrapper_catches_base_exception ( self ) : <EOL> func = mock . MagicMock ( ) <EOL> func . side_effect = BaseException ( ) <EOL> tg = mock . MagicMock ( exc = None , failed_thread = None ) <EOL> context . _wrapper ( None , '<STR_LIT>' , tg , func ) <EOL> self . assertIsNotNone ( tg . exc ) <EOL> self . assertEqual ( '<STR_LIT>' , tg . failed_thread ) <EOL> def test_is_auth_capable_for_admin_ctx ( self ) : <EOL> ctx = context . ctx ( ) <EOL> self . assertFalse ( ctx . is_auth_capable ( ) ) <EOL> def test_is_auth_capable_for_user_ctx ( self ) : <EOL> existing_ctx = context . ctx ( ) <EOL> try : <EOL> ctx = context . Context ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> { "<STR_LIT>" : "<STR_LIT>" } , remote_semaphore = '<STR_LIT>' ) <EOL> self . assertTrue ( ctx . is_auth_capable ( ) ) <EOL> finally : <EOL> context . set_ctx ( existing_ctx ) <EOL> class TestException ( Exception ) : <EOL> pass </s>
