Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use is_resource_active method in tempest

Best Python code snippet using tempest_python

rest_client.py

Source:rest_client.py

...828                                  resource still hasn't been active829        """830        start_time = int(time.time())831        while True:832            if self.is_resource_active(id):833                return834            if int(time.time()) - start_time >= self.build_timeout:835                message = ('Failed to reach active state %(resource_type)s '836                           '%(id)s within the required time (%(timeout)s s).' %837                           {'resource_type': self.resource_type, 'id': id,838                            'timeout': self.build_timeout})839                caller = test_utils.find_test_caller()840                if caller:841                    message = '(%s) %s' % (caller, message)842                raise exceptions.TimeoutException(message)843            time.sleep(self.build_interval)844    def is_resource_deleted(self, id):845        """Subclasses override with specific deletion detection."""846        message = ('"%s" does not implement is_resource_deleted'847                   % self.__class__.__name__)848        raise NotImplementedError(message)849    def is_resource_active(self, id):850        """Subclasses override with specific active detection."""851        message = ('"%s" does not implement is_resource_active'852                   % self.__class__.__name__)853        raise NotImplementedError(message)854    @property855    def resource_type(self):856        """Returns the primary type of resource this client works with."""857        return 'resource'858    @classmethod859    def validate_response(cls, schema, resp, body):860        # Only check the response if the status code is a success code861        # TODO(cyeoh): Eventually we should be able to verify that a failure862        # code if it exists is something that we expect. This is explicitly863        # declared in the V3 API and so we should be able to export this in...

pacemaker_is_active.py

Source:pacemaker_is_active.py

...197    "Representation of a primitive resource."198    get_type = 'primitive'199    def expected_count(self):200        return 1201def is_resource_active(mod):202    """Return success if a resource active, failure otherwise.203    Takes the resource name as an argument and does the following:204    a) master/slave resources205    Returns active only if the needed number of masters is set206    e.g. galera needs to be master on all nodes where galera is207    supposed to run (that is == to the number of controllers in208    pre-composable ha and the number of nodes with galera-role=true209    properties set in composable ha) redis will need to have master on210    only one node.211    b) cloned resources212    Returns active if the resource is started on the needed nodes213    e.g. same as master/slave resources the needed number of nodes is214    equal to the cluster nodes in pre-composable and to the215    haproxy-role property count in composable ha.216    c) primitive resources returns active217    If the resource is started on one node e.g. A/P resources like218    cinder-volume, VIPs.219    """220    max_tries = int(mod.params["max_wait"])221    resource_name = mod.params["resource"]222    current_try = 0223    resource = Resource(mod, resource_name).from_type()224    if resource.get_type is None:225        return resource.fail("Resource '{0}' doesn't exist in the cib.".format(226            resource.name227        ))228    resource_expected_count = resource.expected_count()229    while resource_expected_count != resource.current_count():230        if current_try >= max_tries-1:231            return resource.fail(232                "Max wait time of {0} seconds reached waiting for {1}".format(233                    max_tries, resource.name234                ))235        sleep(1)236        current_try += 1237    return resource.success("{0} resource {1} is active".format(resource.get_type,238                                                                resource.name))239def main():240    "Main function called by Ansible."241    mod = AnsibleModule(242        argument_spec=dict(243            resource=dict(type='str',required=True),244            max_wait=dict(type='int',default=5),  # in seconds245        )246    )247    return is_resource_active(mod)248if __name__ == '__main__':...

test_pacemaker_is_active.py

Source:test_pacemaker_is_active.py

...123        )124        has_type.return_value = pacemaker_is_active.Clone(mod, 'haproxy')125        clone_resource_expected_count.return_value = 3126        clone_resource_current_count.return_value = 3127        pacemaker_is_active.is_resource_active(mod)128        self.assertEqual(0, mod.fail_json.call_count)129        self.assertEqual(1, mod.exit_json.call_count)130class TestMasterResource(unittest.TestCase):131    @patch('modules.pacemaker_is_active.Master.current_count')132    @patch('modules.pacemaker_is_active.Master.expected_count')133    @patch('modules.pacemaker_is_active.Resource.from_type')134    def test__master_resource__happy(self,135                                     has_type,136                                     master_resource_expected_count,137                                     master_resource_current_count):138        mod_cls = create_autospec(AnsibleModule)139        mod = mod_cls.return_value140        mod.params = dict(141            resource="galera",142            max_wait="5"143        )144        has_type.return_value = pacemaker_is_active.Master(mod, 'galera')145        master_resource_expected_count.return_value = 3146        master_resource_current_count.return_value = 3147        pacemaker_is_active.is_resource_active(mod)148        self.assertEqual(0, mod.fail_json.call_count)149        self.assertEqual(1, mod.exit_json.call_count)150class TestPrimitiveResource(unittest.TestCase):151    @patch('modules.pacemaker_is_active.Primitive.current_count')152    @patch('modules.pacemaker_is_active.Resource.from_type')153    def test__primitive_resource__happy(self,154                                        has_type,155                                        primitive_resource_current_count):156        mod_cls = create_autospec(AnsibleModule)157        mod = mod_cls.return_value158        mod.params = dict(159            resource="openstack-cinder-volume",160            max_wait="5"161        )162        has_type.return_value = pacemaker_is_active.Primitive(163            mod,164            "openstack-cinder-volume")165        primitive_resource_current_count.return_value = 1166        pacemaker_is_active.is_resource_active(mod)167        self.assertEqual(0, mod.fail_json.call_count)168        self.assertEqual(1, mod.exit_json.call_count)169class TestTimeout(unittest.TestCase):170    @patch('modules.pacemaker_is_active.Primitive.current_count')171    @patch('modules.pacemaker_is_active.Primitive.expected_count')172    @patch('modules.pacemaker_is_active.Resource.from_type')173    def test__primitive_resource__happy(self,174                                        has_type,175                                        primitive_resource_expected_count,176                                        primitive_resource_current_count):177        mod_cls = create_autospec(AnsibleModule)178        mod = mod_cls.return_value179        mod.params = dict(180            resource="openstack-cinder-volume",181            max_wait="3"182        )183        has_type.return_value = pacemaker_is_active.Primitive(184            mod,185            'openstack-cinder-volume')186        primitive_resource_expected_count.side_effect = [1, 1, 1]187        primitive_resource_current_count.side_effect = [0, 0, 0]188        pacemaker_is_active.is_resource_active(mod)189        self.assertEqual(1, mod.fail_json.call_count)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.