KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use get_resource_name method in localstack
Best Python code snippet using localstack_python
nsxt_base_resource.py
Source:nsxt_base_resource.py 
...67 resource_params = self.module.params68 self.resource_params = resource_params69 self._state = self.get_attribute('state', resource_params)70 if not (hasattr(self, 'id') and self.id):71 if self.get_resource_name() in BASE_RESOURCES:72 self.id = self._get_id_using_attr_name(73 None, resource_params,74 self.get_resource_base_url(self.baseline_args),75 self.get_spec_identifier())76 else:77 self.id = self._get_id_using_attr_name(78 None, resource_params,79 self.get_resource_base_url(self._parent_info),80 self.get_spec_identifier())81 if self.id is None:82 return83 # Extract the resource params from module84 self.nsx_resource_params = self._extract_nsx_resource_params(85 resource_params)86 # parent_info is passed to subresources of a resource automatically87 if not hasattr(self, "_parent_info"):88 self._parent_info = {}89 self.update_parent_info(self._parent_info)90 try:91 # get existing resource schema92 _, self.existing_resource = self._send_request_to_API(93 "/" + self.id, ignore_error=False,94 accepted_error_codes=set([404]))95 # As Policy API's PATCH requires all attributes to be filled,96 # we fill the missing resource params (the params not specified)97 # by user using the existing params98 self._fill_missing_resource_params(99 self.existing_resource, self.nsx_resource_params)100 except Exception as err:101 # the resource does not exist currently on the manager102 self.existing_resource = None103 self._achieve_state(resource_params, successful_resource_exec_logs)104 @classmethod105 def get_spec_identifier(cls):106 # Can be overriden in the subclass to provide different107 # unique_arg_identifier. It is used to infer which args belong to which108 # subresource.109 # By default, class name is used for subresources.110 return cls.get_resource_name()111 def get_state(self):112 return self._state113 def get_parent_info(self):114 return self._parent_info115 @staticmethod116 @abstractmethod117 def get_resource_base_url(parent_info):118 # Must be overridden by the subclass119 raise NotImplementedError120 @staticmethod121 @abstractmethod122 def get_resource_spec():123 # Must be overridden by the subclass124 raise NotImplementedError125 @classmethod126 def get_resource_name(cls):127 return cls.__name__128 def create_or_update_subresource_first(self):129 # return True if subresource should be created/updated before parent130 # resource131 return self.resource_params.get(132 "create_or_update_subresource_first", False)133 def delete_subresource_first(self):134 # return True if subresource should be deleted before parent resource135 return self.resource_params.get("delete_subresource_first", True)136 def achieve_subresource_state_if_del_parent(self):137 # return True if this resource is to be realized with its own specified138 # state irrespective of the state of its parent resource.139 return self.resource_params.get(140 "achieve_subresource_state_if_del_parent", False)141 def do_wait_till_create(self):142 # By default, we do not wait for the parent resource to be created or143 # updated before its subresource is to be realized.144 return self.resource_params.get("do_wait_till_create", False)145 @staticmethod146 def get_resource_update_priority():147 # this priority can be used to create/delete subresources148 # at the same level in a particular order.149 # by default, it returns 1 so the resources are created/updated/150 # deleted in a fixed but random order.151 # should be overloaded in subclass to specify its priority.152 # for creation or update, we iterate in descending order.153 # for deletion, we iterate in ascending order.154 return 1155 def set_arg_spec(self, arg_spec):156 self._arg_spec = arg_spec157 def set_ansible_module(self, ansible_module):158 self.module = ansible_module159 def set_parent_info(self, parent_info):160 self._parent_info = parent_info161 def achieve_subresource_state(162 self, resource_params, successful_resource_exec_logs):163 """164 Achieve the state of each sub-resource.165 """166 for sub_resource_class in self._get_sub_resources_class_of(167 self.resource_class):168 if sub_resource_class.allows_multiple_resource_spec():169 children_resource_spec = (resource_params.get(170 sub_resource_class.get_spec_identifier()) or [])171 else:172 children_resource_spec = ([resource_params.get(173 sub_resource_class.get_spec_identifier())] or [])174 # Update the parent pointer175 my_parent = self._parent_info.get('_parent', '')176 self._update_parent_info()177 for resource_param_spec in children_resource_spec:178 if resource_param_spec is not None:179 sub_resource = sub_resource_class()180 sub_resource.set_arg_spec(self._arg_spec)181 sub_resource.set_ansible_module(self.module)182 sub_resource.set_parent_info(self._parent_info)183 sub_resource.realize(184 successful_resource_exec_logs=(185 successful_resource_exec_logs),186 resource_params=resource_param_spec)187 # Restore the parent pointer188 self._parent_info['_parent'] = my_parent189 def update_resource_params(self, nsx_resource_params):190 # Can be used to updates the params of resource before making191 # the API call.192 # Should be overridden in the subclass if needed193 pass194 def check_for_update(self, existing_params, resource_params):195 """196 resource_params: dict197 existing_params: dict198 Compares the existing_params with resource_params and returns199 True if they are different. At a base level, it traverses the200 params and matches one-to-one. If the value to be matched is a201 - dict, it traverses that also.202 - list, it merely compares the order.203 Can be overriden in the subclass for specific custom checking.204 Returns true if the params differ205 """206 if not existing_params:207 return False208 for k, v in resource_params.items():209 if k not in existing_params:210 return True211 elif type(v).__name__ == 'dict':212 if self.check_for_update(existing_params[k], v):213 return True214 elif v != existing_params[k]:215 def compare_lists(list1, list2):216 # Returns True if list1 and list2 differ217 try:218 # If the lists can be converted into sets, do so and219 # compare lists as sets.220 set1 = set(list1)221 set2 = set(list2)222 return set1 != set2223 except Exception:224 return True225 if type(v).__name__ == 'list':226 if compare_lists(v, existing_params[k]):227 return True228 continue229 return True230 return False231 def update_parent_info(self, parent_info):232 # Override this and fill in self._parent_info if that is to be passed233 # to the sub-resource234 # By default, parent's id is passed235 parent_info[self.get_spec_identifier() + "_id"] = self.id236 def get_attribute(self, attribute, resource_params):237 """238 attribute: String239 resource_params: Parameters of the resource240 """241 if (attribute == "state" and242 self.get_resource_name() not in BASE_RESOURCES):243 # if parent has absent state, subresources should have absent244 # state if . So, irrespective of what user specifies, if parent245 # is to be deleted, the child resources will be deleted.246 # override achieve_subresource_state_if_del_parent247 # in resource class to change this behavior248 if (self._parent_info["_parent"].get_state() == "absent" and249 not self.achieve_subresource_state_if_del_parent()):250 return "absent"251 return resource_params.get(252 attribute, self.INCORRECT_ARGUMENT_NAME_VALUE)253 def set_baseline_args(self, baseline_arg_names):254 # Can be overriden in subclass255 self.baseline_args = {}256 for baseline_arg_name in baseline_arg_names:257 self.baseline_args[baseline_arg_name] = self.module.params[258 baseline_arg_name]259 def do_resource_params_have_attr_with_id_or_display_name(self, attr):260 if (attr + "_id" in self.nsx_resource_params or261 attr + "_display_name" in self.nsx_resource_params):262 return True263 return False264 def get_id_using_attr_name_else_fail(self, attr_name, params,265 resource_base_url, resource_type,266 ignore_not_found_error=True):267 resource_id = self._get_id_using_attr_name(268 attr_name, params, resource_base_url, resource_type,269 ignore_not_found_error)270 if resource_id is not None:271 return resource_id272 # Incorrect usage of Ansible Module273 self.module.fail_json(msg="Please specify either {} id or display_name"274 " for the resource {}".format(275 attr_name, str(resource_type)))276 def exit_with_failure(self, msg, **kwargs):277 self.module.fail_json(msg=msg, **kwargs)278 def skip_delete(self):279 """280 Override in subclass if this resource is skipped to be deleted.281 Note that the children of this resource will still be deleted unless282 they override this method as well.283 """284 return False285 @classmethod286 def is_required_in_spec(cls):287 """288 Override in subclass if this resource is optional to be specified289 in the ansible playbook.290 """291 return False292 @classmethod293 def allows_multiple_resource_spec(cls):294 """295 Override in the resource class definition with False if only one296 resource can be associated with the parent. By default, we accept297 multiple298 """299 return True300 def _get_id_using_attr_name(self, attr_name, params,301 resource_base_url, resource_type,302 ignore_not_found_error=True):303 # Pass attr_name '' or None to infer base resource's ID304 id_identifier = 'id'305 display_name_identifier = 'display_name'306 if attr_name:307 id_identifier = attr_name + "_id"308 display_name_identifier = attr_name + "_display_name"309 if id_identifier in params and params[id_identifier]:310 return params.pop(id_identifier)311 if (display_name_identifier in params and312 params[display_name_identifier]):313 resource_display_name = params.pop(display_name_identifier)314 # Use display_name as ID if ID is not specified.315 return (self.get_id_from_display_name(316 resource_base_url, resource_display_name, resource_type,317 ignore_not_found_error) or resource_display_name)318 def get_id_from_display_name(self, resource_base_url,319 resource_display_name,320 resource_type, ignore_not_found_error=True):321 try:322 # Get the id from the Manager323 (_, resp) = self._send_request_to_API(324 resource_base_url=resource_base_url)325 matched_resource = None326 for resource in resp['results']:327 if (resource.__contains__('display_name') and328 resource['display_name'] == resource_display_name):329 if matched_resource is None:330 matched_resource = resource331 else:332 # Multiple resources with same display_name!333 # Ask the user to specify ID instead.334 self.module.fail_json(335 msg="Multiple {} found with display_name {}. "336 "Please specify the resource using id in "337 "the playbook.".format(resource_type,338 resource_display_name))339 if matched_resource is not None:340 return matched_resource['id']341 else:342 if ignore_not_found_error:343 return None344 else:345 # No resource found with this display_name346 self.module.fail_json(347 msg="No {} found with display_name {} for the "348 "specified configuration.".format(349 resource_type, resource_display_name))350 except Exception as e:351 # Manager replied with invalid URL. It means that the resource352 # does not exist on the Manager. So, return the display_name353 return resource_display_name354 def _update_parent_info(self):355 # This update is always performed and should not be overriden by the356 # subresource's class357 self._parent_info["_parent"] = self358 def _make_ansible_arg_spec(self, supports_check_mode=True):359 """360 We read the arg_spec of all the resources associated that361 are associated with this resource and create one complete362 arg_spec.363 """364 if self.get_resource_name() in BASE_RESOURCES:365 self._arg_spec = {}366 # Update it with VMware arg spec367 self._arg_spec.update(368 PolicyCommunicator.get_vmware_argument_spec())369 # ... then update it with top most resource spec ...370 self._update_arg_spec_with_resource(371 self.resource_class, self._arg_spec)372 # Update with all sub-resources arg spec373 for sub_resources_class in self._get_sub_resources_class_of(374 self.resource_class):375 self._update_arg_spec_with_all_resources(376 sub_resources_class, self._arg_spec)377 def _update_arg_spec_with_resource(self, resource_class, arg_spec):378 # updates _arg_spec with resource_class's arg_spec379 resource_arg_spec = self._get_base_arg_spec_of_resource()380 resource_arg_spec.update(self._get_base_arg_spec_of_nsx_resource())381 resource_arg_spec.update(resource_class.get_resource_spec())382 if resource_class.__name__ not in BASE_RESOURCES:383 arg_spec.update(384 {385 resource_class.get_spec_identifier(): dict(386 options=resource_arg_spec,387 required=resource_class.is_required_in_spec(),388 type='dict',389 )390 })391 if resource_class.allows_multiple_resource_spec():392 arg_spec[resource_class.get_spec_identifier()]['type'] = 'list'393 arg_spec[resource_class.get_spec_identifier()]['elements'] = (394 'dict')395 else:396 arg_spec.update(resource_arg_spec)397 return resource_arg_spec398 def _update_arg_spec_with_all_resources(self, resource_class, arg_spec):399 # updates _arg_spec with resource_class's arg_spec and all it's400 # sub-resources401 resource_arg_spec = self._update_arg_spec_with_resource(402 resource_class, arg_spec)403 # go to each child of resource_class and update it404 for sub_resources_class in self._get_sub_resources_class_of(405 resource_class):406 self._update_arg_spec_with_all_resources(407 sub_resources_class, resource_arg_spec)408 def _get_base_arg_spec_of_nsx_resource(self):409 resource_base_arg_spec = {}410 resource_base_arg_spec.update(411 # these are the base args for any NSXT Resource412 display_name=dict(413 required=False,414 type='str'415 ),416 description=dict(417 required=False,418 type='str'419 ),420 tags=dict(421 required=False,422 type='list',423 elements='dict',424 options=dict(425 scope=dict(426 required=True,427 type='str'428 ),429 tag=dict(430 required=True,431 type='str'432 )433 )434 )435 )436 return resource_base_arg_spec437 def _get_base_arg_spec_of_resource(self):438 resource_base_arg_spec = {}439 resource_base_arg_spec.update(440 id=dict(441 type='str'442 ),443 state=dict(444 required=True,445 type='str',446 choices=['present', 'absent']447 ),448 create_or_update_subresource_first=dict(449 default=False,450 type='bool'451 ),452 delete_subresource_first=dict(453 default=True,454 type='bool'455 ),456 achieve_subresource_state_if_del_parent=dict(457 default=False,458 type='bool'459 ),460 do_wait_till_create=dict(461 default=False,462 type='bool'463 )464 )465 return resource_base_arg_spec466 def _extract_nsx_resource_params(self, resource_params):467 # extract the params belonging to this resource only.468 filtered_params = {}469 def filter_with_spec(spec):470 for key in spec.keys():471 if (key in resource_params and472 resource_params[key] is not None):473 filtered_params[key] = resource_params[key]474 filter_with_spec(self.get_resource_spec())475 filter_with_spec(self._get_base_arg_spec_of_nsx_resource())476 return filtered_params477 def _achieve_present_state(self, successful_resource_exec_logs):478 self.update_resource_params(self.nsx_resource_params)479 is_resource_updated = self.check_for_update(480 self.existing_resource, self.nsx_resource_params)481 if not is_resource_updated:482 # Either the resource does not exist or it exists but was not483 # updated in the YAML.484 if self.module.check_mode:485 successful_resource_exec_logs.append({486 "changed": True,487 "debug_out": self.resource_params,488 "id": '12345',489 "resource_type": self.get_resource_name()490 })491 return492 try:493 if self.existing_resource:494 # Resource already exists495 successful_resource_exec_logs.append({496 "changed": False,497 "id": self.id,498 "message": "%s with id %s already exists." %499 (self.get_resource_name(), self.id),500 "resource_type": self.get_resource_name()501 })502 return503 # Create a new resource504 _, resp = self._send_request_to_API(505 suffix="/" + self.id, method='PATCH',506 data=self.nsx_resource_params)507 if self.do_wait_till_create() and not self._wait_till_create():508 raise Exception509 successful_resource_exec_logs.append({510 "changed": True,511 "id": self.id,512 "body": str(resp),513 "message": "%s with id %s created." %514 (self.get_resource_name(), self.id),515 "resource_type": self.get_resource_name()516 })517 except Exception as err:518 srel = successful_resource_exec_logs519 self.module.fail_json(msg="Failed to add %s with id %s."520 "Request body [%s]. Error[%s]."521 % (self.get_resource_name(),522 self.id, self.nsx_resource_params,523 to_native(err)524 ),525 successfully_updated_resources=srel)526 else:527 # The resource exists and was updated in the YAML.528 if self.module.check_mode:529 successful_resource_exec_logs.append({530 "changed": True,531 "debug_out": self.resource_params,532 "id": self.id,533 "resource_type": self.get_resource_name()534 })535 return536 self.nsx_resource_params['_revision'] = \537 self.existing_resource['_revision']538 try:539 _, resp = self._send_request_to_API(540 suffix="/"+self.id, method="PATCH",541 data=self.nsx_resource_params)542 successful_resource_exec_logs.append({543 "changed": True,544 "id": self.id,545 "body": str(resp),546 "message": "%s with id %s updated." %547 (self.get_resource_name(), self.id),548 "resource_type": self.get_resource_name()549 })550 except Exception as err:551 srel = successful_resource_exec_logs552 self.module.fail_json(msg="Failed to update %s with id %s."553 "Request body [%s]. Error[%s]." %554 (self.get_resource_name(), self.id,555 self.nsx_resource_params, to_native(556 err)557 ),558 successfully_updated_resources=srel)559 def _achieve_absent_state(self, successful_resource_exec_logs):560 if self.skip_delete():561 return562 if self.existing_resource is None:563 successful_resource_exec_logs.append({564 "changed": False,565 "msg": 'No %s exist with id %s' %566 (self.get_resource_name(), self.id),567 "resource_type": self.get_resource_name()568 })569 return570 if self.module.check_mode:571 successful_resource_exec_logs.append({572 "changed": True,573 "debug_out": self.resource_params,574 "id": self.id,575 "resource_type": self.get_resource_name()576 })577 return578 try:579 self._send_request_to_API("/" + self.id, method='DELETE')580 self._wait_till_delete()581 successful_resource_exec_logs.append({582 "changed": True,583 "id": self.id,584 "message": "%s with id %s deleted." %585 (self.get_resource_name(), self.id)586 })587 except Exception as err:588 srel = successful_resource_exec_logs589 self.module.fail_json(msg="Failed to delete %s with id %s. "590 "Error[%s]." % (self.get_resource_name(),591 self.id, to_native(err)),592 successfully_updated_resources=srel)593 def _send_request_to_API(self, suffix="", ignore_error=False,594 method='GET', data=None,595 resource_base_url=None,596 accepted_error_codes=set()):597 try:598 if not resource_base_url:599 if self.get_resource_name() not in BASE_RESOURCES:600 resource_base_url = (self.resource_class.601 get_resource_base_url(602 parent_info=self._parent_info))603 else:604 resource_base_url = (self.resource_class.605 get_resource_base_url(606 baseline_args=self.baseline_args))607 (rc, resp) = self.policy_communicator.request(608 resource_base_url + suffix,609 ignore_errors=ignore_error, method=method, data=data)610 return (rc, resp)611 except DuplicateRequestError:612 self.module.fail_json(msg='Duplicate request')613 except Exception as e:614 if (e.args[0] not in accepted_error_codes and615 self.get_resource_name() in BASE_RESOURCES):616 msg = ('Received {} from NSX Manager. Please try '617 'again. '.format(e.args[0]))618 if len(e.args) == 2 and e.args[1] and (619 'error_message' in e.args[1]):620 msg += e.args[1]['error_message']621 self.module.fail_json(msg=msg)622 raise e623 def _achieve_state(self, resource_params,624 successful_resource_exec_logs=[]):625 """626 Achieves `present` or `absent` state as specified in the YAML.627 """628 if self.id == self.INCORRECT_ARGUMENT_NAME_VALUE:629 # The resource was not specified in the YAML.630 # So, no need to realize it.631 return632 if (self._state == "present" and633 self.create_or_update_subresource_first()):634 self.achieve_subresource_state(635 resource_params, successful_resource_exec_logs)636 if self._state == "absent" and self.delete_subresource_first():637 self.achieve_subresource_state(638 resource_params, successful_resource_exec_logs)639 if self._state == 'present':640 self._achieve_present_state(641 successful_resource_exec_logs)642 else:643 self._achieve_absent_state(successful_resource_exec_logs)644 if self._state == "present" and not (645 self.create_or_update_subresource_first()):646 self.achieve_subresource_state(647 resource_params,648 successful_resource_exec_logs=successful_resource_exec_logs)649 if self._state == "absent" and not self.delete_subresource_first():650 self.achieve_subresource_state(651 resource_params, successful_resource_exec_logs)652 if self.get_resource_name() in BASE_RESOURCES:653 changed = False654 for successful_resource_exec_log in successful_resource_exec_logs:655 if successful_resource_exec_log["changed"]:656 changed = True657 break658 srel = successful_resource_exec_logs659 self.module.exit_json(changed=changed,660 successfully_updated_resources=srel)661 def _get_sub_resources_class_of(self, resource_class):662 subresources = []663 for attr in resource_class.__dict__.values():664 if (inspect.isclass(attr) and665 issubclass(attr, NSXTBaseRealizableResource)):666 subresources.append(attr)...
resource.py
Source:resource.py
...16 if self.is_entity():17 return EntityResponse(self, *args, **kwargs)18 else:19 return CollectionResponse(self, *args, **kwargs)20 def get_resource_name(self):21 return self.resource_name \22 if hasattr(self, 'resource_name') \23 else self.__class__.__name__.lower()24 def default_kwargs_for_urls(self):25 """26 Default keyword arguments for building the resource's urls.27 :return: dict28 """29 return {}30 def get_index_url(self, resource=None, **kwargs):31 """32 Builds the url of the resource's index.33 :param resource: name of the resource or None34 :param kwargs: additional keyword arguments to build the url35 :return: url of the resource's index36 """37 default_kwargs = self.default_kwargs_for_urls() \38 if resource == self.get_resource_name() else {}39 default_kwargs.update(kwargs)40 return self.get_full_url(41 self.app.reverse(42 '{}_index'.format(resource or self.get_resource_name()),43 **default_kwargs44 )45 )46 def get_definition_url(self, resource=None, **kwargs):47 default_kwargs = self.default_kwargs_for_urls() \48 if resource is None else {}49 default_kwargs.update(kwargs)50 return self.get_full_url(51 self.app.reverse(52 '{}_definition'.format(resource or self.get_resource_name()),53 **default_kwargs54 )55 )56 def get_object_url(self, id, resource=None, **kwargs):57 default_kwargs = self.default_kwargs_for_urls() \58 if resource is None else {}59 default_kwargs.update(kwargs)60 return self.get_full_url(61 self.app.reverse(62 '{}_item'.format(resource or self.get_resource_name()),63 id=id,64 **default_kwargs65 )66 )67 def get_self_url(self):68 resource = self.get_resource_name()69 return self.get_index_url(70 resource,71 **self.default_kwargs_for_urls()) if self.is_collection() \72 else self.get_object_url(73 self.request.match_info.get('id', None),74 resource, **self.default_kwargs_for_urls()75 )76 def get_encoder_class(self):77 return getattr(self, 'encoder_class', None)78 def is_entity(self):79 """80 Hacky solution to detect if current instance of the class is an entity81 or collection endpoint.82 """83 return 'id' in self.request.match_info84 def is_collection(self):85 return not self.is_entity()86 def has_nested_resources(self):87 if self.is_collection():88 return True if self.nested_collection_resources else False89 else:90 return True if self.nested_entity_resources else False91 def get_nested_resources(self):92 if self.is_collection():93 return self.nested_collection_resources94 else:95 return self.nested_entity_resources96 def get_nested_urls(self):97 current_args = self.request.match_info.copy()98 if 'id' in current_args:99 id = current_args.pop('id')100 current_args['{}_id'.format(self.get_resource_name())] = id101 def _nested_url(k, c):102 if issubclass(c, Resource):103 return self.get_index_url(c.__name__.lower(), **current_args)104 return self.get_full_url(105 self.app.reverse(106 '{}'.format(c.__name__.lower()),107 **current_args108 )109 )110 return {111 k: _nested_url(k, c)112 for k, c in self.get_nested_resources().items()113 }114 def get_parent(self):...
main.py
Source:main.py
1# See LICENSE for license details.2import pkg_resources3def get_resource_name(name):4 return pkg_resources.resource_filename(__name__, name)5mill_source = get_resource_name('assets/mill')...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
