Synnefo » snf-ganeti

#!/usr/bin/python

#

# Copyright (C) 2011 Google Inc.

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful, but

# WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

# General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA

# 02110-1301, USA.

"""Script for testing ganeti.utils.retry"""

import unittest

from ganeti import constants

from ganeti import errors

from ganeti import utils

import testutils

class TestRetry(testutils.GanetiTestCase):

  def setUp(self):

    testutils.GanetiTestCase.setUp(self)

    self.retries = 0

    self.called = 0

    self.time = 1379601882.0

    self.time_for_time_fn = 0

    self.time_for_retry_and_succeed = 0

  def _time_fn(self):

    self.time += self.time_for_time_fn

    return self.time

  def _wait_fn(self, delay):

    self.time += delay

  @staticmethod

  def _RaiseRetryAgain():

    raise utils.RetryAgain()

  @staticmethod

  def _RaiseRetryAgainWithArg(args):

    raise utils.RetryAgain(*args)

  def _WrongNestedLoop(self):

    return utils.Retry(self._RaiseRetryAgain, 0.01, 0.02)

  def _RetryAndSucceed(self, retries):

    self.time += self.time_for_retry_and_succeed

    if self.retries < retries:

      self.retries += 1

      raise utils.RetryAgain()

    else:

      return True

  def _SimpleRetryAndSucceed(self, retries):

    self.called += 1

    if self.retries < retries:

      self.retries += 1

      return False

    else:

      return True

  def testRaiseTimeout(self):

    self.failUnlessRaises(utils.RetryTimeout, utils.Retry,

                          self._RaiseRetryAgain, 0.01, 0.02,

                          wait_fn = self._wait_fn, _time_fn = self._time_fn)

    self.failUnlessRaises(utils.RetryTimeout, utils.Retry,

                          self._RetryAndSucceed, 0.01, 0, args=[1],

                          wait_fn = self._wait_fn, _time_fn = self._time_fn)

    self.failUnlessEqual(self.retries, 1)

  def testComplete(self):

    self.failUnlessEqual(utils.Retry(lambda: True, 0, 1,

                                     wait_fn = self._wait_fn,

                                     _time_fn = self._time_fn),

                         True)

    self.failUnlessEqual(utils.Retry(self._RetryAndSucceed, 0, 1, args=[2],

                                     wait_fn = self._wait_fn,

                                     _time_fn = self._time_fn),

                         True)

    self.failUnlessEqual(self.retries, 2)

  def testCompleteNontrivialTimes(self):

    self.time_for_time_fn = 0.01

    self.time_for_retry_and_succeed = 0.1

    self.failUnlessEqual(utils.Retry(self._RetryAndSucceed, 0, 1, args=[2],

                                     wait_fn = self._wait_fn,

                                     _time_fn = self._time_fn),

                         True)

    self.failUnlessEqual(self.retries, 2)

  def testNestedLoop(self):

    try:

      self.failUnlessRaises(errors.ProgrammerError, utils.Retry,

                            self._WrongNestedLoop, 0, 1,

                            wait_fn = self._wait_fn, _time_fn = self._time_fn)

    except utils.RetryTimeout:

      self.fail("Didn't detect inner loop's exception")

  def testTimeoutArgument(self):

    retry_arg="my_important_debugging_message"

    try:

      utils.Retry(self._RaiseRetryAgainWithArg, 0.01, 0.02, args=[[retry_arg]],

                  wait_fn = self._wait_fn, _time_fn = self._time_fn)

    except utils.RetryTimeout, err:

      self.failUnlessEqual(err.args, (retry_arg, ))

    else:

      self.fail("Expected timeout didn't happen")

  def testTimeout(self):

    self.time_for_time_fn = 0.01

    self.time_for_retry_and_succeed = 10

    try:

      utils.Retry(self._RetryAndSucceed, 1, 18, args=[2],

                  wait_fn = self._wait_fn, _time_fn = self._time_fn)

    except utils.RetryTimeout, err:

      self.failUnlessEqual(err.args, ())

    else:

      self.fail("Expected timeout didn't happen")

  def testNoTimeout(self):

    self.time_for_time_fn = 0.01

    self.time_for_retry_and_succeed = 8

    self.failUnlessEqual(

      utils.Retry(self._RetryAndSucceed, 1, 18, args=[2],

                  wait_fn = self._wait_fn, _time_fn = self._time_fn),

      True)

  def testRaiseInnerWithExc(self):

    retry_arg="my_important_debugging_message"

    try:

      try:

        utils.Retry(self._RaiseRetryAgainWithArg, 0.01, 0.02,

                    args=[[errors.GenericError(retry_arg, retry_arg)]],

                    wait_fn = self._wait_fn, _time_fn = self._time_fn)

      except utils.RetryTimeout, err:

        err.RaiseInner()

      else:

        self.fail("Expected timeout didn't happen")

    except errors.GenericError, err:

      self.failUnlessEqual(err.args, (retry_arg, retry_arg))

    else:

      self.fail("Expected GenericError didn't happen")

  def testRaiseInnerWithMsg(self):

    retry_arg="my_important_debugging_message"

    try:

      try:

        utils.Retry(self._RaiseRetryAgainWithArg, 0.01, 0.02,

                    args=[[retry_arg, retry_arg]],

                    wait_fn = self._wait_fn, _time_fn = self._time_fn)

      except utils.RetryTimeout, err:

        err.RaiseInner()

      else:

        self.fail("Expected timeout didn't happen")

    except utils.RetryTimeout, err:

      self.failUnlessEqual(err.args, (retry_arg, retry_arg))

    else:

      self.fail("Expected RetryTimeout didn't happen")

  def testSimpleRetry(self):

    self.assertFalse(utils.SimpleRetry(True, lambda: False, 0.01, 0.02,

                                       wait_fn = self._wait_fn,

                                       _time_fn = self._time_fn))

    self.assertFalse(utils.SimpleRetry(lambda x: x, lambda: False, 0.01, 0.02,

                                       wait_fn = self._wait_fn,

                                       _time_fn = self._time_fn))

    self.assertTrue(utils.SimpleRetry(True, lambda: True, 0, 1,

                                      wait_fn = self._wait_fn,

                                      _time_fn = self._time_fn))

    self.assertTrue(utils.SimpleRetry(lambda x: x, lambda: True, 0, 1,

                                      wait_fn = self._wait_fn,

                                      _time_fn = self._time_fn))

    self.assertTrue(utils.SimpleRetry(True, self._SimpleRetryAndSucceed, 0, 1,

                                      args=[1], wait_fn = self._wait_fn,

                                      _time_fn = self._time_fn))

    self.assertEqual(self.retries, 1)

    self.assertEqual(self.called, 2)

    self.called = self.retries = 0

    self.assertTrue(utils.SimpleRetry(True, self._SimpleRetryAndSucceed, 0, 1,

                                      args=[2], wait_fn = self._wait_fn,

                                      _time_fn = self._time_fn))

    self.assertEqual(self.called, 3)

if __name__ == "__main__":

  testutils.GanetiTestProgram()