Merge "Make sure looping calls are properly mocked"

This commit is contained in:
Zuul 2019-02-20 18:25:30 +00:00 committed by Gerrit Code Review
commit fb5ebc2f9e
2 changed files with 9 additions and 27 deletions

View File

@ -14,36 +14,11 @@
import mock
import os
from oslo_service import loopingcall
from os_brick import exception
from os_brick.initiator.connectors import aoe
from os_brick.tests.initiator import test_connector
class FakeFixedIntervalLoopingCall(object):
def __init__(self, f=None, *args, **kw):
self.args = args
self.kw = kw
self.f = f
self._stop = False
def stop(self):
self._stop = True
def wait(self):
return self
def start(self, interval, initial_delay=None):
while not self._stop:
try:
self.f(*self.args, **self.kw)
except loopingcall.LoopingCallDone:
return self
except Exception:
raise
class AoEConnectorTestCase(test_connector.ConnectorTestCase):
"""Test cases for AoE initiator class."""
@ -52,8 +27,6 @@ class AoEConnectorTestCase(test_connector.ConnectorTestCase):
self.connector = aoe.AoEConnector('sudo')
self.connection_properties = {'target_shelf': 'fake_shelf',
'target_lun': 'fake_lun'}
self.mock_object(loopingcall, 'FixedIntervalLoopingCall',
FakeFixedIntervalLoopingCall)
def test_get_search_path(self):
expected = "/dev/etherd"

View File

@ -17,6 +17,7 @@ import sys
import mock
from oslo_concurrency import processutils as putils
from oslo_service import loopingcall
from os_brick import exception
from os_brick.initiator import connector
@ -32,6 +33,12 @@ MY_IP = '10.0.0.1'
FAKE_SCSI_WWN = '1234567890'
class ZeroIntervalLoopingCall(loopingcall.FixedIntervalLoopingCall):
def start(self, interval, initial_delay=None, stop_on_exception=True):
return super(ZeroIntervalLoopingCall, self).start(
0, 0, stop_on_exception)
class ConnectorUtilsTestCase(test_base.TestCase):
@mock.patch.object(nvme.NVMeConnector, '_get_system_uuid',
@ -126,6 +133,8 @@ class ConnectorTestCase(test_base.TestCase):
def setUp(self):
super(ConnectorTestCase, self).setUp()
self.cmds = []
self.mock_object(loopingcall, 'FixedIntervalLoopingCall',
ZeroIntervalLoopingCall)
def fake_execute(self, *cmd, **kwargs):
self.cmds.append(" ".join(cmd))