<s> '''<STR_LIT>''' <EOL> import unittest <EOL> from unittest import TestCase <EOL> from mock import MagicMock <EOL> import time <EOL> import redis <EOL> import argparse <EOL> from scutils . redis_queue import RedisQueue , RedisPriorityQueue , RedisStack <EOL> from scutils . stats_collector import ( ThreadedCounter , TimeWindow , <EOL> RollingTimeWindow , Counter , UniqueCounter , <EOL> HyperLogLogCounter , BitMapCounter ) <EOL> class RedisMixin ( object ) : <EOL> def __init__ ( self , name , redis_conn ) : <EOL> self . redis_conn = redis_conn <EOL> TestCase . __init__ ( self , name ) <EOL> class QueueMixin ( RedisMixin , object ) : <EOL> queue_class = None <EOL> def setUp ( self ) : <EOL> self . queue = self . queue_class ( self . redis_conn , '<STR_LIT:key>' ) <EOL> class TestRedisFifoQueue ( QueueMixin , TestCase ) : <EOL> queue_class = RedisQueue <EOL> def test_fifo_queue ( self ) : <EOL> item1 = { "<STR_LIT>" : '<STR_LIT:a>' , "<STR_LIT>" : <NUM_LIT:2> } <EOL> item2 = <NUM_LIT> <EOL> item3 = "<STR_LIT>" <EOL> self . queue . push ( item1 ) <EOL> self . queue . push ( item2 ) <EOL> self . queue . push ( item3 ) <EOL> self . assertEqual ( item1 , self . queue . pop ( ) ) <EOL> self . assertEqual ( item2 , self . queue . pop ( ) ) <EOL> self . assertEqual ( item3 , self . queue . pop ( ) ) <EOL> class TestRedisPriorityQueue ( QueueMixin , TestCase ) : <EOL> queue_class = RedisPriorityQueue <EOL> def test_priority_queue ( self ) : <EOL> item1 = [ '<STR_LIT:a>' , '<STR_LIT:b>' , '<STR_LIT:c>' ] <EOL> item2 = <NUM_LIT> <EOL> item3 = "<STR_LIT>" <EOL> self . queue . push ( item1 , <NUM_LIT:3> ) <EOL> self . queue . push ( item2 , <NUM_LIT:1> ) <EOL> self . queue . push ( item3 , <NUM_LIT:2> ) <EOL> self . assertEqual ( item1 , self . queue . pop ( ) ) <EOL> self . assertEqual ( item3 , self . queue . pop ( ) ) <EOL> self . assertEqual ( item2 , self . queue . pop ( ) ) <EOL> class TestRedisStack ( QueueMixin , TestCase ) : <EOL> queue_class = RedisStack <EOL> def test_stack ( self ) : <EOL> item1 = [ '<STR_LIT:a>' , '<STR_LIT:b>' , '<STR_LIT:c>' ] <EOL> item2 = <NUM_LIT> <EOL> item3 = "<STR_LIT>" <EOL> self . queue . push ( item1 ) <EOL> self . queue . push ( item2 ) <EOL> self . queue . push ( item3 ) <EOL> self . assertEqual ( item3 , self . queue . pop ( ) ) <EOL> self . assertEqual ( item2 , self . queue . pop ( ) ) <EOL> self . assertEqual ( item1 , self . queue . pop ( ) ) <EOL> class TestStatsThreaded ( RedisMixin , TestCase ) : <EOL> def test_set_key ( self ) : <EOL> tc = ThreadedCounter ( start_time = <NUM_LIT> ) <EOL> self . assertEqual ( '<STR_LIT>' , tc . get_key ( ) ) <EOL> tc . roll = True <EOL> tc . key = '<STR_LIT>' <EOL> tc . _set_key ( ) <EOL> self . assertEqual ( '<STR_LIT>' , tc . get_key ( ) ) <EOL> def test_is_expired ( self ) : <EOL> tc = ThreadedCounter ( start_time = <NUM_LIT> , window = <NUM_LIT:5> ) <EOL> tc . _time = MagicMock ( return_value = <NUM_LIT> ) <EOL> self . assertFalse ( tc . is_expired ( ) ) <EOL> tc . _time = MagicMock ( return_value = <NUM_LIT> ) <EOL> self . assertTrue ( tc . is_expired ( ) ) <EOL> def test_threading ( self ) : <EOL> tc = ThreadedCounter ( start_time = <NUM_LIT> , cycle_time = <NUM_LIT:1> ) <EOL> tc . expire = MagicMock ( ) <EOL> tc . setup ( redis_conn = self . redis_conn ) <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> tc . stop ( ) <EOL> def test_purge_old ( self ) : <EOL> tc = ThreadedCounter ( keep_max = <NUM_LIT:1> ) <EOL> tc . keep_max = <NUM_LIT:1> <EOL> tc . redis_conn = self . redis_conn <EOL> self . redis_conn . set ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . redis_conn . set ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> tc . purge_old ( ) <EOL> self . assertEqual ( [ '<STR_LIT>' ] , <EOL> self . redis_conn . keys ( tc . get_key ( ) + '<STR_LIT>' ) ) <EOL> self . redis_conn . delete ( '<STR_LIT>' ) <EOL> def tearDown ( self ) : <EOL> keys = self . redis_conn . keys ( '<STR_LIT>' ) <EOL> while len ( keys ) > <NUM_LIT:0> : <EOL> key = keys . pop ( ) <EOL> self . redis_conn . delete ( key ) <EOL> class CleanMixin ( object ) : <EOL> def clean_keys ( self , key ) : <EOL> keys = self . redis_conn . keys ( key + '<STR_LIT:*>' ) <EOL> while len ( keys ) > <NUM_LIT:0> : <EOL> key = keys . pop ( ) <EOL> self . redis_conn . delete ( key ) <EOL> class TestStatsTimeWindow ( RedisMixin , TestCase , CleanMixin ) : <EOL> def test_window ( self ) : <EOL> counter = TimeWindow ( key = '<STR_LIT>' , cycle_time = <NUM_LIT> , window = <NUM_LIT:1> ) <EOL> counter . setup ( self . redis_conn ) <EOL> counter . increment ( ) <EOL> counter . increment ( ) <EOL> counter . increment ( ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> counter . increment ( ) <EOL> counter . stop ( ) <EOL> self . assertEqual ( counter . value ( ) , <NUM_LIT:3> ) <EOL> self . clean_keys ( counter . key ) <EOL> def test_roll_window ( self ) : <EOL> counter = TimeWindow ( key = '<STR_LIT>' , cycle_time = <NUM_LIT> , window = <NUM_LIT:1> , <EOL> roll = True ) <EOL> counter . setup ( self . redis_conn ) <EOL> counter . increment ( ) <EOL> counter . increment ( ) <EOL> counter . increment ( ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> counter . increment ( ) <EOL> counter . increment ( ) <EOL> counter . stop ( ) <EOL> keys = self . redis_conn . keys ( '<STR_LIT>' ) <EOL> keys . sort ( ) <EOL> self . assertEqual ( len ( keys ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( self . redis_conn . zcard ( keys [ <NUM_LIT:0> ] ) , <NUM_LIT:3> ) <EOL> self . assertEqual ( counter . value ( ) , <NUM_LIT:2> ) <EOL> counter . stop ( ) <EOL> self . clean_keys ( counter . key ) <EOL> class TestStatsRollingTimeWindow ( RedisMixin , TestCase , CleanMixin ) : <EOL> def test_rolling_window ( self ) : <EOL> time . sleep ( <NUM_LIT:5> - ( int ( time . time ( ) ) % <NUM_LIT:5> ) ) <EOL> counter = RollingTimeWindow ( key = '<STR_LIT>' , cycle_time = <NUM_LIT> , window = <NUM_LIT:5> ) <EOL> counter . setup ( self . redis_conn ) <EOL> counter . increment ( ) <EOL> counter . increment ( ) <EOL> time . sleep ( <NUM_LIT:3> ) <EOL> counter . increment ( ) <EOL> counter . increment ( ) <EOL> time . sleep ( <NUM_LIT:3> ) <EOL> counter . increment ( ) <EOL> self . assertEqual ( counter . value ( ) , <NUM_LIT:3> ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . assertEqual ( counter . value ( ) , <NUM_LIT:0> ) <EOL> counter . increment ( ) <EOL> self . assertEqual ( counter . value ( ) , <NUM_LIT:1> ) <EOL> counter . stop ( ) <EOL> self . clean_keys ( counter . key ) <EOL> class TestStatsCounter ( RedisMixin , TestCase , CleanMixin ) : <EOL> def test_generic_counter ( self ) : <EOL> counter = Counter ( key = '<STR_LIT>' ) <EOL> counter . setup ( redis_conn = self . redis_conn ) <EOL> i = <NUM_LIT:0> <EOL> while i < <NUM_LIT> : <EOL> counter . increment ( ) <EOL> i += <NUM_LIT:1> <EOL> self . assertEqual ( <NUM_LIT> , counter . value ( ) ) <EOL> counter . stop ( ) <EOL> self . clean_keys ( counter . key ) <EOL> def test_roll_generic_counter ( self ) : <EOL> time . sleep ( <NUM_LIT:2> - ( int ( time . time ( ) ) % <NUM_LIT:2> ) ) <EOL> counter = Counter ( key = '<STR_LIT>' , window = <NUM_LIT:2> , cycle_time = <NUM_LIT> , roll = True ) <EOL> counter . setup ( redis_conn = self . redis_conn ) <EOL> i = <NUM_LIT:0> <EOL> while i < <NUM_LIT:100> : <EOL> counter . increment ( ) <EOL> i += <NUM_LIT:1> <EOL> self . assertEqual ( counter . value ( ) , <NUM_LIT:100> ) <EOL> time . sleep ( <NUM_LIT:3> ) <EOL> i = <NUM_LIT:0> <EOL> while i < <NUM_LIT:50> : <EOL> counter . increment ( ) <EOL> i += <NUM_LIT:1> <EOL> self . assertEqual ( counter . value ( ) , <NUM_LIT:50> ) <EOL> counter . stop ( ) <EOL> self . clean_keys ( counter . key ) <EOL> class TestStatsUniqueCounter ( RedisMixin , TestCase , CleanMixin ) : <EOL> def test_uniques ( self ) : <EOL> counter = UniqueCounter ( key = '<STR_LIT>' ) <EOL> counter . setup ( redis_conn = self . redis_conn ) <EOL> i = <NUM_LIT:0> <EOL> while i < <NUM_LIT:1000> : <EOL> counter . increment ( i % <NUM_LIT:100> ) <EOL> i += <NUM_LIT:1> <EOL> self . assertEqual ( counter . value ( ) , <NUM_LIT:100> ) <EOL> counter . stop ( ) <EOL> self . clean_keys ( counter . key ) <EOL> def test_roll_uniques ( self ) : <EOL> time . sleep ( <NUM_LIT:2> - ( int ( time . time ( ) ) % <NUM_LIT:2> ) ) <EOL> counter = UniqueCounter ( key = '<STR_LIT>' , window = <NUM_LIT:2> , cycle_time = <NUM_LIT> , <EOL> roll = True ) <EOL> counter . setup ( redis_conn = self . redis_conn ) <EOL> i = <NUM_LIT:0> <EOL> while i < <NUM_LIT:100> : <EOL> counter . increment ( i % <NUM_LIT:10> ) <EOL> i += <NUM_LIT:1> <EOL> self . assertEqual ( counter . value ( ) , <NUM_LIT:10> ) <EOL> time . sleep ( <NUM_LIT:3> ) <EOL> i = <NUM_LIT:0> <EOL> while i < <NUM_LIT:50> : <EOL> counter . increment ( i % <NUM_LIT:10> ) <EOL> i += <NUM_LIT:1> <EOL> self . assertEqual ( counter . value ( ) , <NUM_LIT:10> ) <EOL> counter . stop ( ) <EOL> self . clean_keys ( counter . key ) <EOL> class TestStatsHyperLogLogCounter ( RedisMixin , TestCase , CleanMixin ) : <EOL> tolerance = <NUM_LIT:2> <EOL> def get_percent_diff ( self , value , actual ) : <EOL> return abs ( actual - value ) / ( ( value + actual ) / <NUM_LIT> ) * <NUM_LIT> <EOL> def test_hll_counter ( self ) : <EOL> counter = HyperLogLogCounter ( key = '<STR_LIT>' ) <EOL> counter . setup ( redis_conn = self . redis_conn ) <EOL> for n in range ( <NUM_LIT:3> ) : <EOL> i = <NUM_LIT:0> <EOL> while i < <NUM_LIT> * ( n + <NUM_LIT:1> ) : <EOL> counter . increment ( i % ( <NUM_LIT> * ( n + <NUM_LIT:1> ) ) ) <EOL> i += <NUM_LIT:1> <EOL> value = <NUM_LIT> * ( n + <NUM_LIT:1> ) <EOL> actual = counter . value ( ) <EOL> diff = self . get_percent_diff ( value , actual ) <EOL> self . assertLessEqual ( diff , self . tolerance ) <EOL> self . clean_keys ( counter . key ) <EOL> counter . stop ( ) <EOL> def test_roll_hll_counter ( self ) : <EOL> time . sleep ( <NUM_LIT> - ( time . time ( ) % <NUM_LIT> ) ) <EOL> counter = HyperLogLogCounter ( key = '<STR_LIT>' , window = <NUM_LIT:5> , roll = True , <EOL> cycle_time = <NUM_LIT> , ) <EOL> counter . setup ( redis_conn = self . redis_conn ) <EOL> i = <NUM_LIT:0> <EOL> while i < <NUM_LIT> : <EOL> counter . increment ( i % <NUM_LIT> ) <EOL> i += <NUM_LIT:1> <EOL> value = <NUM_LIT> <EOL> actual = counter . value ( ) <EOL> diff = self . get_percent_diff ( value , actual ) <EOL> self . assertLessEqual ( diff , self . tolerance ) <EOL> self . clean_keys ( counter . key ) <EOL> time . sleep ( <NUM_LIT> - ( time . time ( ) % <NUM_LIT> ) ) <EOL> i = <NUM_LIT:0> <EOL> while i < <NUM_LIT> : <EOL> counter . increment ( i % <NUM_LIT> ) <EOL> i += <NUM_LIT:1> <EOL> value = <NUM_LIT> <EOL> actual = counter . value ( ) <EOL> diff = self . get_percent_diff ( value , actual ) <EOL> self . assertLessEqual ( diff , self . tolerance ) <EOL> counter . stop ( ) <EOL> self . clean_keys ( counter . key ) <EOL> class TestStatsBitMapCounter ( RedisMixin , TestCase , CleanMixin ) : <EOL> def test_bitmap_counter ( self ) : <EOL> counter = BitMapCounter ( key = '<STR_LIT>' ) <EOL> counter . setup ( redis_conn = self . redis_conn ) <EOL> counter . increment ( <NUM_LIT:0> ) <EOL> counter . increment ( <NUM_LIT:1> ) <EOL> counter . increment ( <NUM_LIT:5> ) <EOL> counter . increment ( <NUM_LIT:0> ) <EOL> self . assertEqual ( <NUM_LIT:3> , counter . value ( ) ) <EOL> counter . stop ( ) <EOL> self . clean_keys ( counter . key ) <EOL> def test_roll_bitmap_counter ( self ) : <EOL> time . sleep ( <NUM_LIT:2> - ( int ( time . time ( ) ) % <NUM_LIT:2> ) ) <EOL> counter = BitMapCounter ( key = '<STR_LIT>' , window = <NUM_LIT:2> , cycle_time = <NUM_LIT> , <EOL> roll = True ) <EOL> counter . setup ( redis_conn = self . redis_conn ) <EOL> counter . increment ( <NUM_LIT:0> ) <EOL> counter . increment ( <NUM_LIT:1> ) <EOL> counter . increment ( <NUM_LIT:0> ) <EOL> self . assertEqual ( <NUM_LIT:2> , counter . value ( ) ) <EOL> time . sleep ( <NUM_LIT:3> ) <EOL> counter . increment ( <NUM_LIT:0> ) <EOL> self . assertEqual ( <NUM_LIT:1> , counter . value ( ) ) <EOL> counter . stop ( ) <EOL> self . clean_keys ( counter . key ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> parser = argparse . ArgumentParser ( description = "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> parser . add_argument ( '<STR_LIT>' , '<STR_LIT>' , action = '<STR_LIT:store>' , <EOL> default = '<STR_LIT:localhost>' , help = "<STR_LIT>" ) <EOL> parser . add_argument ( '<STR_LIT>' , '<STR_LIT>' , action = '<STR_LIT:store>' , default = '<STR_LIT>' , <EOL> help = "<STR_LIT>" ) <EOL> args = vars ( parser . parse_args ( ) ) <EOL> redis_conn = redis . Redis ( host = args [ '<STR_LIT>' ] , port = args [ '<STR_LIT>' ] ) <EOL> suite = unittest . TestSuite ( ) <EOL> suite . addTest ( TestStatsHyperLogLogCounter ( '<STR_LIT>' , redis_conn ) ) <EOL> suite . addTest ( TestStatsHyperLogLogCounter ( '<STR_LIT>' , <EOL> redis_conn ) ) <EOL> suite . addTest ( TestRedisFifoQueue ( '<STR_LIT>' , redis_conn ) ) <EOL> suite . addTest ( TestRedisPriorityQueue ( '<STR_LIT>' , redis_conn ) ) <EOL> suite . addTest ( TestRedisStack ( '<STR_LIT>' , redis_conn ) ) <EOL> suite . addTest ( TestStatsThreaded ( '<STR_LIT>' , redis_conn ) ) <EOL> suite . addTest ( TestStatsThreaded ( '<STR_LIT>' , redis_conn ) ) <EOL> suite . addTest ( TestStatsThreaded ( '<STR_LIT>' , redis_conn ) ) <EOL> suite . addTest ( TestStatsThreaded ( '<STR_LIT>' , redis_conn ) ) <EOL> suite . addTest ( TestStatsTimeWindow ( '<STR_LIT>' , redis_conn ) ) <EOL> suite . addTest ( TestStatsTimeWindow ( '<STR_LIT>' , redis_conn ) ) <EOL> suite . addTest ( TestStatsRollingTimeWindow ( '<STR_LIT>' , <EOL> redis_conn ) ) <EOL> suite . addTest ( TestStatsCounter ( '<STR_LIT>' , redis_conn ) ) <EOL> suite . addTest ( TestStatsCounter ( '<STR_LIT>' , redis_conn ) ) <EOL> suite . addTest ( TestStatsUniqueCounter ( '<STR_LIT>' , redis_conn ) ) <EOL> suite . addTest ( TestStatsUniqueCounter ( '<STR_LIT>' , redis_conn ) ) <EOL> suite . addTest ( TestStatsBitMapCounter ( '<STR_LIT>' , redis_conn ) ) <EOL> suite . addTest ( TestStatsBitMapCounter ( '<STR_LIT>' , <EOL> redis_conn ) ) <EOL> unittest . TextTestRunner ( verbosity = <NUM_LIT:2> ) . run ( suite ) </s>
