<s> import unittest <EOL> from time import sleep <EOL> from hazelcast import SerializationConfig <EOL> from hazelcast . config import NearCacheConfig <EOL> from hazelcast . near_cache import * <EOL> from hazelcast . serialization import SerializationServiceV1 <EOL> class NearCacheTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> logging . basicConfig ( format = '<STR_LIT>' , datefmt = "<STR_LIT>" ) <EOL> logging . getLogger ( ) . setLevel ( logging . DEBUG ) <EOL> self . service = SerializationServiceV1 ( serialization_config = SerializationConfig ( ) ) <EOL> def tearDown ( self ) : <EOL> self . service . destroy ( ) <EOL> def test_near_cache_config ( self ) : <EOL> config = NearCacheConfig ( ) <EOL> with self . assertRaises ( ValueError ) : <EOL> config . in_memory_format = <NUM_LIT:100> <EOL> with self . assertRaises ( ValueError ) : <EOL> config . eviction_policy = <NUM_LIT:100> <EOL> with self . assertRaises ( ValueError ) : <EOL> config . time_to_live_seconds = - <NUM_LIT:1> <EOL> with self . assertRaises ( ValueError ) : <EOL> config . max_idle_seconds = - <NUM_LIT:1> <EOL> with self . assertRaises ( ValueError ) : <EOL> config . eviction_max_size = <NUM_LIT:0> <EOL> def test_DataRecord_expire_time ( self ) : <EOL> now = current_time ( ) <EOL> print int ( now ) , now <EOL> data_rec = DataRecord ( "<STR_LIT:key>" , "<STR_LIT:value>" , create_time = now , ttl_seconds = <NUM_LIT:1> ) <EOL> sleep ( <NUM_LIT:2> ) <EOL> self . assertTrue ( data_rec . is_expired ( max_idle_seconds = <NUM_LIT:1000> ) ) <EOL> def test_DataRecord_max_idle_seconds ( self ) : <EOL> now = current_time ( ) <EOL> data_rec = DataRecord ( "<STR_LIT:key>" , "<STR_LIT:value>" , create_time = now , ttl_seconds = <NUM_LIT:1000> ) <EOL> sleep ( <NUM_LIT:2> ) <EOL> self . assertTrue ( data_rec . is_expired ( max_idle_seconds = <NUM_LIT:1> ) ) <EOL> def test_put_get_data ( self ) : <EOL> near_cache = self . create_near_cache ( self . service , IN_MEMORY_FORMAT . BINARY , <NUM_LIT:1000> , <NUM_LIT:1000> , EVICTION_POLICY . LRU , <NUM_LIT:1000> ) <EOL> key_data = self . service . to_data ( "<STR_LIT:key>" ) <EOL> near_cache [ key_data ] = "<STR_LIT:value>" <EOL> self . assertEqual ( "<STR_LIT:value>" , near_cache [ key_data ] ) <EOL> def test_put_get ( self ) : <EOL> near_cache = self . create_near_cache ( self . service , IN_MEMORY_FORMAT . OBJECT , <NUM_LIT:1000> , <NUM_LIT:1000> , EVICTION_POLICY . LRU , <NUM_LIT:1000> ) <EOL> for i in xrange ( <NUM_LIT:0> , <NUM_LIT> ) : <EOL> key = "<STR_LIT>" . format ( i ) <EOL> value = "<STR_LIT>" . format ( i ) <EOL> near_cache [ key ] = value <EOL> self . assertEqual ( value , near_cache [ key ] ) <EOL> self . assertEqual ( "<STR_LIT>" , near_cache [ "<STR_LIT>" ] ) <EOL> self . assertGreaterEqual ( near_cache . eviction_max_size * <NUM_LIT> , near_cache . __len__ ( ) ) <EOL> def test_expiry_time ( self ) : <EOL> near_cache = self . create_near_cache ( self . service , IN_MEMORY_FORMAT . OBJECT , <NUM_LIT:1> , <NUM_LIT:1000> , EVICTION_POLICY . LRU , <NUM_LIT:1000> ) <EOL> for i in xrange ( <NUM_LIT:0> , <NUM_LIT:1000> ) : <EOL> key = "<STR_LIT>" . format ( i ) <EOL> value = "<STR_LIT>" . format ( i ) <EOL> near_cache [ key ] = value <EOL> sleep ( <NUM_LIT:2> ) <EOL> for i in xrange ( <NUM_LIT> , <NUM_LIT> ) : <EOL> key = "<STR_LIT>" . format ( i ) <EOL> value = "<STR_LIT>" . format ( i ) <EOL> near_cache [ key ] = value <EOL> evict , expire = near_cache . get_statistics ( ) <EOL> self . assertLess ( evict , <NUM_LIT:2> ) <EOL> self . assertGreater ( expire , <NUM_LIT:8> ) <EOL> def test_LRU_time ( self ) : <EOL> near_cache = self . create_near_cache ( self . service , IN_MEMORY_FORMAT . OBJECT , <NUM_LIT:1000> , <NUM_LIT:1000> , EVICTION_POLICY . LRU , <NUM_LIT> , <NUM_LIT:16> , <NUM_LIT:16> ) <EOL> for i in xrange ( <NUM_LIT:0> , <NUM_LIT> ) : <EOL> key = "<STR_LIT>" . format ( i ) <EOL> value = "<STR_LIT>" . format ( i ) <EOL> near_cache [ key ] = value <EOL> self . assertEqual ( "<STR_LIT>" , near_cache [ "<STR_LIT>" ] ) <EOL> for i in xrange ( <NUM_LIT> , <NUM_LIT> ) : <EOL> key = "<STR_LIT>" . format ( i ) <EOL> value = "<STR_LIT>" . format ( i ) <EOL> near_cache [ key ] = value <EOL> evict , expire = near_cache . get_statistics ( ) <EOL> self . assertEqual ( expire , <NUM_LIT:0> ) <EOL> self . assertLess ( evict , <NUM_LIT> ) <EOL> def test_LRU_time_with_update ( self ) : <EOL> near_cache = self . create_near_cache ( self . service , IN_MEMORY_FORMAT . OBJECT , <NUM_LIT:1000> , <NUM_LIT:1000> , EVICTION_POLICY . LRU , <NUM_LIT:10> , <NUM_LIT:10> , <NUM_LIT:10> ) <EOL> for i in xrange ( <NUM_LIT:0> , <NUM_LIT:10> ) : <EOL> key = "<STR_LIT>" . format ( i ) <EOL> value = "<STR_LIT>" . format ( i ) <EOL> near_cache [ key ] = value <EOL> for i in xrange ( <NUM_LIT:0> , <NUM_LIT:9> ) : <EOL> key = "<STR_LIT>" . format ( i ) <EOL> value = "<STR_LIT>" . format ( i ) <EOL> self . assertEqual ( value , near_cache [ key ] ) <EOL> near_cache [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> with self . assertRaises ( KeyError ) : <EOL> val = near_cache [ "<STR_LIT>" ] <EOL> def test_LFU_time ( self ) : <EOL> near_cache = self . create_near_cache ( self . service , IN_MEMORY_FORMAT . BINARY , <NUM_LIT:1000> , <NUM_LIT:1000> , EVICTION_POLICY . LFU , <NUM_LIT:1000> ) <EOL> for i in xrange ( <NUM_LIT:0> , <NUM_LIT:1000> ) : <EOL> key = "<STR_LIT>" . format ( i ) <EOL> value = "<STR_LIT>" . format ( i ) <EOL> near_cache [ key ] = value <EOL> for j in xrange ( <NUM_LIT:0> , i + <NUM_LIT:1> ) : <EOL> v = near_cache [ key ] <EOL> for i in xrange ( <NUM_LIT> , <NUM_LIT> ) : <EOL> key = "<STR_LIT>" . format ( i ) <EOL> value = "<STR_LIT>" . format ( i ) <EOL> near_cache [ key ] = value <EOL> evict , expire = near_cache . get_statistics ( ) <EOL> self . assertEqual ( expire , <NUM_LIT:0> ) <EOL> self . assertLess ( evict , <NUM_LIT:1000> ) <EOL> def test_RANDOM_time ( self ) : <EOL> near_cache = self . create_near_cache ( self . service , IN_MEMORY_FORMAT . BINARY , <NUM_LIT:1000> , <NUM_LIT:1000> , EVICTION_POLICY . LFU , <NUM_LIT:1000> ) <EOL> for i in xrange ( <NUM_LIT:0> , <NUM_LIT> ) : <EOL> key = "<STR_LIT>" . format ( i ) <EOL> value = "<STR_LIT>" . format ( i ) <EOL> near_cache [ key ] = value <EOL> evict , expire = near_cache . get_statistics ( ) <EOL> self . assertEqual ( expire , <NUM_LIT:0> ) <EOL> self . assertGreaterEqual ( evict , <NUM_LIT:1000> ) <EOL> def create_near_cache ( self , service , im_format , ttl , max_idle , policy , max_size , eviction_sampling_count = None , <EOL> eviction_sampling_pool_size = None ) : <EOL> return NearCache ( service , im_format , ttl , max_idle , True , policy , max_size , eviction_sampling_count , <EOL> eviction_sampling_pool_size ) </s>
