#NEWLINE# def test_timeout_thread( self ) : #NEWLINE# #INDENT# 'Create and kill thread by timeout.\n\n        This single test covers 3 methods: terminate_thread, timeout_thread,\n        and interruptable_sleep.\n\n        This test is more like integrated then unit, but it is much better\n        then unreadable 500 lines of mocking and checking.\n        ' #NEWLINE# queue = Queue . Queue ( ) #NEWLINE# killer_thread = threading . Thread ( target = utils . timeout_thread , args = ( queue , ) ) #NEWLINE# test_thread = threading . Thread ( target = utils . interruptable_sleep , args = ( 30 , 0.01 ) ) #NEWLINE# test_thread . start ( ) #NEWLINE# start_time = time . time ( ) #NEWLINE# queue . put ( ( test_thread . ident , ( start_time + 1 ) ) ) #NEWLINE# killer_thread . start ( ) #NEWLINE# test_thread . join ( ) #NEWLINE# end_time = time . time ( ) #NEWLINE# queue . put ( ( None , None ) ) #NEWLINE# killer_thread . join ( ) #NEWLINE# time_elapsed = ( end_time - start_time ) #NEWLINE# self . assertTrue ( ( time_elapsed < 11 ) , ( 'Thread killed too late (%s seconds)' % test_thread ) )