#NEWLINE# @ patch ( 'carbon.state.events.metricGenerated' ) #NEWLINE# def test_compute_value_can_flush_interval_multiple_times( self , metric_generated_mock ) : #NEWLINE# #INDENT# interval_buffer = IntervalBuffer ( 600 ) #NEWLINE# interval_buffer . input ( ( 600 , 1.0 ) ) #NEWLINE# interval_buffer . input ( ( 601 , 2.0 ) ) #NEWLINE# interval_buffer . input ( ( 602 , 3.0 ) ) #NEWLINE# self . metric_buffer . interval_buffers [ 600 ] = interval_buffer #NEWLINE# with patch ( 'time.time' ) as time_mock : #NEWLINE# #INDENT# time_mock . return_value = 600 #NEWLINE# self . metric_buffer . compute_value ( ) #NEWLINE# calls = [ call ( 'carbon.foo.bar' , ( 600 , 6.0 ) ) ] #NEWLINE# time_mock . return_value = 630 #NEWLINE# interval_buffer . input ( ( 604 , 4.0 ) ) #NEWLINE# self . metric_buffer . compute_value ( ) #NEWLINE# calls . append ( call ( 'carbon.foo.bar' , ( 600 , 10.0 ) ) ) #NEWLINE# metric_generated_mock . assert_has_calls ( calls )