<s> import time <EOL> import uuid <EOL> import unittest <EOL> from . import conf <EOL> from . import timelines <EOL> from . import keys <EOL> client = conf . register_connection ( ) <EOL> class TimelineTestCase ( unittest . TestCase ) : <EOL> timeline_class = timelines . Timeline <EOL> def setup_timeline ( self ) : <EOL> self . timeline = self . timeline_class ( '<STR_LIT:test>' ) <EOL> self . start_time = int ( time . time ( ) ) <EOL> self . items = [ ] <EOL> for i , t in enumerate ( range ( self . start_time , self . start_time + <NUM_LIT:10> ) ) : <EOL> item = { '<STR_LIT:index>' : i } <EOL> self . items . append ( ( item , t ) ) <EOL> self . timeline . add ( item , timestamp = t ) <EOL> def teardown_timeline ( self ) : <EOL> self . timeline . delete ( ) <EOL> def assert_ranges_equal ( self , items1 , items2 ) : <EOL> self . assertEqual ( len ( items1 ) , len ( items2 ) ) <EOL> for ( item1 , t1 ) , ( item2 , t2 ) in zip ( items1 , items2 ) : <EOL> self . assertEqual ( t1 , t2 ) <EOL> self . assertEqual ( item1 , item2 ) <EOL> def setUp ( self ) : <EOL> self . setup_timeline ( ) <EOL> def tearDown ( self ) : <EOL> self . teardown_timeline ( ) <EOL> def test_count ( self ) : <EOL> self . assertEqual ( self . timeline . count ( ) , len ( self . items ) ) <EOL> self . assertEqual ( len ( self . timeline ) , len ( self . items ) ) <EOL> def test_tail ( self ) : <EOL> last1 , t1 = self . items [ - <NUM_LIT:1> ] <EOL> last2 , t2 = self . timeline . tail ( ) [ <NUM_LIT:0> ] <EOL> self . assertEqual ( t1 , t2 ) <EOL> self . assertEqual ( last1 , last2 ) <EOL> tail1 = self . items [ - <NUM_LIT:2> : ] <EOL> tail2 = self . timeline . tail ( <NUM_LIT:2> ) <EOL> self . assert_ranges_equal ( tail1 , tail2 ) <EOL> def test_head ( self ) : <EOL> first1 , t1 = self . items [ <NUM_LIT:0> ] <EOL> first2 , t2 = self . timeline . head ( ) [ <NUM_LIT:0> ] <EOL> self . assertEqual ( t1 , t2 ) <EOL> self . assertEqual ( first1 , first2 ) <EOL> head1 = self . items [ : <NUM_LIT:2> ] <EOL> head2 = self . timeline . head ( <NUM_LIT:2> ) <EOL> self . assert_ranges_equal ( head1 , head2 ) <EOL> def test_range ( self ) : <EOL> self . assert_ranges_equal ( self . items [ <NUM_LIT:2> : <NUM_LIT:4> ] , self . timeline . range ( <NUM_LIT:2> , <NUM_LIT:3> ) ) <EOL> self . assert_ranges_equal ( self . items [ <NUM_LIT:2> : ] , self . timeline . range ( <NUM_LIT:2> ) ) <EOL> self . assert_ranges_equal ( self . items [ <NUM_LIT:2> : - <NUM_LIT:2> ] , self . timeline . range ( <NUM_LIT:2> , - <NUM_LIT:3> ) ) <EOL> self . assert_ranges_equal ( self . items [ - <NUM_LIT:4> : - <NUM_LIT:1> ] , self . timeline . range ( - <NUM_LIT:4> , - <NUM_LIT:2> ) ) <EOL> def test_delete_range ( self ) : <EOL> items1 = self . items [ : <NUM_LIT:2> ] + self . items [ <NUM_LIT:4> : ] <EOL> self . timeline . delete_range ( <NUM_LIT:2> , <NUM_LIT:3> ) <EOL> items2 = self . timeline . items ( ) <EOL> self . assertEqual ( len ( items2 ) , len ( self . items ) - <NUM_LIT:2> ) <EOL> self . assert_ranges_equal ( items1 , self . timeline . items ( ) ) <EOL> def test_timerange ( self ) : <EOL> items1 = self . items [ <NUM_LIT:2> : <NUM_LIT:5> ] <EOL> start , end = self . items [ <NUM_LIT:2> ] , self . items [ <NUM_LIT:4> ] <EOL> start_ts , end_ts = start [ <NUM_LIT:1> ] , end [ <NUM_LIT:1> ] <EOL> items2 = self . timeline . timerange ( start_ts , end_ts ) <EOL> self . assert_ranges_equal ( items1 , items2 ) <EOL> def test_count_timerange ( self ) : <EOL> items1 = self . items [ <NUM_LIT:2> : <NUM_LIT:5> ] <EOL> start , end = self . items [ <NUM_LIT:2> ] , self . items [ <NUM_LIT:4> ] <EOL> start_ts , end_ts = start [ <NUM_LIT:1> ] , end [ <NUM_LIT:1> ] <EOL> self . assertEqual ( len ( items1 ) , self . timeline . count_timerange ( start_ts , end_ts ) ) <EOL> def test_delete_timerange ( self ) : <EOL> items1 = self . items [ : <NUM_LIT:2> ] + self . items [ <NUM_LIT:4> : ] <EOL> start , end = self . items [ <NUM_LIT:2> ] , self . items [ <NUM_LIT:3> ] <EOL> start_ts , end_ts = start [ <NUM_LIT:1> ] , end [ <NUM_LIT:1> ] <EOL> self . timeline . delete_timerange ( start_ts , end_ts ) <EOL> items2 = self . timeline . items ( ) <EOL> self . assertEqual ( len ( items2 ) , len ( self . items ) - <NUM_LIT:2> ) <EOL> self . assert_ranges_equal ( items1 , self . timeline . items ( ) ) <EOL> class HourTimelineTestCase ( TimelineTestCase ) : <EOL> timeline_class = timelines . HourTimeline <EOL> class DayTimelineTestCase ( TimelineTestCase ) : <EOL> timeline_class = timelines . DayTimeline <EOL> class WeekTimelineTestCase ( TimelineTestCase ) : <EOL> timeline_class = timelines . WeekTimeline <EOL> class MonthTimelineTestCase ( TimelineTestCase ) : <EOL> timeline_class = timelines . MonthTimeline <EOL> class YearTimelineTestCase ( TimelineTestCase ) : <EOL> timeline_class = timelines . YearTimeline <EOL> class TimeIndexedKeyTestCase ( unittest . TestCase ) : <EOL> key_class = keys . TimeIndexedKey <EOL> def setup_indexed_key ( self ) : <EOL> self . tik = self . key_class ( '<STR_LIT:test>' ) <EOL> self . start_time = int ( time . time ( ) ) <EOL> self . items = [ ] <EOL> for i , ts in enumerate ( range ( self . start_time , self . start_time + <NUM_LIT:10> ) ) : <EOL> key = str ( uuid . uuid4 ( ) . hex ) [ : <NUM_LIT:5> ] <EOL> value = { '<STR_LIT:index>' : i } <EOL> self . items . append ( ( key , value , ts ) ) <EOL> self . tik . set ( key , value , timestamp = ts , update_index = True ) <EOL> def teardown_indexed_key ( self ) : <EOL> self . tik . delete ( ) <EOL> def assert_ranges_equal ( self , items1 , items2 ) : <EOL> self . assertEqual ( len ( items1 ) , len ( items2 ) ) <EOL> for ( k1 , v1 , t1 ) , ( k2 , v2 , t2 ) in zip ( items1 , items2 ) : <EOL> self . assertEqual ( ( k1 , v1 , t1 ) , ( k2 , v2 , t2 ) ) <EOL> def setUp ( self ) : <EOL> self . setup_indexed_key ( ) <EOL> def tearDown ( self ) : <EOL> self . teardown_indexed_key ( ) <EOL> def test_get_set ( self ) : <EOL> now = int ( time . time ( ) ) <EOL> key = str ( uuid . uuid4 ( ) . hex ) [ : <NUM_LIT:5> ] <EOL> val = { '<STR_LIT:foo>' : <NUM_LIT:1> } <EOL> self . tik . set ( key , val , timestamp = now , update_index = True ) <EOL> val1 , t1 = self . tik . get ( key ) <EOL> val2 , t2 = self . tik [ key ] <EOL> self . assertEqual ( now , t1 , t2 ) <EOL> self . assertEqual ( val , val1 , val2 ) <EOL> val = { '<STR_LIT:bar>' : <NUM_LIT:2> } <EOL> past = now - <NUM_LIT:10> <EOL> self . tik . set ( key , val , timestamp = past , update_index = False ) <EOL> val1 , t1 = self . tik . get ( key ) <EOL> val2 , t2 = self . tik [ key ] <EOL> self . assertEqual ( now , t1 , t2 ) <EOL> self . assertEqual ( val , val1 , val2 ) <EOL> val = { '<STR_LIT>' : <NUM_LIT:3> } <EOL> self . tik . set ( key , val , timestamp = past , update_index = True ) <EOL> val1 , t1 = self . tik . get ( key ) <EOL> val2 , t2 = self . tik [ key ] <EOL> self . assertEqual ( past , t1 , t2 ) <EOL> self . assertEqual ( val , val1 , val2 ) <EOL> self . tik [ key ] = val = { '<STR_LIT:abc>' : <NUM_LIT:4> } <EOL> val1 , t1 = self . tik . get ( key ) <EOL> val2 , t2 = self . tik [ key ] <EOL> self . assertNotEqual ( t1 , past , now ) <EOL> self . assertEqual ( t1 , t2 ) <EOL> self . assertEqual ( val , val1 , val2 ) <EOL> def test_remove ( self ) : <EOL> now = int ( time . time ( ) ) <EOL> key = str ( uuid . uuid4 ( ) . hex ) [ : <NUM_LIT:5> ] <EOL> val = { '<STR_LIT:foo>' : <NUM_LIT:1> } <EOL> self . tik . set ( key , val , timestamp = now , update_index = True ) <EOL> self . assertNotEqual ( len ( self . tik ) , len ( self . items ) ) <EOL> self . tik . remove ( key ) <EOL> self . assertEqual ( len ( self . tik ) , len ( self . items ) ) <EOL> self . tik . set ( key , val , timestamp = now , update_index = True ) <EOL> self . assertNotEqual ( len ( self . tik ) , len ( self . items ) ) <EOL> del self . tik [ key ] <EOL> self . assertEqual ( len ( self . tik ) , len ( self . items ) ) <EOL> def test_count ( self ) : <EOL> self . assertEqual ( self . tik . count ( ) , len ( self . items ) ) <EOL> self . assertEqual ( len ( self . tik ) , len ( self . items ) ) <EOL> def test_keys ( self ) : <EOL> keys1 = self . tik . keys ( ) <EOL> keys2 = [ i [ <NUM_LIT:0> ] for i in self . items ] <EOL> self . assertEqual ( keys1 , keys2 ) <EOL> keys1 = self . tik . keys ( limit = <NUM_LIT:5> ) <EOL> keys2 = [ i [ <NUM_LIT:0> ] for i in self . items [ : <NUM_LIT:5> ] ] <EOL> self . assertEqual ( keys1 , keys2 ) <EOL> keys1 = self . tik . keys ( with_timestamp = True ) <EOL> keys2 = [ ( i [ <NUM_LIT:0> ] , i [ <NUM_LIT:2> ] ) for i in self . items ] <EOL> self . assertEqual ( keys1 , keys2 ) <EOL> start_ts , end_ts = self . items [ <NUM_LIT:1> ] [ <NUM_LIT:2> ] , self . items [ <NUM_LIT:4> ] [ <NUM_LIT:2> ] <EOL> keys1 = self . tik . keys ( start_ts , end_ts ) <EOL> keys2 = [ i [ <NUM_LIT:0> ] for i in self . items [ <NUM_LIT:1> : <NUM_LIT:5> ] ] <EOL> self . assertEqual ( keys1 , keys2 ) <EOL> keys1 = self . tik . keys ( start_ts , end_ts , limit = <NUM_LIT:2> , with_timestamp = True ) <EOL> keys2 = [ ( i [ <NUM_LIT:0> ] , i [ <NUM_LIT:2> ] ) for i in self . items [ <NUM_LIT:1> : <NUM_LIT:3> ] ] <EOL> self . assertEqual ( keys1 , keys2 ) <EOL> def test_values ( self ) : <EOL> keys = [ k for k , _ , _ in self . items ] <EOL> data1 = self . tik . values ( * keys ) <EOL> data2 = [ ( k , v ) for k , v , _ in self . items ] <EOL> self . assertEqual ( data1 , data2 ) <EOL> def test_timerange ( self ) : <EOL> items1 = self . tik . timerange ( ) <EOL> self . assert_ranges_equal ( items1 , self . items ) <EOL> items1 = self . tik . timerange ( limit = <NUM_LIT:5> ) <EOL> self . assert_ranges_equal ( items1 , self . items [ : <NUM_LIT:5> ] ) <EOL> start_ts , end_ts = self . items [ <NUM_LIT:1> ] [ <NUM_LIT:2> ] , self . items [ <NUM_LIT:4> ] [ <NUM_LIT:2> ] <EOL> items1 = self . tik . timerange ( start_ts , end_ts ) <EOL> self . assert_ranges_equal ( items1 , self . items [ <NUM_LIT:1> : <NUM_LIT:5> ] ) <EOL> items1 = self . tik . timerange ( start_ts , end_ts , limit = <NUM_LIT:2> ) <EOL> self . assert_ranges_equal ( items1 , self . items [ <NUM_LIT:1> : <NUM_LIT:3> ] ) <EOL> items1 = self . tik . timerange ( start_ts , limit = <NUM_LIT:2> ) <EOL> self . assert_ranges_equal ( items1 , self . items [ <NUM_LIT:1> : <NUM_LIT:3> ] ) <EOL> def test_count_timerange ( self ) : <EOL> self . assertEqual ( self . tik . count_timerange ( ) , len ( self . items ) ) <EOL> start_ts , end_ts = self . items [ <NUM_LIT:1> ] [ <NUM_LIT:2> ] , self . items [ <NUM_LIT:4> ] [ <NUM_LIT:2> ] <EOL> count = self . tik . count_timerange ( start_ts , end_ts ) <EOL> self . assertEqual ( count , <NUM_LIT:4> ) <EOL> count = self . tik . count_timerange ( start_ts ) <EOL> self . assertEqual ( count , len ( self . items ) - <NUM_LIT:1> ) <EOL> def test_delete_timerange ( self ) : <EOL> start_ts , end_ts = self . items [ <NUM_LIT:1> ] [ <NUM_LIT:2> ] , self . items [ <NUM_LIT:4> ] [ <NUM_LIT:2> ] <EOL> self . tik . delete_timerange ( start_ts , end_ts ) <EOL> self . assertEqual ( len ( self . tik ) , len ( self . items ) - <NUM_LIT:4> ) <EOL> class HourIndexedKeyTestCase ( TimeIndexedKeyTestCase ) : <EOL> key_class = keys . HourIndexedKey <EOL> class DayIndexedKeyTestCase ( TimeIndexedKeyTestCase ) : <EOL> key_class = keys . DayIndexedKey <EOL> class WeekIndexedKeyTestCase ( TimeIndexedKeyTestCase ) : <EOL> key_class = keys . WeekIndexedKey <EOL> class MonthIndexedKeyTestCase ( TimeIndexedKeyTestCase ) : <EOL> key_class = keys . MonthIndexedKey <EOL> class YaerIndexedKeyTestCase ( TimeIndexedKeyTestCase ) : <EOL> key_class = keys . YearIndexedKey <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
