<s> from time import time <EOL> import pytest <EOL> from funcy . flow import * <EOL> def test_silent ( ) : <EOL> assert silent ( int ) ( <NUM_LIT:1> ) == <NUM_LIT:1> <EOL> assert silent ( int ) ( '<STR_LIT:1>' ) == <NUM_LIT:1> <EOL> assert silent ( int ) ( '<STR_LIT:hello>' ) is None <EOL> assert silent ( str . upper ) ( '<STR_LIT:hello>' ) == '<STR_LIT>' <EOL> class MyError ( Exception ) : <EOL> pass <EOL> def test_ignore ( ) : <EOL> assert ignore ( Exception ) ( raiser ( Exception ) ) ( ) is None <EOL> assert ignore ( Exception ) ( raiser ( MyError ) ) ( ) is None <EOL> assert ignore ( ( TypeError , MyError ) ) ( raiser ( MyError ) ) ( ) is None <EOL> with pytest . raises ( TypeError ) : <EOL> ignore ( MyError ) ( raiser ( TypeError ) ) ( ) <EOL> assert ignore ( MyError , default = <NUM_LIT> ) ( raiser ( MyError ) ) ( ) == <NUM_LIT> <EOL> def test_raiser ( ) : <EOL> with pytest . raises ( Exception ) as e : raiser ( ) ( ) <EOL> assert e . type is Exception <EOL> with pytest . raises ( MyError ) : raiser ( MyError ) ( ) <EOL> with pytest . raises ( MyError ) as e : raiser ( MyError , '<STR_LIT>' ) ( ) <EOL> assert e . value . args == ( '<STR_LIT>' , ) <EOL> with pytest . raises ( MyError ) : raiser ( MyError ( '<STR_LIT>' ) ) ( ) <EOL> with pytest . raises ( MyError ) : raiser ( MyError ) ( '<STR_LIT>' , keyword = '<STR_LIT>' ) <EOL> def test_suppress ( ) : <EOL> with suppress ( Exception ) : <EOL> raise Exception <EOL> with suppress ( Exception ) : <EOL> raise MyError <EOL> with pytest . raises ( TypeError ) : <EOL> with suppress ( MyError ) : <EOL> raise TypeError <EOL> with suppress ( TypeError , MyError ) : <EOL> raise MyError <EOL> def test_retry ( ) : <EOL> calls = [ ] <EOL> def failing ( n = <NUM_LIT:1> ) : <EOL> if len ( calls ) < n : <EOL> calls . append ( <NUM_LIT:1> ) <EOL> raise MyError <EOL> return <NUM_LIT:1> <EOL> with pytest . raises ( MyError ) : failing ( ) <EOL> calls = [ ] <EOL> assert retry ( <NUM_LIT:2> , MyError ) ( failing ) ( ) == <NUM_LIT:1> <EOL> calls = [ ] <EOL> with pytest . raises ( MyError ) : retry ( <NUM_LIT:2> , MyError ) ( failing ) ( <NUM_LIT:2> ) <EOL> def test_retry_timeout ( ) : <EOL> def failing ( ) : <EOL> raise MyError <EOL> start_time = time ( ) <EOL> with pytest . raises ( MyError ) : retry ( <NUM_LIT:11> , MyError , timeout = <NUM_LIT> ) ( failing ) ( ) <EOL> assert <NUM_LIT:0.1> < time ( ) - start_time < <NUM_LIT> <EOL> start_time = time ( ) <EOL> with pytest . raises ( MyError ) : retry ( <NUM_LIT:4> , MyError , timeout = lambda a : <NUM_LIT> * <NUM_LIT:2> ** a ) ( failing ) ( ) <EOL> d = time ( ) - start_time <EOL> assert <NUM_LIT> < d < <NUM_LIT> <EOL> def test_retry_many_errors ( ) : <EOL> calls = [ ] <EOL> def failing ( n = <NUM_LIT:1> ) : <EOL> if len ( calls ) < n : <EOL> calls . append ( <NUM_LIT:1> ) <EOL> raise MyError <EOL> return <NUM_LIT:1> <EOL> assert retry ( <NUM_LIT:2> , ( MyError , RuntimeError ) ) ( failing ) ( ) == <NUM_LIT:1> <EOL> calls = [ ] <EOL> assert retry ( <NUM_LIT:2> , [ MyError , RuntimeError ] ) ( failing ) ( ) == <NUM_LIT:1> <EOL> def test_fallback ( ) : <EOL> assert fallback ( raiser ( ) , lambda : <NUM_LIT:1> ) == <NUM_LIT:1> <EOL> with pytest . raises ( Exception ) : fallback ( ( raiser ( ) , MyError ) , lambda : <NUM_LIT:1> ) <EOL> assert fallback ( ( raiser ( MyError ) , MyError ) , lambda : <NUM_LIT:1> ) == <NUM_LIT:1> <EOL> def test_limit_error_rate ( ) : <EOL> calls = [ ] <EOL> @ limit_error_rate ( <NUM_LIT:2> , <NUM_LIT> , MyError ) <EOL> def limited ( x ) : <EOL> calls . append ( x ) <EOL> raise TypeError <EOL> with pytest . raises ( TypeError ) : limited ( <NUM_LIT:1> ) <EOL> with pytest . raises ( TypeError ) : limited ( <NUM_LIT:2> ) <EOL> with pytest . raises ( MyError ) : limited ( <NUM_LIT:3> ) <EOL> assert calls == [ <NUM_LIT:1> , <NUM_LIT:2> ] <EOL> def test_post_processing ( ) : <EOL> @ post_processing ( max ) <EOL> def my_max ( l ) : <EOL> return l <EOL> assert my_max ( [ <NUM_LIT:1> , <NUM_LIT:3> , <NUM_LIT:2> ] ) == <NUM_LIT:3> <EOL> def test_collecting ( ) : <EOL> @ collecting <EOL> def doubles ( l ) : <EOL> for i in l : <EOL> yield i * <NUM_LIT:2> <EOL> assert doubles ( [ <NUM_LIT:1> , <NUM_LIT:2> ] ) == [ <NUM_LIT:2> , <NUM_LIT:4> ] <EOL> def test_once ( ) : <EOL> calls = [ ] <EOL> @ once <EOL> def call ( n ) : <EOL> calls . append ( n ) <EOL> return n <EOL> call ( <NUM_LIT:1> ) <EOL> call ( <NUM_LIT:2> ) <EOL> assert calls == [ <NUM_LIT:1> ] <EOL> def test_once_per ( ) : <EOL> calls = [ ] <EOL> @ once_per ( '<STR_LIT:n>' ) <EOL> def call ( n , x = None ) : <EOL> calls . append ( n ) <EOL> return n <EOL> call ( <NUM_LIT:1> ) <EOL> call ( <NUM_LIT:2> ) <EOL> call ( <NUM_LIT:1> , <NUM_LIT> ) <EOL> assert calls == [ <NUM_LIT:1> , <NUM_LIT:2> ] <EOL> def test_once_per_args ( ) : <EOL> calls = [ ] <EOL> @ once_per_args <EOL> def call ( n , x = None ) : <EOL> calls . append ( n ) <EOL> return n <EOL> call ( <NUM_LIT:1> ) <EOL> call ( <NUM_LIT:2> ) <EOL> call ( <NUM_LIT:1> , <NUM_LIT> ) <EOL> assert calls == [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:1> ] <EOL> call ( <NUM_LIT:1> ) <EOL> assert calls == [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:1> ] </s>
