<s> import sys <EOL> import time <EOL> import math <EOL> import unittest <EOL> from ostrich import stats , timing , histogram <EOL> class StatsTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> stats . clear_all ( ) <EOL> def test_counters ( self ) : <EOL> stats . incr ( "<STR_LIT>" , <NUM_LIT:1> ) <EOL> stats . incr ( "<STR_LIT>" , <NUM_LIT:12> ) <EOL> stats . incr ( "<STR_LIT>" ) <EOL> self . assertEquals ( { "<STR_LIT>" : <NUM_LIT:1> , "<STR_LIT>" : <NUM_LIT> } , stats . get_counter_stats ( ) ) <EOL> def test_timings_empty ( self ) : <EOL> stats . add_timing ( "<STR_LIT:test>" , <NUM_LIT:0> ) <EOL> test = stats . get_timing ( "<STR_LIT:test>" ) <EOL> self . assertEqual ( timing . TimingStat ( <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> ) , test . get ( reset = True ) ) <EOL> self . assertEqual ( timing . TimingStat ( <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) , test . get ( ) ) <EOL> def test_timings_basic ( self ) : <EOL> stats . add_timing ( "<STR_LIT:test>" , <NUM_LIT:1> ) <EOL> stats . add_timing ( "<STR_LIT:test>" , <NUM_LIT:2> ) <EOL> stats . add_timing ( "<STR_LIT:test>" , <NUM_LIT:3> ) <EOL> test = stats . get_timing ( "<STR_LIT:test>" ) <EOL> self . assertEqual ( timing . TimingStat ( <NUM_LIT:3> , <NUM_LIT:3> , <NUM_LIT:1> , <NUM_LIT> , <NUM_LIT> , histogram . Histogram ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ) ) , test . get ( ) ) <EOL> def test_timings_report ( self ) : <EOL> x = <NUM_LIT:0> <EOL> with stats . time ( "<STR_LIT>" ) : <EOL> for i in xrange ( <NUM_LIT:100> ) : <EOL> x += i <EOL> timings = stats . get_timing_stats ( ) <EOL> self . assertEquals ( [ "<STR_LIT>" ] , timings . keys ( ) ) <EOL> self . assertEquals ( <NUM_LIT:1> , timings [ "<STR_LIT>" ] . count ) <EOL> self . assertEquals ( timings [ "<STR_LIT>" ] . average , timings [ "<STR_LIT>" ] . min ) <EOL> self . assertEquals ( timings [ "<STR_LIT>" ] . average , timings [ "<STR_LIT>" ] . max ) <EOL> def test_timings_average ( self ) : <EOL> stats . add_timing ( "<STR_LIT:test>" , <NUM_LIT:0> ) <EOL> test = stats . get_timing ( "<STR_LIT:test>" ) <EOL> self . assertEquals ( timing . TimingStat ( <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> ) , test . get ( ) ) <EOL> def test_timings_negative ( self ) : <EOL> stats . add_timing ( "<STR_LIT:test>" , <NUM_LIT:1> ) <EOL> stats . add_timing ( "<STR_LIT:test>" , - <NUM_LIT:1> ) <EOL> test = stats . get_timing ( "<STR_LIT:test>" ) <EOL> self . assertEquals ( timing . TimingStat ( <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1.0> , <NUM_LIT:0.0> , histogram . Histogram ( <NUM_LIT:1> ) ) , test . get ( ) ) <EOL> def test_timing_boundarys ( self ) : <EOL> stats . add_timing ( "<STR_LIT:test>" , sys . maxint ) <EOL> stats . add_timing ( "<STR_LIT:test>" , <NUM_LIT:5> ) <EOL> sum = <NUM_LIT> + sys . maxint <EOL> avg = sum / <NUM_LIT> <EOL> sumsq = <NUM_LIT> * <NUM_LIT> + float ( sys . maxint ) * sys . maxint <EOL> partial = sumsq - sum * avg <EOL> test = stats . get_timing ( "<STR_LIT:test>" ) <EOL> self . assertEquals ( timing . TimingStat ( <NUM_LIT:2> , sys . maxint , <NUM_LIT:5> , avg , partial , histogram . Histogram ( <NUM_LIT:5> , sys . maxint ) ) , test . get ( ) ) <EOL> def test_timing_with ( self ) : <EOL> with stats . time ( "<STR_LIT:test>" ) : <EOL> time . sleep ( <NUM_LIT> ) <EOL> test = stats . get_timing ( "<STR_LIT:test>" ) <EOL> self . assertTrue ( test . get ( ) . average >= <NUM_LIT:10> ) <EOL> def test_timing_decorator ( self ) : <EOL> @ stats . time ( "<STR_LIT:test>" ) <EOL> def _test_timing_decorator ( ) : <EOL> time . sleep ( <NUM_LIT> ) <EOL> _test_timing_decorator ( ) <EOL> test = stats . get_timing ( "<STR_LIT:test>" ) <EOL> self . assertTrue ( test . get ( ) . average >= <NUM_LIT:10> ) <EOL> def test_timing_reset ( self ) : <EOL> x = <NUM_LIT:0> <EOL> with stats . time ( "<STR_LIT>" ) : <EOL> for i in xrange ( <NUM_LIT:100> ) : x += i <EOL> self . assertEqual ( <NUM_LIT:1> , stats . get_timing_stats ( reset = False ) [ "<STR_LIT>" ] . count ) <EOL> with stats . time ( "<STR_LIT>" ) : <EOL> for i in xrange ( <NUM_LIT:100> ) : x += i <EOL> self . assertEqual ( <NUM_LIT:2> , stats . get_timing_stats ( reset = False ) [ "<STR_LIT>" ] . count ) <EOL> self . assertEqual ( <NUM_LIT:2> , stats . get_timing_stats ( reset = True ) [ "<STR_LIT>" ] . count ) <EOL> with stats . time ( "<STR_LIT>" ) : <EOL> for i in xrange ( <NUM_LIT:100> ) : x += i <EOL> self . assertEqual ( <NUM_LIT:1> , stats . get_timing_stats ( reset = False ) [ "<STR_LIT>" ] . count ) <EOL> def test_timing_bundle ( self ) : <EOL> timing_stat = timing . TimingStat ( <NUM_LIT:3> , <NUM_LIT:20> , <NUM_LIT:10> , <NUM_LIT> , <NUM_LIT> , histogram . Histogram ( <NUM_LIT:10> , <NUM_LIT:15> , <NUM_LIT:20> ) ) <EOL> stats . add_timing ( "<STR_LIT:test>" , timing_stat ) <EOL> stats . add_timing ( "<STR_LIT:test>" , <NUM_LIT> ) <EOL> test = stats . get_timing_stats ( reset = False ) [ "<STR_LIT:test>" ] <EOL> self . assertEqual ( <NUM_LIT:4> , test . count ) <EOL> self . assertEqual ( <NUM_LIT> , test . average ) <EOL> self . assertEqual ( <NUM_LIT:6> , int ( test . std_dev ) ) <EOL> stats . clear_all ( ) <EOL> timing_stat1 = timing . TimingStat ( <NUM_LIT:2> , <NUM_LIT> , <NUM_LIT:15> , <NUM_LIT> , <NUM_LIT> , histogram . Histogram ( <NUM_LIT:15> , <NUM_LIT> ) ) <EOL> timing_stat2 = timing . TimingStat ( <NUM_LIT:2> , <NUM_LIT:20> , <NUM_LIT:10> , <NUM_LIT> , <NUM_LIT> , histogram . Histogram ( <NUM_LIT:10> , <NUM_LIT:20> ) ) <EOL> stats . add_timing ( "<STR_LIT:test>" , timing_stat1 ) <EOL> stats . add_timing ( "<STR_LIT:test>" , timing_stat2 ) <EOL> test = stats . get_timing_stats ( reset = False ) [ "<STR_LIT:test>" ] <EOL> self . assertEqual ( <NUM_LIT:4> , test . count ) <EOL> self . assertEqual ( <NUM_LIT> , test . average ) <EOL> self . assertEqual ( <NUM_LIT:6> , int ( test . std_dev ) ) <EOL> def test_timing_add ( self ) : <EOL> x = <NUM_LIT:0> <EOL> with stats . time ( "<STR_LIT>" ) : <EOL> for i in xrange ( <NUM_LIT:100> ) : x += <NUM_LIT:1> <EOL> self . assertEquals ( <NUM_LIT:1> , len ( stats . get_timing_stats ( reset = False ) ) ) <EOL> stats . add_timing ( "<STR_LIT>" , timing . TimingStat ( <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> ) ) <EOL> self . assertEquals ( <NUM_LIT:2> , len ( stats . get_timing_stats ( reset = False ) ) ) <EOL> self . assertEquals ( <NUM_LIT:1> , stats . get_timing_stats ( reset = True ) [ "<STR_LIT>" ] . count ) <EOL> stats . add_timing ( "<STR_LIT>" , timing . TimingStat ( <NUM_LIT:3> , <NUM_LIT:0> , <NUM_LIT:0> ) ) <EOL> self . assertEquals ( <NUM_LIT:3> , stats . get_timing_stats ( reset = False ) [ "<STR_LIT>" ] . count ) <EOL> def test_timing_report_sorted ( self ) : <EOL> stats . add_timing ( "<STR_LIT>" , timing . TimingStat ( <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> ) ) <EOL> string = str ( stats . get_timing_stats ( reset = False ) [ "<STR_LIT>" ] ) <EOL> self . assertEquals ( "<STR_LIT>" , string ) <EOL> def test_gauge_report ( self ) : <EOL> @ stats . gauge ( "<STR_LIT>" ) <EOL> def _pi ( ) : <EOL> return math . pi <EOL> stats . make_gauge ( "<STR_LIT:e>" , lambda : math . e ) <EOL> self . assertEquals ( { "<STR_LIT:e>" : math . e , "<STR_LIT>" : math . pi } , stats . get_gauge_stats ( ) ) <EOL> def test_gauge_update ( self ) : <EOL> potatoes = [ <NUM_LIT> ] <EOL> @ stats . gauge ( "<STR_LIT>" ) <EOL> def _stew ( ) : <EOL> potatoes [ <NUM_LIT:0> ] += <NUM_LIT:1.0> <EOL> return potatoes [ <NUM_LIT:0> ] <EOL> self . assertEquals ( { "<STR_LIT>" : <NUM_LIT> } , stats . get_gauge_stats ( ) ) <EOL> self . assertEquals ( { "<STR_LIT>" : <NUM_LIT> } , stats . get_gauge_stats ( ) ) <EOL> self . assertEquals ( { "<STR_LIT>" : <NUM_LIT> } , stats . get_gauge_stats ( ) ) <EOL> def test_fork ( self ) : <EOL> collection = stats . fork ( ) <EOL> stats . incr ( "<STR_LIT>" , <NUM_LIT:5> ) <EOL> self . assertEquals ( { "<STR_LIT>" : <NUM_LIT:5> } , collection . get_counter_stats ( ) ) <EOL> self . assertEquals ( { "<STR_LIT>" : <NUM_LIT:5> } , stats . get_counter_stats ( reset = True ) ) <EOL> stats . incr ( "<STR_LIT>" , <NUM_LIT:5> ) <EOL> self . assertEquals ( { "<STR_LIT>" : <NUM_LIT:10> } , collection . get_counter_stats ( ) ) <EOL> self . assertEquals ( { "<STR_LIT>" : <NUM_LIT:5> } , stats . get_counter_stats ( reset = True ) ) </s>
