<s> import time <EOL> import unittest <EOL> from retrying import RetryError <EOL> from retrying import Retrying <EOL> from retrying import retry <EOL> class TestStopConditions ( unittest . TestCase ) : <EOL> def test_never_stop ( self ) : <EOL> r = Retrying ( ) <EOL> self . assertFalse ( r . stop ( <NUM_LIT:3> , <NUM_LIT> ) ) <EOL> def test_stop_after_attempt ( self ) : <EOL> r = Retrying ( stop_max_attempt_number = <NUM_LIT:3> ) <EOL> self . assertFalse ( r . stop ( <NUM_LIT:2> , <NUM_LIT> ) ) <EOL> self . assertTrue ( r . stop ( <NUM_LIT:3> , <NUM_LIT> ) ) <EOL> self . assertTrue ( r . stop ( <NUM_LIT:4> , <NUM_LIT> ) ) <EOL> def test_stop_after_delay ( self ) : <EOL> r = Retrying ( stop_max_delay = <NUM_LIT:1000> ) <EOL> self . assertFalse ( r . stop ( <NUM_LIT:2> , <NUM_LIT> ) ) <EOL> self . assertTrue ( r . stop ( <NUM_LIT:2> , <NUM_LIT:1000> ) ) <EOL> self . assertTrue ( r . stop ( <NUM_LIT:2> , <NUM_LIT> ) ) <EOL> def test_legacy_explicit_stop_type ( self ) : <EOL> Retrying ( stop = "<STR_LIT>" ) <EOL> def test_stop_func ( self ) : <EOL> r = Retrying ( stop_func = lambda attempt , delay : attempt == delay ) <EOL> self . assertFalse ( r . stop ( <NUM_LIT:1> , <NUM_LIT:3> ) ) <EOL> self . assertFalse ( r . stop ( <NUM_LIT:100> , <NUM_LIT> ) ) <EOL> self . assertTrue ( r . stop ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> class TestWaitConditions ( unittest . TestCase ) : <EOL> def test_no_sleep ( self ) : <EOL> r = Retrying ( ) <EOL> self . assertEqual ( <NUM_LIT:0> , r . wait ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> def test_fixed_sleep ( self ) : <EOL> r = Retrying ( wait_fixed = <NUM_LIT:1000> ) <EOL> self . assertEqual ( <NUM_LIT:1000> , r . wait ( <NUM_LIT:12> , <NUM_LIT> ) ) <EOL> def test_incrementing_sleep ( self ) : <EOL> r = Retrying ( wait_incrementing_start = <NUM_LIT> , wait_incrementing_increment = <NUM_LIT:100> ) <EOL> self . assertEqual ( <NUM_LIT> , r . wait ( <NUM_LIT:1> , <NUM_LIT> ) ) <EOL> self . assertEqual ( <NUM_LIT> , r . wait ( <NUM_LIT:2> , <NUM_LIT> ) ) <EOL> self . assertEqual ( <NUM_LIT> , r . wait ( <NUM_LIT:3> , <NUM_LIT> ) ) <EOL> def test_random_sleep ( self ) : <EOL> r = Retrying ( wait_random_min = <NUM_LIT:1000> , wait_random_max = <NUM_LIT> ) <EOL> times = set ( ) <EOL> times . add ( r . wait ( <NUM_LIT:1> , <NUM_LIT> ) ) <EOL> times . add ( r . wait ( <NUM_LIT:1> , <NUM_LIT> ) ) <EOL> times . add ( r . wait ( <NUM_LIT:1> , <NUM_LIT> ) ) <EOL> times . add ( r . wait ( <NUM_LIT:1> , <NUM_LIT> ) ) <EOL> self . assertTrue ( len ( times ) > <NUM_LIT:1> ) <EOL> for t in times : <EOL> self . assertTrue ( t >= <NUM_LIT:1000> ) <EOL> self . assertTrue ( t <= <NUM_LIT> ) <EOL> def test_random_sleep_without_min ( self ) : <EOL> r = Retrying ( wait_random_max = <NUM_LIT> ) <EOL> times = set ( ) <EOL> times . add ( r . wait ( <NUM_LIT:1> , <NUM_LIT> ) ) <EOL> times . add ( r . wait ( <NUM_LIT:1> , <NUM_LIT> ) ) <EOL> times . add ( r . wait ( <NUM_LIT:1> , <NUM_LIT> ) ) <EOL> times . add ( r . wait ( <NUM_LIT:1> , <NUM_LIT> ) ) <EOL> self . assertTrue ( len ( times ) > <NUM_LIT:1> ) <EOL> for t in times : <EOL> self . assertTrue ( t >= <NUM_LIT:0> ) <EOL> self . assertTrue ( t <= <NUM_LIT> ) <EOL> def test_exponential ( self ) : <EOL> r = Retrying ( wait_exponential_max = <NUM_LIT> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:1> , <NUM_LIT:0> ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:2> , <NUM_LIT:0> ) , <NUM_LIT:4> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:3> , <NUM_LIT:0> ) , <NUM_LIT:8> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:4> , <NUM_LIT:0> ) , <NUM_LIT:16> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:5> , <NUM_LIT:0> ) , <NUM_LIT:32> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:6> , <NUM_LIT:0> ) , <NUM_LIT:64> ) <EOL> def test_exponential_with_max_wait ( self ) : <EOL> r = Retrying ( wait_exponential_max = <NUM_LIT> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:1> , <NUM_LIT:0> ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:2> , <NUM_LIT:0> ) , <NUM_LIT:4> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:3> , <NUM_LIT:0> ) , <NUM_LIT:8> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:4> , <NUM_LIT:0> ) , <NUM_LIT:16> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:5> , <NUM_LIT:0> ) , <NUM_LIT:32> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:6> , <NUM_LIT:0> ) , <NUM_LIT> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:7> , <NUM_LIT:0> ) , <NUM_LIT> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:50> , <NUM_LIT:0> ) , <NUM_LIT> ) <EOL> def test_exponential_with_max_wait_and_multiplier ( self ) : <EOL> r = Retrying ( wait_exponential_max = <NUM_LIT> , wait_exponential_multiplier = <NUM_LIT:1000> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:1> , <NUM_LIT:0> ) , <NUM_LIT> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:2> , <NUM_LIT:0> ) , <NUM_LIT> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:3> , <NUM_LIT:0> ) , <NUM_LIT> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:4> , <NUM_LIT:0> ) , <NUM_LIT> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:5> , <NUM_LIT:0> ) , <NUM_LIT> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:6> , <NUM_LIT:0> ) , <NUM_LIT> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:7> , <NUM_LIT:0> ) , <NUM_LIT> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:50> , <NUM_LIT:0> ) , <NUM_LIT> ) <EOL> def test_legacy_explicit_wait_type ( self ) : <EOL> Retrying ( wait = "<STR_LIT>" ) <EOL> def test_wait_func ( self ) : <EOL> r = Retrying ( wait_func = lambda attempt , delay : attempt * delay ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:1> , <NUM_LIT:5> ) , <NUM_LIT:5> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:2> , <NUM_LIT:11> ) , <NUM_LIT> ) <EOL> self . assertEqual ( r . wait ( <NUM_LIT:10> , <NUM_LIT:100> ) , <NUM_LIT:1000> ) <EOL> class NoneReturnUntilAfterCount : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , count ) : <EOL> self . counter = <NUM_LIT:0> <EOL> self . count = count <EOL> def go ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . counter < self . count : <EOL> self . counter += <NUM_LIT:1> <EOL> return None <EOL> return True <EOL> class NoIOErrorAfterCount : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , count ) : <EOL> self . counter = <NUM_LIT:0> <EOL> self . count = count <EOL> def go ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . counter < self . count : <EOL> self . counter += <NUM_LIT:1> <EOL> raise IOError ( "<STR_LIT>" ) <EOL> return True <EOL> class NoNameErrorAfterCount : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , count ) : <EOL> self . counter = <NUM_LIT:0> <EOL> self . count = count <EOL> def go ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . counter < self . count : <EOL> self . counter += <NUM_LIT:1> <EOL> raise NameError ( "<STR_LIT>" ) <EOL> return True <EOL> class CustomError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , value ) : <EOL> self . value = value <EOL> def __str__ ( self ) : <EOL> return repr ( self . value ) <EOL> class NoCustomErrorAfterCount : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , count ) : <EOL> self . counter = <NUM_LIT:0> <EOL> self . count = count <EOL> def go ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . counter < self . count : <EOL> self . counter += <NUM_LIT:1> <EOL> derived_message = "<STR_LIT>" <EOL> raise CustomError ( derived_message ) <EOL> return True <EOL> def retry_if_result_none ( result ) : <EOL> return result is None <EOL> def retry_if_exception_of_type ( retryable_types ) : <EOL> def retry_if_exception_these_types ( exception ) : <EOL> print ( "<STR_LIT>" . format ( str ( type ( exception ) ) ) ) <EOL> return isinstance ( exception , retryable_types ) <EOL> return retry_if_exception_these_types <EOL> def current_time_ms ( ) : <EOL> return int ( round ( time . time ( ) * <NUM_LIT:1000> ) ) <EOL> @ retry ( wait_fixed = <NUM_LIT:50> , retry_on_result = retry_if_result_none ) <EOL> def _retryable_test_with_wait ( thing ) : <EOL> return thing . go ( ) <EOL> @ retry ( stop_max_attempt_number = <NUM_LIT:3> , retry_on_result = retry_if_result_none ) <EOL> def _retryable_test_with_stop ( thing ) : <EOL> return thing . go ( ) <EOL> @ retry ( retry_on_exception = retry_if_exception_of_type ( IOError ) ) <EOL> def _retryable_test_with_exception_type_io ( thing ) : <EOL> return thing . go ( ) <EOL> @ retry ( retry_on_exception = retry_if_exception_of_type ( IOError ) , wrap_exception = True ) <EOL> def _retryable_test_with_exception_type_io_wrap ( thing ) : <EOL> return thing . go ( ) <EOL> @ retry ( <EOL> stop_max_attempt_number = <NUM_LIT:3> , <EOL> retry_on_exception = retry_if_exception_of_type ( IOError ) ) <EOL> def _retryable_test_with_exception_type_io_attempt_limit ( thing ) : <EOL> return thing . go ( ) <EOL> @ retry ( <EOL> stop_max_attempt_number = <NUM_LIT:3> , <EOL> retry_on_exception = retry_if_exception_of_type ( IOError ) , <EOL> wrap_exception = True ) <EOL> def _retryable_test_with_exception_type_io_attempt_limit_wrap ( thing ) : <EOL> return thing . go ( ) <EOL> @ retry <EOL> def _retryable_default ( thing ) : <EOL> return thing . go ( ) <EOL> @ retry ( ) <EOL> def _retryable_default_f ( thing ) : <EOL> return thing . go ( ) <EOL> @ retry ( retry_on_exception = retry_if_exception_of_type ( CustomError ) ) <EOL> def _retryable_test_with_exception_type_custom ( thing ) : <EOL> return thing . go ( ) <EOL> @ retry ( retry_on_exception = retry_if_exception_of_type ( CustomError ) , wrap_exception = True ) <EOL> def _retryable_test_with_exception_type_custom_wrap ( thing ) : <EOL> return thing . go ( ) <EOL> @ retry ( <EOL> stop_max_attempt_number = <NUM_LIT:3> , <EOL> retry_on_exception = retry_if_exception_of_type ( CustomError ) ) <EOL> def _retryable_test_with_exception_type_custom_attempt_limit ( thing ) : <EOL> return thing . go ( ) <EOL> @ retry ( <EOL> stop_max_attempt_number = <NUM_LIT:3> , <EOL> retry_on_exception = retry_if_exception_of_type ( CustomError ) , <EOL> wrap_exception = True ) <EOL> def _retryable_test_with_exception_type_custom_attempt_limit_wrap ( thing ) : <EOL> return thing . go ( ) <EOL> class TestDecoratorWrapper ( unittest . TestCase ) : <EOL> def test_with_wait ( self ) : <EOL> start = current_time_ms ( ) <EOL> result = _retryable_test_with_wait ( NoneReturnUntilAfterCount ( <NUM_LIT:5> ) ) <EOL> t = current_time_ms ( ) - start <EOL> self . assertTrue ( t >= <NUM_LIT> ) <EOL> self . assertTrue ( result ) <EOL> def test_with_stop_on_return_value ( self ) : <EOL> try : <EOL> _retryable_test_with_stop ( NoneReturnUntilAfterCount ( <NUM_LIT:5> ) ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> except RetryError as re : <EOL> self . assertFalse ( re . last_attempt . has_exception ) <EOL> self . assertEqual ( <NUM_LIT:3> , re . last_attempt . attempt_number ) <EOL> self . assertTrue ( re . last_attempt . value is None ) <EOL> print ( re ) <EOL> def test_with_stop_on_exception ( self ) : <EOL> try : <EOL> _retryable_test_with_stop ( NoIOErrorAfterCount ( <NUM_LIT:5> ) ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> except IOError as re : <EOL> self . assertTrue ( isinstance ( re , IOError ) ) <EOL> print ( re ) <EOL> def test_retry_if_exception_of_type ( self ) : <EOL> self . assertTrue ( _retryable_test_with_exception_type_io ( NoIOErrorAfterCount ( <NUM_LIT:5> ) ) ) <EOL> try : <EOL> _retryable_test_with_exception_type_io ( NoNameErrorAfterCount ( <NUM_LIT:5> ) ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> except NameError as n : <EOL> self . assertTrue ( isinstance ( n , NameError ) ) <EOL> print ( n ) <EOL> try : <EOL> _retryable_test_with_exception_type_io_attempt_limit_wrap ( NoIOErrorAfterCount ( <NUM_LIT:5> ) ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> except RetryError as re : <EOL> self . assertEqual ( <NUM_LIT:3> , re . last_attempt . attempt_number ) <EOL> self . assertTrue ( re . last_attempt . has_exception ) <EOL> self . assertTrue ( re . last_attempt . value [ <NUM_LIT:0> ] is not None ) <EOL> self . assertTrue ( isinstance ( re . last_attempt . value [ <NUM_LIT:1> ] , IOError ) ) <EOL> self . assertTrue ( re . last_attempt . value [ <NUM_LIT:2> ] is not None ) <EOL> print ( re ) <EOL> self . assertTrue ( _retryable_test_with_exception_type_custom ( NoCustomErrorAfterCount ( <NUM_LIT:5> ) ) ) <EOL> try : <EOL> _retryable_test_with_exception_type_custom ( NoNameErrorAfterCount ( <NUM_LIT:5> ) ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> except NameError as n : <EOL> self . assertTrue ( isinstance ( n , NameError ) ) <EOL> print ( n ) <EOL> try : <EOL> _retryable_test_with_exception_type_custom_attempt_limit_wrap ( NoCustomErrorAfterCount ( <NUM_LIT:5> ) ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> except RetryError as re : <EOL> self . assertEqual ( <NUM_LIT:3> , re . last_attempt . attempt_number ) <EOL> self . assertTrue ( re . last_attempt . has_exception ) <EOL> self . assertTrue ( re . last_attempt . value [ <NUM_LIT:0> ] is not None ) <EOL> self . assertTrue ( isinstance ( re . last_attempt . value [ <NUM_LIT:1> ] , CustomError ) ) <EOL> self . assertTrue ( re . last_attempt . value [ <NUM_LIT:2> ] is not None ) <EOL> print ( re ) <EOL> def test_wrapped_exception ( self ) : <EOL> self . assertTrue ( _retryable_test_with_exception_type_io_wrap ( NoIOErrorAfterCount ( <NUM_LIT:5> ) ) ) <EOL> try : <EOL> _retryable_test_with_exception_type_io_wrap ( NoNameErrorAfterCount ( <NUM_LIT:5> ) ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> except RetryError as re : <EOL> self . assertTrue ( isinstance ( re . last_attempt . value [ <NUM_LIT:1> ] , NameError ) ) <EOL> print ( re ) <EOL> try : <EOL> _retryable_test_with_exception_type_io_attempt_limit_wrap ( NoIOErrorAfterCount ( <NUM_LIT:5> ) ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> except RetryError as re : <EOL> self . assertEqual ( <NUM_LIT:3> , re . last_attempt . attempt_number ) <EOL> self . assertTrue ( re . last_attempt . has_exception ) <EOL> self . assertTrue ( re . last_attempt . value [ <NUM_LIT:0> ] is not None ) <EOL> self . assertTrue ( isinstance ( re . last_attempt . value [ <NUM_LIT:1> ] , IOError ) ) <EOL> self . assertTrue ( re . last_attempt . value [ <NUM_LIT:2> ] is not None ) <EOL> print ( re ) <EOL> self . assertTrue ( _retryable_test_with_exception_type_custom_wrap ( NoCustomErrorAfterCount ( <NUM_LIT:5> ) ) ) <EOL> try : <EOL> _retryable_test_with_exception_type_custom_wrap ( NoNameErrorAfterCount ( <NUM_LIT:5> ) ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> except RetryError as re : <EOL> self . assertTrue ( re . last_attempt . value [ <NUM_LIT:0> ] is not None ) <EOL> self . assertTrue ( isinstance ( re . last_attempt . value [ <NUM_LIT:1> ] , NameError ) ) <EOL> self . assertTrue ( re . last_attempt . value [ <NUM_LIT:2> ] is not None ) <EOL> print ( re ) <EOL> try : <EOL> _retryable_test_with_exception_type_custom_attempt_limit_wrap ( NoCustomErrorAfterCount ( <NUM_LIT:5> ) ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> except RetryError as re : <EOL> self . assertEqual ( <NUM_LIT:3> , re . last_attempt . attempt_number ) <EOL> self . assertTrue ( re . last_attempt . has_exception ) <EOL> self . assertTrue ( re . last_attempt . value [ <NUM_LIT:0> ] is not None ) <EOL> self . assertTrue ( isinstance ( re . last_attempt . value [ <NUM_LIT:1> ] , CustomError ) ) <EOL> self . assertTrue ( re . last_attempt . value [ <NUM_LIT:2> ] is not None ) <EOL> self . assertTrue ( "<STR_LIT>" in str ( re . last_attempt . value [ <NUM_LIT:1> ] ) ) <EOL> print ( re ) <EOL> def test_defaults ( self ) : <EOL> self . assertTrue ( _retryable_default ( NoNameErrorAfterCount ( <NUM_LIT:5> ) ) ) <EOL> self . assertTrue ( _retryable_default_f ( NoNameErrorAfterCount ( <NUM_LIT:5> ) ) ) <EOL> self . assertTrue ( _retryable_default ( NoCustomErrorAfterCount ( <NUM_LIT:5> ) ) ) <EOL> self . assertTrue ( _retryable_default_f ( NoCustomErrorAfterCount ( <NUM_LIT:5> ) ) ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
