<s> import unittest <EOL> import os <EOL> import sys <EOL> from ufora . util . ManagedThread import ManagedThread <EOL> import ufora . util . ThreadLocalStack as ThreadLocalStack <EOL> class TestPushable ( ThreadLocalStack . ThreadLocalStackPushable ) : <EOL> def __init__ ( self , ix ) : <EOL> ThreadLocalStack . ThreadLocalStackPushable . __init__ ( self ) <EOL> self . ix = ix <EOL> class TestManagedThread ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . testValue = <NUM_LIT:0> <EOL> self . exception = None <EOL> def stackCheckTest ( self ) : <EOL> try : <EOL> current = TestPushable . getCurrent ( ) <EOL> self . assertIsNotNone ( current ) <EOL> except Exception as e : <EOL> self . exception = e <EOL> def incrementTestValue ( self ) : <EOL> self . testValue += <NUM_LIT:1> <EOL> def setTestValue ( self , value ) : <EOL> self . testValue = value <EOL> def allocateTooMuchMemory ( self ) : <EOL> x = "<STR_LIT:x>" * ( <NUM_LIT> * <NUM_LIT> * <NUM_LIT> * <NUM_LIT> ) <EOL> def test_createManagedThread ( self ) : <EOL> def doNothing ( ) : <EOL> pass <EOL> thread = ManagedThread ( target = doNothing ) <EOL> def test_daemonByDefault ( self ) : <EOL> def doNothing ( ) : <EOL> pass <EOL> thread = ManagedThread ( target = doNothing ) <EOL> self . assertTrue ( thread . daemon ) <EOL> def test_thread_local_storage_dict ( self ) : <EOL> tester = TestPushable ( <NUM_LIT:2> ) <EOL> with tester : <EOL> thread = ManagedThread ( target = self . stackCheckTest ) <EOL> thread . start ( ) <EOL> thread . join ( ) <EOL> if self . exception is not None : <EOL> raise self . exception <EOL> def test_startManagedThread ( self ) : <EOL> thread = ManagedThread ( target = self . incrementTestValue ) <EOL> thread . start ( ) <EOL> thread . join ( ) <EOL> self . assertEqual ( self . testValue , <NUM_LIT:1> ) <EOL> def test_startWithArgument ( self ) : <EOL> thread = ManagedThread ( target = self . setTestValue , args = ( <NUM_LIT:2> , ) ) <EOL> thread . start ( ) <EOL> thread . join ( ) <EOL> self . assertEqual ( self . testValue , <NUM_LIT:2> ) <EOL> def test_outOfMemory ( self ) : <EOL> def handleException ( ex ) : <EOL> self . assertTrue ( isinstance ( ex , Exception ) ) <EOL> self . setTestValue ( <NUM_LIT:3> ) <EOL> sys . stderr . flush ( ) <EOL> newstderr = os . dup ( sys . stderr . fileno ( ) ) <EOL> devnull = open ( os . devnull , '<STR_LIT:w>' ) <EOL> os . dup2 ( devnull . fileno ( ) , sys . stderr . fileno ( ) ) <EOL> devnull . close ( ) <EOL> thread = ManagedThread ( target = self . allocateTooMuchMemory ) <EOL> thread . criticalErrorHandler = handleException <EOL> thread . start ( ) <EOL> thread . join ( ) <EOL> os . dup2 ( newstderr , sys . stderr . fileno ( ) ) <EOL> self . assertEqual ( self . testValue , <NUM_LIT:3> ) </s>
