openstack
/
tempest
Code Issues Proposed changes

Add support of args and kwargs in call_until_true 

Now call_until_true doesn't accept args and kwargs,
so if want to call a callable with parameters, we have to
do like this(test_network_v6.py):
    srv1_v6_addr_assigned = functools.partial(
        guest_has_address, sshv4_1, ips_from_api_1['6'][i])
    self.assertTrue(test_utils.call_until_true(srv1_v6_addr_assigned,
                    CONF.validation.ping_timeout, 1))
So this is to add support of args and kwargs in call_until_true,
and to log the cost time when call_until_true returns True or
False for debugging.

Change-Id: Ib7a392f1a3999c2f2bd3cccaf2fd356cd7879950
This commit is contained in:
zhufl 2017-11-13 10:05:13 +08:00
parent 93a42fd79d
commit cf52e342e8
4 changed files with 45 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
releasenotes/notes/add-support-args-kwargs-in-call-until-true-a91k592h5a64exf7.yaml Normal file
View File

@ -0,0 +1,5 @@
---
features:
- Add support of args and kwargs when calling func in call_until_true,
also to log the cost time when call_until_true returns True or False
for debuggin.

13
tempest/lib/common/utils/test_utils.py
View File

@ -86,22 +86,29 @@ def call_and_ignore_notfound_exc(func, *args, **kwargs):
pass
def call_until_true(func, duration, sleep_for):
def call_until_true(func, duration, sleep_for, *args, **kwargs):
"""Call the given function until it returns True (and return True)
or until the specified duration (in seconds) elapses (and return False).
:param func: A zero argument callable that returns True on success.
:param func: A callable that returns True on success.
:param duration: The number of seconds for which to attempt a
successful call of the function.
:param sleep_for: The number of seconds to sleep after an unsuccessful
invocation of the function.
:param args: args that are passed to func.
:param kwargs: kwargs that are passed to func.
"""
now = time.time()
begin_time = now
timeout = now + duration
while now < timeout:
if func():
if func(*args, **kwargs):
LOG.debug("Call %s returns true in %f seconds",
getattr(func, '__name__'), time.time() - begin_time)
return True
time.sleep(sleep_for)
now = time.time()
LOG.debug("Call %s returns false in %f seconds",
getattr(func, '__name__'), duration)
return False

18
tempest/scenario/test_network_v6.py
View File

@ -12,8 +12,6 @@
# 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 functools
from tempest.common import utils
from tempest import config
from tempest.lib.common.utils import test_utils
@ -183,17 +181,13 @@ class TestGettingAddress(manager.NetworkScenarioTest):
for i in range(n_subnets6):
# v6 should be configured since the image supports it
# It can take time for ipv6 automatic address to get assigned
srv1_v6_addr_assigned = functools.partial(
guest_has_address, sshv4_1, ips_from_api_1['6'][i])
self.assertTrue(test_utils.call_until_true(guest_has_address,
CONF.validation.ping_timeout, 1,
sshv4_1, ips_from_api_1['6'][i]))
srv2_v6_addr_assigned = functools.partial(
guest_has_address, sshv4_2, ips_from_api_2['6'][i])
self.assertTrue(test_utils.call_until_true(srv1_v6_addr_assigned,
CONF.validation.ping_timeout, 1))
self.assertTrue(test_utils.call_until_true(srv2_v6_addr_assigned,
CONF.validation.ping_timeout, 1))
self.assertTrue(test_utils.call_until_true(guest_has_address,
CONF.validation.ping_timeout, 1,
sshv4_2, ips_from_api_2['6'][i]))
self.check_remote_connectivity(sshv4_1, ips_from_api_2['4'])
self.check_remote_connectivity(sshv4_2, ips_from_api_1['4'])

27
tempest/tests/lib/common/utils/test_test_utils.py
View File

@ -81,11 +81,13 @@ class TestTestUtils(base.TestCase):
@mock.patch('time.sleep')
@mock.patch('time.time')
def test_call_until_true_when_f_never_returns_true(self, m_time, m_sleep):
def set_value(bool_value):
return bool_value
timeout = 42 # The value doesn't matter as we mock time.time()
sleep = 60 # The value doesn't matter as we mock time.sleep()
m_time.side_effect = utils.generate_timeout_series(timeout)
self.assertEqual(
False, test_utils.call_until_true(lambda: False, timeout, sleep)
False, test_utils.call_until_true(set_value, timeout, sleep, False)
)
m_sleep.call_args_list = [mock.call(sleep)] * 2
m_time.call_args_list = [mock.call()] * 2
@ -93,11 +95,30 @@ class TestTestUtils(base.TestCase):
@mock.patch('time.sleep')
@mock.patch('time.time')
def test_call_until_true_when_f_returns_true(self, m_time, m_sleep):
def set_value(bool_value=False):
return bool_value
timeout = 42 # The value doesn't matter as we mock time.time()
sleep = 60 # The value doesn't matter as we mock time.sleep()
m_time.return_value = 0
self.assertEqual(
True, test_utils.call_until_true(lambda: True, timeout, sleep)
True, test_utils.call_until_true(set_value, timeout, sleep,
bool_value=True)
)
self.assertEqual(0, m_sleep.call_count)
self.assertEqual(1, m_time.call_count)
# when logging cost time we need to acquire current time.
self.assertEqual(2, m_time.call_count)
@mock.patch('time.sleep')
@mock.patch('time.time')
def test_call_until_true_when_f_returns_true_no_param(
self, m_time, m_sleep):
def set_value(bool_value=False):
return bool_value
timeout = 42 # The value doesn't matter as we mock time.time()
sleep = 60 # The value doesn't matter as we mock time.sleep()
m_time.side_effect = utils.generate_timeout_series(timeout)
self.assertEqual(
False, test_utils.call_until_true(set_value, timeout, sleep)
)
m_sleep.call_args_list = [mock.call(sleep)] * 2
m_time.call_args_list = [mock.call()] * 2