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

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.