utility_testcase.py 文件源码

def wait_until_true(self, lambda_expression, error_message='', timeout=TEST_TIMEOUT):
        """Evaluates lambda_expression once/1s until True or hits timeout."""
        assert hasattr(lambda_expression, '__call__'), 'lambda_expression is not callable: %s' % type(lambda_expression)
        assert hasattr(error_message, '__call__') or type(error_message) is str, 'error_message is not callable and not a str: %s' % type(error_message)
        assert type(timeout) == int, 'timeout is not an int: %s' % type(timeout)

        running_time = 0
        lambda_result = None
        wait_time = 0.01
        while not lambda_result and running_time < timeout:
            lambda_result = lambda_expression()
            logger.debug("%s evaluated to %s" % (inspect.getsource(lambda_expression), lambda_result))

            if not lambda_result:
                time.sleep(wait_time)
                wait_time = min(1, wait_time * 10)
                running_time += wait_time

        if hasattr(error_message, '__call__'):
            error_message = error_message()

        self.assertLess(running_time,
                        timeout,
                        'Timed out waiting for %s\nError Message %s' % (inspect.getsource(lambda_expression), error_message))