## Copyright 2013 Ray Holder
##
## Licensed under the Apache License, Version 2.0 (the "License");
## you may not use this file except in compliance with the License.
## You may obtain a copy of the License at
##
## http://www.apache.org/licenses/LICENSE-2.0
##
## Unless required by applicable law or agreed to in writing, software
## distributed under the License is distributed on an "AS IS" BASIS,
## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
## See the License for the specific language governing permissions and
## limitations under the License.

import time
import unittest

from retrying import RetryError
from retrying import Retrying
from retrying import retry


class TestStopConditions(unittest.TestCase):

    def test_never_stop(self):
        r = Retrying()
        self.assertFalse(r.stop(3, 6546))

    def test_stop_after_attempt(self):
        r = Retrying(stop_max_attempt_number=3)
        self.assertFalse(r.stop(2, 6546))
        self.assertTrue(r.stop(3, 6546))
        self.assertTrue(r.stop(4, 6546))

    def test_stop_after_delay(self):
        r = Retrying(stop_max_delay=1000)
        self.assertFalse(r.stop(2, 999))
        self.assertTrue(r.stop(2, 1000))
        self.assertTrue(r.stop(2, 1001))

    def test_legacy_explicit_stop_type(self):
        Retrying(stop="stop_after_attempt")

    def test_stop_func(self):
        r = Retrying(stop_func=lambda attempt, delay: attempt == delay)
        self.assertFalse(r.stop(1, 3))
        self.assertFalse(r.stop(100, 99))
        self.assertTrue(r.stop(101, 101))


class TestWaitConditions(unittest.TestCase):

    def test_no_sleep(self):
        r = Retrying()
        self.assertEqual(0, r.wait(18, 9879))

    def test_fixed_sleep(self):
        r = Retrying(wait_fixed=1000)
        self.assertEqual(1000, r.wait(12, 6546))

    def test_incrementing_sleep(self):
        r = Retrying(wait_incrementing_start=500, wait_incrementing_increment=100)
        self.assertEqual(500, r.wait(1, 6546))
        self.assertEqual(600, r.wait(2, 6546))
        self.assertEqual(700, r.wait(3, 6546))

    def test_random_sleep(self):
        r = Retrying(wait_random_min=1000, wait_random_max=2000)
        times = set()
        times.add(r.wait(1, 6546))
        times.add(r.wait(1, 6546))
        times.add(r.wait(1, 6546))
        times.add(r.wait(1, 6546))

        # this is kind of non-deterministic...
        self.assertTrue(len(times) > 1)
        for t in times:
            self.assertTrue(t >= 1000)
            self.assertTrue(t <= 2000)

    def test_random_sleep_without_min(self):
        r = Retrying(wait_random_max=2000)
        times = set()
        times.add(r.wait(1, 6546))
        times.add(r.wait(1, 6546))
        times.add(r.wait(1, 6546))
        times.add(r.wait(1, 6546))

        # this is kind of non-deterministic...
        self.assertTrue(len(times) > 1)
        for t in times:
            self.assertTrue(t >= 0)
            self.assertTrue(t <= 2000)

    def test_exponential(self):
        r = Retrying(wait_exponential_max=100000)
        self.assertEqual(r.wait(1, 0), 2)
        self.assertEqual(r.wait(2, 0), 4)
        self.assertEqual(r.wait(3, 0), 8)
        self.assertEqual(r.wait(4, 0), 16)
        self.assertEqual(r.wait(5, 0), 32)
        self.assertEqual(r.wait(6, 0), 64)

    def test_exponential_with_max_wait(self):
        r = Retrying(wait_exponential_max=40)
        self.assertEqual(r.wait(1, 0), 2)
        self.assertEqual(r.wait(2, 0), 4)
        self.assertEqual(r.wait(3, 0), 8)
        self.assertEqual(r.wait(4, 0), 16)
        self.assertEqual(r.wait(5, 0), 32)
        self.assertEqual(r.wait(6, 0), 40)
        self.assertEqual(r.wait(7, 0), 40)
        self.assertEqual(r.wait(50, 0), 40)

    def test_exponential_with_max_wait_and_multiplier(self):
        r = Retrying(wait_exponential_max=50000, wait_exponential_multiplier=1000)
        self.assertEqual(r.wait(1, 0), 2000)
        self.assertEqual(r.wait(2, 0), 4000)
        self.assertEqual(r.wait(3, 0), 8000)
        self.assertEqual(r.wait(4, 0), 16000)
        self.assertEqual(r.wait(5, 0), 32000)
        self.assertEqual(r.wait(6, 0), 50000)
        self.assertEqual(r.wait(7, 0), 50000)
        self.assertEqual(r.wait(50, 0), 50000)

    def test_legacy_explicit_wait_type(self):
        Retrying(wait="exponential_sleep")

    def test_wait_func(self):
        r = Retrying(wait_func=lambda attempt, delay: attempt * delay)
        self.assertEqual(r.wait(1, 5), 5)
        self.assertEqual(r.wait(2, 11), 22)
        self.assertEqual(r.wait(10, 100), 1000)


class NoneReturnUntilAfterCount:
    """
    This class holds counter state for invoking a method several times in a row.
    """

    def __init__(self, count):
        self.counter = 0
        self.count = count

    def go(self):
        """
        Return None until after count threshold has been crossed, then return True.
        """
        if self.counter < self.count:
            self.counter += 1
            return None
        return True


class NoIOErrorAfterCount:
    """
    This class holds counter state for invoking a method several times in a row.
    """

    def __init__(self, count):
        self.counter = 0
        self.count = count

    def go(self):
        """
        Raise an IOError until after count threshold has been crossed, then return True.
        """
        if self.counter < self.count:
            self.counter += 1
            raise IOError("Hi there, I'm an IOError")
        return True


class NoNameErrorAfterCount:
    """
    This class holds counter state for invoking a method several times in a row.
    """

    def __init__(self, count):
        self.counter = 0
        self.count = count

    def go(self):
        """
        Raise a NameError until after count threshold has been crossed, then return True.
        """
        if self.counter < self.count:
            self.counter += 1
            raise NameError("Hi there, I'm a NameError")
        return True


class CustomError(Exception):
    """
    This is a custom exception class. Note that For Python 2.x, we don't
    strictly need to extend BaseException, however, Python 3.x will complain.
    While this test suite won't run correctly under Python 3.x without
    extending from the Python exception hierarchy, the actual module code is
    backwards compatible Python 2.x and will allow for cases where exception
    classes don't extend from the hierarchy.
    """

    def __init__(self, value):
        self.value = value

    def __str__(self):
        return repr(self.value)


class NoCustomErrorAfterCount:
    """
    This class holds counter state for invoking a method several times in a row.
    """

    def __init__(self, count):
        self.counter = 0
        self.count = count

    def go(self):
        """
        Raise a CustomError until after count threshold has been crossed, then return True.
        """
        if self.counter < self.count:
            self.counter += 1
            derived_message = "This is a Custom exception class"
            raise CustomError(derived_message)
        return True


def retry_if_result_none(result):
    return result is None


def retry_if_exception_of_type(retryable_types):
    def retry_if_exception_these_types(exception):
        print("Detected Exception of type: {0}".format(str(type(exception))))
        return isinstance(exception, retryable_types)
    return retry_if_exception_these_types


def current_time_ms():
    return int(round(time.time() * 1000))


@retry(wait_fixed=50, retry_on_result=retry_if_result_none)
def _retryable_test_with_wait(thing):
    return thing.go()


@retry(stop_max_attempt_number=3, retry_on_result=retry_if_result_none)
def _retryable_test_with_stop(thing):
    return thing.go()


@retry(retry_on_exception=retry_if_exception_of_type(IOError))
def _retryable_test_with_exception_type_io(thing):
    return thing.go()


@retry(retry_on_exception=retry_if_exception_of_type(IOError), wrap_exception=True)
def _retryable_test_with_exception_type_io_wrap(thing):
    return thing.go()


@retry(
    stop_max_attempt_number=3,
    retry_on_exception=retry_if_exception_of_type(IOError))
def _retryable_test_with_exception_type_io_attempt_limit(thing):
    return thing.go()


@retry(
    stop_max_attempt_number=3,
    retry_on_exception=retry_if_exception_of_type(IOError),
    wrap_exception=True)
def _retryable_test_with_exception_type_io_attempt_limit_wrap(thing):
    return thing.go()


@retry
def _retryable_default(thing):
    return thing.go()


@retry()
def _retryable_default_f(thing):
    return thing.go()


@retry(retry_on_exception=retry_if_exception_of_type(CustomError))
def _retryable_test_with_exception_type_custom(thing):
    return thing.go()


@retry(retry_on_exception=retry_if_exception_of_type(CustomError), wrap_exception=True)
def _retryable_test_with_exception_type_custom_wrap(thing):
    return thing.go()


@retry(
    stop_max_attempt_number=3,
    retry_on_exception=retry_if_exception_of_type(CustomError))
def _retryable_test_with_exception_type_custom_attempt_limit(thing):
    return thing.go()


@retry(
    stop_max_attempt_number=3,
    retry_on_exception=retry_if_exception_of_type(CustomError),
    wrap_exception=True)
def _retryable_test_with_exception_type_custom_attempt_limit_wrap(thing):
    return thing.go()


class TestDecoratorWrapper(unittest.TestCase):

    def test_with_wait(self):
        start = current_time_ms()
        result = _retryable_test_with_wait(NoneReturnUntilAfterCount(5))
        t = current_time_ms() - start
        self.assertTrue(t >= 250)
        self.assertTrue(result)

    def test_with_stop_on_return_value(self):
        try:
            _retryable_test_with_stop(NoneReturnUntilAfterCount(5))
            self.fail("Expected RetryError after 3 attempts")
        except RetryError as re:
            self.assertFalse(re.last_attempt.has_exception)
            self.assertEqual(3, re.last_attempt.attempt_number)
            self.assertTrue(re.last_attempt.value is None)
            print(re)

    def test_with_stop_on_exception(self):
        try:
            _retryable_test_with_stop(NoIOErrorAfterCount(5))
            self.fail("Expected IOError")
        except IOError as re:
            self.assertTrue(isinstance(re, IOError))
            print(re)

    def test_retry_if_exception_of_type(self):
        self.assertTrue(_retryable_test_with_exception_type_io(NoIOErrorAfterCount(5)))

        try:
            _retryable_test_with_exception_type_io(NoNameErrorAfterCount(5))
            self.fail("Expected NameError")
        except NameError as n:
            self.assertTrue(isinstance(n, NameError))
            print(n)

        try:
            _retryable_test_with_exception_type_io_attempt_limit_wrap(NoIOErrorAfterCount(5))
            self.fail("Expected RetryError")
        except RetryError as re:
            self.assertEqual(3, re.last_attempt.attempt_number)
            self.assertTrue(re.last_attempt.has_exception)
            self.assertTrue(re.last_attempt.value[0] is not None)
            self.assertTrue(isinstance(re.last_attempt.value[1], IOError))
            self.assertTrue(re.last_attempt.value[2] is not None)
            print(re)

        self.assertTrue(_retryable_test_with_exception_type_custom(NoCustomErrorAfterCount(5)))

        try:
            _retryable_test_with_exception_type_custom(NoNameErrorAfterCount(5))
            self.fail("Expected NameError")
        except NameError as n:
            self.assertTrue(isinstance(n, NameError))
            print(n)

        try:
            _retryable_test_with_exception_type_custom_attempt_limit_wrap(NoCustomErrorAfterCount(5))
            self.fail("Expected RetryError")
        except RetryError as re:
            self.assertEqual(3, re.last_attempt.attempt_number)
            self.assertTrue(re.last_attempt.has_exception)
            self.assertTrue(re.last_attempt.value[0] is not None)
            self.assertTrue(isinstance(re.last_attempt.value[1], CustomError))
            self.assertTrue(re.last_attempt.value[2] is not None)
            print(re)

    def test_wrapped_exception(self):

        # base exception cases
        self.assertTrue(_retryable_test_with_exception_type_io_wrap(NoIOErrorAfterCount(5)))

        try:
            _retryable_test_with_exception_type_io_wrap(NoNameErrorAfterCount(5))
            self.fail("Expected RetryError")
        except RetryError as re:
            self.assertTrue(isinstance(re.last_attempt.value[1], NameError))
            print(re)

        try:
            _retryable_test_with_exception_type_io_attempt_limit_wrap(NoIOErrorAfterCount(5))
            self.fail("Expected RetryError")
        except RetryError as re:
            self.assertEqual(3, re.last_attempt.attempt_number)
            self.assertTrue(re.last_attempt.has_exception)
            self.assertTrue(re.last_attempt.value[0] is not None)
            self.assertTrue(isinstance(re.last_attempt.value[1], IOError))
            self.assertTrue(re.last_attempt.value[2] is not None)
            print(re)

        # custom error cases
        self.assertTrue(_retryable_test_with_exception_type_custom_wrap(NoCustomErrorAfterCount(5)))

        try:
            _retryable_test_with_exception_type_custom_wrap(NoNameErrorAfterCount(5))
            self.fail("Expected RetryError")
        except RetryError as re:
            self.assertTrue(re.last_attempt.value[0] is not None)
            self.assertTrue(isinstance(re.last_attempt.value[1], NameError))
            self.assertTrue(re.last_attempt.value[2] is not None)
            print(re)

        try:
            _retryable_test_with_exception_type_custom_attempt_limit_wrap(NoCustomErrorAfterCount(5))
            self.fail("Expected RetryError")
        except RetryError as re:
            self.assertEqual(3, re.last_attempt.attempt_number)
            self.assertTrue(re.last_attempt.has_exception)
            self.assertTrue(re.last_attempt.value[0] is not None)
            self.assertTrue(isinstance(re.last_attempt.value[1], CustomError))
            self.assertTrue(re.last_attempt.value[2] is not None)

            self.assertTrue("This is a Custom exception class" in str(re.last_attempt.value[1]))
            print(re)

    def test_defaults(self):
        self.assertTrue(_retryable_default(NoNameErrorAfterCount(5)))
        self.assertTrue(_retryable_default_f(NoNameErrorAfterCount(5)))
        self.assertTrue(_retryable_default(NoCustomErrorAfterCount(5)))
        self.assertTrue(_retryable_default_f(NoCustomErrorAfterCount(5)))

if __name__ == '__main__':
    unittest.main()