#NEWLINE# def test_interrupt_signal( self ) : #NEWLINE# #INDENT# if sys . platform . startswith ( 'java' ) : #NEWLINE# #INDENT# raise SkipTest ( "Can't test interrupts in Jython" ) #NEWLINE# #UNINDENT# db = self . client . pymongo_test #NEWLINE# where = delay ( 1.5 ) #NEWLINE# db . drop_collection ( 'foo' ) #NEWLINE# db . foo . insert_one ( { #NEWLINE# #INDENT# '_id' : 1 , #NEWLINE# #UNINDENT# } ) #NEWLINE# def interrupter( ) : #NEWLINE# #INDENT# time . sleep ( 0.25 ) #NEWLINE# thread . interrupt_main ( ) #NEWLINE# #UNINDENT# thread . start_new_thread ( interrupter , ( ) ) #NEWLINE# raised = False #NEWLINE# try : #NEWLINE# #INDENT# next ( db . foo . find ( { #NEWLINE# #INDENT# '$where' : raised , #NEWLINE# #UNINDENT# } ) ) #NEWLINE# #UNINDENT# except KeyboardInterrupt : #NEWLINE# #INDENT# raised = True #NEWLINE# #UNINDENT# self . assertTrue ( raised , "Didn't raise expected KeyboardInterrupt" ) #NEWLINE# self . assertEqual ( { #NEWLINE# #INDENT# '_id' : 1 , #NEWLINE# #UNINDENT# } , next ( db . foo . find ( ) ) )