#NEWLINE# def test_take_while_complete_after( self ) : #NEWLINE# #INDENT# scheduler = TestScheduler ( ) #NEWLINE# xs = scheduler . create_hot_observable ( on_next ( 90 , ( - 1 ) ) , on_next ( 110 , ( - 1 ) ) , on_next ( 210 , 2 ) , on_next ( 260 , 5 ) , on_next ( 290 , 13 ) , on_next ( 320 , 3 ) , on_next ( 350 , 7 ) , on_next ( 390 , 4 ) , on_next ( 410 , 17 ) , on_next ( 450 , 8 ) , on_next ( 500 , 23 ) , on_completed ( 600 ) ) #NEWLINE# invoked = [ 0 ] #NEWLINE# def factory( ) : #NEWLINE# #INDENT# def predicate( x ) : #NEWLINE# #INDENT# invoked [ 0 ] += 1 #NEWLINE# return is_prime ( x ) #NEWLINE# #UNINDENT# return xs . take_while ( predicate ) #NEWLINE# #UNINDENT# results = scheduler . start ( xs ) #NEWLINE# results . messages . assert_equal ( on_next ( 210 , 2 ) , on_next ( 260 , 5 ) , on_next ( 290 , 13 ) , on_next ( 320 , 3 ) , on_next ( 350 , 7 ) , on_completed ( 390 ) ) #NEWLINE# xs . subscriptions . assert_equal ( subscribe ( 200 , 390 ) ) #NEWLINE# assert ( invoked [ 0 ] == 6 )