KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use set_resource_status method in localstack

Best Python code snippet using localstack_python

pacemaker_resource.py

Source:pacemaker_resource.py

...106    p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stderr=subprocess.PIPE)107    stdout, stderr = p.communicate(cib_xml)108    if p.returncode != 0:109        raise Exception(stderr)110def set_resource_status(root, enabled=True):111    name = root.get('id')112    meta = root.find('./meta_attributes')113    if meta is not None:114        node = meta.find("./nvpair[@name='target-role']")115    else:116        meta = append_meta_attribute_node(root, parent_id=name)117        node = None118    if enabled:119        if node is not None:120            if node.get('value') == 'Stopped':121                meta.remove(node)122                return True123    else:124        if node is None:125            attrib = {126                'id': "%s-meta_attributes-target-role" % name,127                'name': 'target-role',128                'value': 'Stopped'129            }130            ET.SubElement(meta, 'nvpair', attrib)131            return True132        else:133            if node.get('value') != 'Stopped':134                node.set('value', 'Stopped')135                return True136    return False137def option_str_to_dict(opts):138    if not isinstance(opts, basestring):139        return {}140    opts = opts.replace('\n', ' ')141    ret = {}142    for opt in shlex.split(opts):143        if '=' in opt:144            key, value = opt.split('=', 1)145            if value[0] == '"' and value[-1] == '"':146                value = value.strip('"')147            ret[key] = value148        else:149            ret['name'] = opt150    return ret151def append_nvpair_node(root, parent_id='', name='', value=''):152    node_id = "%s-%s" % (parent_id, name)153    attrib = {'id': node_id, 'name': name, 'value': value}154    node = ET.SubElement(root, "nvpair", attrib)155    return node156def append_meta_attribute_node(root, parent_id='', **kwargs):157    node_id = "%s-meta_attributes" % parent_id158    attrib = {'id': node_id}159    node = ET.SubElement(root, "meta_attributes", attrib)160    for name, value in kwargs.items():161        append_nvpair_node(node, parent_id=node_id, name=name, value=value)162    return node163def append_instance_attribute_node(root, parent_id='', **kwargs):164    if len(kwargs) == 0:165        return166    node_id = "%s-instance_attributes" % parent_id167    attrib = {'id': node_id}168    node = ET.SubElement(root, "instance_attributes", attrib)169    for name, value in kwargs.items():170        append_nvpair_node(node, parent_id=node_id, name=name, value=value)171    return node172def append_op_node(root, parent_id='', **kwargs):173    name = kwargs['name']174    interval = kwargs.get('interval', '0s')175    kwargs['interval'] = interval176    kwargs['id'] = "%s-%s-interval-%s" % (parent_id, name, interval)177    node = ET.SubElement(root, "op", kwargs)178    return node179def append_operations_node(root, parent_id='', op=[]):180    node = ET.SubElement(root, "operations")181    for o in op:182        o_dict = option_str_to_dict(o)183        append_op_node(node, parent_id=parent_id, **o_dict)184    return node185def append_resource_node(root, name='', type='', op=[],186                         meta='', params='', **kwargs):187    try:188        c, p, t = type.split(':')189        attrib = {'id': name, 'class': c, 'provider': p, 'type': t}190    except ValueError:191        c, t = type.split(':')192        attrib = {'id': name, 'class': c, 'type': t}193    node = ET.SubElement(root, "primitive", attrib)194    meta_dict = option_str_to_dict(meta)195    if meta_dict:196        append_meta_attribute_node(node, parent_id=name, **meta_dict)197    params_dict = option_str_to_dict(params)198    if params_dict:199        append_instance_attribute_node(node, parent_id=name, **params_dict)200    append_operations_node(node, parent_id=name, op=op)201    return node202def append_clone_node(root, clone=None, name='', **kwargs):203    meta_dict = option_str_to_dict(clone)204    node_id = meta_dict.pop('id', '%s-clone' % name)205    attrib = {'id': node_id}206    node = ET.SubElement(root, "clone", attrib)207    append_meta_attribute_node(node, parent_id=node_id, **meta_dict)208    resource_node = append_resource_node(node, parent_id=node_id, name=name,209                                         **kwargs)210    return node, resource_node211def append_master_node(root, master=None, name='', **kwargs):212    meta_dict = option_str_to_dict(master)213    node_id = meta_dict.pop('id', '%s-master' % name)214    attrib = {'id': node_id}215    node = ET.SubElement(root, "master", attrib)216    resource_node = append_resource_node(node, parent_id=node_id, name=name,217                                         **kwargs)218    append_meta_attribute_node(node, parent_id=node_id, **meta_dict)219    return node, resource_node220def has_difference(current, new):221    if current.tag != new.tag:222        return True223    if current.attrib != new.attrib:224        return True225    for n_child in list(new):226        child_id = n_child.get('id')227        if child_id:228            c_child = current.find("./*[@id='%s']" % child_id)229            if c_child is None or has_difference(c_child, n_child):230                return True231        else:232            for c_child in current.findall("./%s" % n_child.tag):233                if not has_difference(c_child, n_child):234                    break235            else:236                return True237    return False238def main():239    module = AnsibleModule(240        argument_spec=dict(241            name=dict(type='str', required=True),242            type=dict(type='str'),243            params=dict(type='str'),244            meta=dict(type='str'),245            op=dict(type='list', default=[]),246            clone=dict(type='str'),247            master=dict(type='str'),248            state=dict(type='str', default='present',249                       choices=['absent', 'present', 'enabled', 'disabled']),250            force=dict(type='bool', default=None),251        ),252        supports_check_mode=True,253    )254    name = module.params['name']255    type = module.params['type']256    params = module.params['params']257    meta = module.params['meta']258    op = module.params['op']259    clone = module.params['clone']260    master = module.params['master']261    state = module.params['state']262    force = module.params['force']263    check_only = module.check_mode264    result = dict(265        changed=False,266        name=name,267        type=type,268        params=params,269        meta=meta,270        op=op,271        clone=clone,272        master=master,273        state=state,274        force=force,275    )276    try:277        cib = get_cib_resources()278        node = cib.find('.//primitive[@id="%s"]' % name)279        parent_node = cib.find('.//primitive[@id="%s"]/..' % name)280        grand_node = cib.find('.//primitive[@id="%s"]/../..' % name)281        # Add/remove the resource as needed282        if state == 'absent':283            if node is not None:284                if not check_only:285                    if parent_node.tag in ['clone', 'master']:286                        grand_node.remove(parent_node)287                    else:288                        parent_node.remove(node)289                    set_cib_resources(cib)290                result['changed'] = True291            # Report the success result and exit292            module.exit_json(**result)293        else:294            if node is None:295                if master is not None:296                    parent_node, node = append_master_node(cib,297                                                           **module.params)298                elif clone is not None:299                    parent_node, node = append_clone_node(cib, **module.params)300                else:301                    node = append_resource_node(cib, **module.params)302                result['changed'] = True303            else:304                if master is not None:305                    if parent_node.tag in ['clone', 'master']:306                        new_parent_node, new_node = append_master_node(307                            grand_node, **module.params)308                        if force or has_difference(parent_node,309                                                   new_parent_node):310                            grand_node.remove(parent_node)311                            node = new_node312                            result['changed'] = True313                        else:314                            grand_node.remove(new_parent_node)315                    else:316                        parent_node.remove(node)317                        append_master_node(parent_node, **module.params)318                        result['changed'] = True319                elif clone is not None:320                    if parent_node.tag in ['clone', 'master']:321                        new_parent_node, new_node = append_clone_node(322                            grand_node, **module.params)323                        if force or has_difference(parent_node,324                                                   new_parent_node):325                            grand_node.remove(parent_node)326                            node = new_node327                            result['changed'] = True328                        else:329                            grand_node.remove(new_parent_node)330                    else:331                        parent_node.remove(node)332                        parent_node, node = append_clone_node(parent_node,333                                                              **module.params)334                        result['changed'] = True335                else:336                    if parent_node.tag in ['clone', 'master']:337                        grand_node.remove(parent_node)338                        append_resource_node(grand_node, **module.params)339                        result['changed'] = True340                    else:341                        new_node = append_resource_node(parent_node,342                                                        **module.params)343                        if force or has_difference(node, new_node):344                            parent_node.remove(node)345                            node = new_node346                            result['changed'] = True347                        else:348                            parent_node.remove(new_node)349        # Start/stop the resource as needed350        if state == "enabled":351            if set_resource_status(node, enabled=True):352                result['changed'] = True353        elif state == "disabled":354            if set_resource_status(node, enabled=False):355                result['changed'] = True356        # Apply the modified CIB as needed357        if result['changed'] and not check_only:358            set_cib_resources(cib)359        # Report the success result and exit360        module.exit_json(**result)361    except Exception as e:362        # Report the failure result and exit363        module.fail_json(msg=to_native(e),364                         exception=traceback.format_exc(),365                         **result)366if __name__ == '__main__':...

resource.py

Source:resource.py

...78        s.delete(resource)79        s.commit()80    show_resources(message)81@respond_to(r'^status\s+(\S+)$')82def unset_resource_status(message, name):83    """ãªã½ã¼ã¹ã®è¨å®ãåæå¤ã«æ»ã84    :param message: slackbot.dispatcher.Message85    :param str name: ãªã½ã¼ã¹å86    """87    if name in COMMANDS:88        return89    channel_id = message.body["channel"]90    s = Session()91    resource = s.query(Resource).filter(92        Resource.channel_id == channel_id,93        Resource.name == name,94    ).one_or_none()95    if resource:96        resource.status = None97        s.commit()98    show_resources(message)99@respond_to(r'^status\s+(\S+)\s+(\S+)$')100def set_resource_status(message, name, value):101    """ãªã½ã¼ã¹ã®è¨å®102    :param message: slackbot.dispatcher.Message103    :param str name: ãªã½ã¼ã¹å104    :param str value: ç¶æ105    """106    if name in COMMANDS:107        return108    channel_id = message.body["channel"]109    s = Session()110    resource = s.query(Resource).filter(111        Resource.channel_id == channel_id,112        Resource.name == name,113    ).one_or_none()114    if resource:...

qrm_urls.py

Source:qrm_urls.py

1URL_API_VERSION = '/v1'2URL_POST_NEW_REQUEST = f'/new_request{URL_API_VERSION}'3URL_GET_TOKEN_STATUS = f'/get_token_status{URL_API_VERSION}'4URL_POST_CANCEL_TOKEN = f'/cancel_token{URL_API_VERSION}'5URL_GET_ROOT = '/'6URL_GET_UPTIME = f'/uptime'7URL_GET_IS_SERVER_UP = '/is_server_up'8MGMT_STATUS_API = '/status'9SET_SERVER_STATUS = '/set_server_status'10REMOVE_RESOURCES = '/remove_resources'11ADD_RESOURCES = '/add_resources'12SET_RESOURCE_STATUS = '/set_resource_status'13ADD_TAG_TO_RESOURCE = '/add_tag_to_resource'...

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.