<s> from __future__ import absolute_import , unicode_literals <EOL> import pickle <EOL> from itertools import count <EOL> from kombu . five import items <EOL> from kombu . utils . functional import LRUCache , memoize , lazy , maybe_evaluate <EOL> from kombu . tests . case import Case , skip <EOL> def double ( x ) : <EOL> return x * <NUM_LIT:2> <EOL> class test_LRUCache ( Case ) : <EOL> def test_expires ( self ) : <EOL> limit = <NUM_LIT:100> <EOL> x = LRUCache ( limit = limit ) <EOL> slots = list ( range ( limit * <NUM_LIT:2> ) ) <EOL> for i in slots : <EOL> x [ i ] = i <EOL> self . assertListEqual ( list ( x . keys ( ) ) , list ( slots [ limit : ] ) ) <EOL> self . assertTrue ( x . items ( ) ) <EOL> self . assertTrue ( x . values ( ) ) <EOL> def test_is_pickleable ( self ) : <EOL> x = LRUCache ( limit = <NUM_LIT:10> ) <EOL> x . update ( luke = <NUM_LIT:1> , leia = <NUM_LIT:2> ) <EOL> y = pickle . loads ( pickle . dumps ( x ) ) <EOL> self . assertEqual ( y . limit , y . limit ) <EOL> self . assertEqual ( y , x ) <EOL> def test_update_expires ( self ) : <EOL> limit = <NUM_LIT:100> <EOL> x = LRUCache ( limit = limit ) <EOL> slots = list ( range ( limit * <NUM_LIT:2> ) ) <EOL> for i in slots : <EOL> x . update ( { i : i } ) <EOL> self . assertListEqual ( list ( x . keys ( ) ) , list ( slots [ limit : ] ) ) <EOL> def test_least_recently_used ( self ) : <EOL> x = LRUCache ( <NUM_LIT:3> ) <EOL> x [ <NUM_LIT:1> ] , x [ <NUM_LIT:2> ] , x [ <NUM_LIT:3> ] = <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> <EOL> self . assertEqual ( list ( x . keys ( ) ) , [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) <EOL> x [ <NUM_LIT:4> ] , x [ <NUM_LIT:5> ] = <NUM_LIT:4> , <NUM_LIT:5> <EOL> self . assertEqual ( list ( x . keys ( ) ) , [ <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> ] ) <EOL> x [ <NUM_LIT:3> ] <EOL> x [ <NUM_LIT:6> ] = <NUM_LIT:6> <EOL> self . assertEqual ( list ( x . keys ( ) ) , [ <NUM_LIT:5> , <NUM_LIT:3> , <NUM_LIT:6> ] ) <EOL> x [ <NUM_LIT:7> ] = <NUM_LIT:7> <EOL> self . assertEqual ( list ( x . keys ( ) ) , [ <NUM_LIT:3> , <NUM_LIT:6> , <NUM_LIT:7> ] ) <EOL> def test_update_larger_than_cache_size ( self ) : <EOL> x = LRUCache ( <NUM_LIT:2> ) <EOL> x . update ( { x : x for x in range ( <NUM_LIT:100> ) } ) <EOL> self . assertEqual ( list ( x . keys ( ) ) , [ <NUM_LIT> , <NUM_LIT> ] ) <EOL> def test_items ( self ) : <EOL> c = LRUCache ( ) <EOL> c . update ( a = <NUM_LIT:1> , b = <NUM_LIT:2> , c = <NUM_LIT:3> ) <EOL> self . assertTrue ( list ( items ( c ) ) ) <EOL> def test_incr ( self ) : <EOL> c = LRUCache ( ) <EOL> c . update ( a = '<STR_LIT:1>' ) <EOL> c . incr ( '<STR_LIT:a>' ) <EOL> self . assertEqual ( c [ '<STR_LIT:a>' ] , '<STR_LIT:2>' ) <EOL> class test_memoize ( Case ) : <EOL> def test_memoize ( self ) : <EOL> counter = count ( <NUM_LIT:1> ) <EOL> @ memoize ( maxsize = <NUM_LIT:2> ) <EOL> def x ( i ) : <EOL> return next ( counter ) <EOL> self . assertEqual ( x ( <NUM_LIT:1> ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( x ( <NUM_LIT:1> ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( x ( <NUM_LIT:2> ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( x ( <NUM_LIT:3> ) , <NUM_LIT:3> ) <EOL> self . assertEqual ( x ( <NUM_LIT:1> ) , <NUM_LIT:4> ) <EOL> x . clear ( ) <EOL> self . assertEqual ( x ( <NUM_LIT:3> ) , <NUM_LIT:5> ) <EOL> class test_lazy ( Case ) : <EOL> def test__str__ ( self ) : <EOL> self . assertEqual ( <EOL> str ( lazy ( lambda : '<STR_LIT>' ) ) , <EOL> '<STR_LIT>' , <EOL> ) <EOL> def test__repr__ ( self ) : <EOL> self . assertEqual ( <EOL> repr ( lazy ( lambda : '<STR_LIT>' ) ) . strip ( '<STR_LIT:u>' ) , <EOL> "<STR_LIT>" , <EOL> ) <EOL> @ skip . if_python3 ( ) <EOL> def test__cmp__ ( self ) : <EOL> self . assertEqual ( lazy ( lambda : <NUM_LIT:10> ) . __cmp__ ( lazy ( lambda : <NUM_LIT:20> ) ) , - <NUM_LIT:1> ) <EOL> self . assertEqual ( lazy ( lambda : <NUM_LIT:10> ) . __cmp__ ( <NUM_LIT:5> ) , <NUM_LIT:1> ) <EOL> def test_evaluate ( self ) : <EOL> self . assertEqual ( lazy ( lambda : <NUM_LIT:2> + <NUM_LIT:2> ) ( ) , <NUM_LIT:4> ) <EOL> self . assertEqual ( lazy ( lambda x : x * <NUM_LIT:4> , <NUM_LIT:2> ) , <NUM_LIT:8> ) <EOL> self . assertEqual ( lazy ( lambda x : x * <NUM_LIT:8> , <NUM_LIT:2> ) ( ) , <NUM_LIT:16> ) <EOL> def test_cmp ( self ) : <EOL> self . assertEqual ( lazy ( lambda : <NUM_LIT:10> ) , lazy ( lambda : <NUM_LIT:10> ) ) <EOL> self . assertNotEqual ( lazy ( lambda : <NUM_LIT:10> ) , lazy ( lambda : <NUM_LIT:20> ) ) <EOL> def test__reduce__ ( self ) : <EOL> x = lazy ( double , <NUM_LIT:4> ) <EOL> y = pickle . loads ( pickle . dumps ( x ) ) <EOL> self . assertEqual ( x ( ) , y ( ) ) <EOL> def test__deepcopy__ ( self ) : <EOL> from copy import deepcopy <EOL> x = lazy ( double , <NUM_LIT:4> ) <EOL> y = deepcopy ( x ) <EOL> self . assertEqual ( x . _fun , y . _fun ) <EOL> self . assertEqual ( x . _args , y . _args ) <EOL> self . assertEqual ( x ( ) , y ( ) ) <EOL> class test_maybe_evaluate ( Case ) : <EOL> def test_evaluates ( self ) : <EOL> self . assertEqual ( maybe_evaluate ( lazy ( lambda : <NUM_LIT:10> ) ) , <NUM_LIT:10> ) <EOL> self . assertEqual ( maybe_evaluate ( <NUM_LIT:20> ) , <NUM_LIT:20> ) </s>
