set(dir(_io.BytesIO))))
else:
file_spec = file
+
+
+def generate_timeout_series(timeout):
+ """Generate a series of times that exceeds the given timeout.
+
+ Yields a series of fake time.time() floating point numbers
+ such that the difference between each pair in the series just
+ exceeds the timeout value that is passed in. Useful for
+ mocking time.time() in methods that otherwise wait for timeout
+ seconds.
+ """
+ iteration = 0
+ while True:
+ iteration += 1
+ yield (iteration * timeout) + iteration
import copy
import mock
import six
+import time
from cinder import exception
from cinder.tests.unit import fake_volume
+from cinder.tests.unit import utils as utils
from cinder.tests.unit.volume.drivers import disco
self.test_pending = True
self.test_create_cloned_volume()
- def test_create_cloned_volume_timeout(self):
+ @mock.patch.object(time, 'time')
+ def test_create_cloned_volume_timeout(self, mock_time):
"""Clone request timeout."""
- self.driver.configuration.clone_check_timeout = 3
+ timeout = 3
+ mock_time.side_effect = utils.generate_timeout_series(timeout)
+ self.driver.configuration.clone_check_timeout = timeout
self.response = self.FAKE_SOAP_RESPONSE['standard']['success']
self.response_detail = (
self.FAKE_SOAP_RESPONSE['clone_detail']['pending'])
import copy
import mock
+import time
from cinder import db
from cinder import exception
from cinder.tests.unit import fake_snapshot
+from cinder.tests.unit import utils as utils
from cinder.tests.unit.volume.drivers import disco
self.test_pending = True
self.test_create_snapshot()
- def test_create_snapshot_timeout(self):
+ @mock.patch.object(time, 'time')
+ def test_create_snapshot_timeout(self, mock_time):
"""Snapshot request timeout."""
- self.driver.configuration.snapshot_check_timeout = 3
+ timeout = 3
+ mock_time.side_effect = utils.generate_timeout_series(timeout)
+ self.driver.configuration.snapshot_check_timeout = timeout
self.response = self.FAKE_SOAP_RESPONSE['standard']['success']
self.response_detail = (
self.FAKE_SOAP_RESPONSE['snapshot_detail']['pending'])
import copy
import mock
+import time
from cinder import exception
from cinder.tests.unit import fake_snapshot
+from cinder.tests.unit import utils as utils
from cinder.tests.unit.volume.drivers import disco
self.test_pending = True
self.test_create_volume_from_snapshot()
- def test_create_volume_from_snapshot_timeout(self):
+ @mock.patch.object(time, 'time')
+ def test_create_volume_from_snapshot_timeout(self, mock_time):
"""Create volume from snapshot task timeout."""
- self.driver.configuration.restore_check_timeout = 3
+ timeout = 3
+ mock_time.side_effect = utils.generate_timeout_series(timeout)
+ self.driver.configuration.restore_check_timeout = timeout
self.response = self.FAKE_SOAP_RESPONSE['standard']['success']
self.response_detail = (
self.FAKE_SOAP_RESPONSE['restore_detail']['pending'])