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

How to use get_endpoint_attributes method in localstack

Best Python code snippet using localstack_python

__init__.py

Source:__init__.py

...158                )159    def __init__(self, endpoint_definition):160        super(EndpointBinder, self).__init__()161        self.endpoint_definition = endpoint_definition162        self.endpoint_attributes = endpoint_definition.get_endpoint_attributes()163        self.request_properties = endpoint_definition.get_request_properties()164        self.required_request_properties = (165            endpoint_definition.get_required_request_properties()166        )167        try:168            self.consumer_attributes = endpoint_definition.get_consumer_attributes()169        except AttributeError:170            self.consumer_attributes = []171        self.request_fields = endpoint_definition.get_request_fields()172        self.required_request_fields = endpoint_definition.get_required_request_fields()173        self.endpoint_tasks = endpoint_definition.get_tasks()174        self.url_fields = endpoint_definition.get_url_fields()175        self.adhoc_queries = endpoint_definition.get_adhoc_queries()176    def create_bound_endpoint(self, manager, request, *args, **kwargs):177        endpoint = self.endpoint_definition()178        for url_field in self.url_fields:179            if (url_field.api_name or url_field.name) in kwargs:180                url_field.set_value(kwargs.get(url_field.api_name or url_field.name))181        for adhoc_query_field in self.adhoc_queries:182            adhoc_query_field.set_value(183                {184                    key: val185                    for (key, val) in request.GET.items()186                    if key.startswith(adhoc_query_field.name)187                }188            )189        # Bind the request object within the instance (this allows RequestProperties to access the request190        # without the endpoint definition having direct access to it)191        RequestProperty.bind_request_to_instance(endpoint, request)192        bound_endpoint_manager = EndpointBinder.BoundEndpointManager(manager, endpoint)193        try:194            self._bind_endpoint(endpoint)195        except Exception as e:  # noqa196            bound_endpoint_manager.binding_exc_info = sys.exc_info()197            return bound_endpoint_manager198        try:199            self._validate_endpoint(endpoint)200        except Exception as e:  # noqa201            bound_endpoint_manager.validation_exc_info = sys.exc_info()202        return bound_endpoint_manager203    def _bind_endpoint(self, endpoint):204        # Access all request properties (this validates a request using the definition and caches the values)205        extra_error_message = ""206        missing_required_properties = []207        invalid_value_properties = []208        for request_property in self.request_properties:209            try:210                value = getattr(endpoint, request_property.name)211                if value is None and request_property.required:212                    if isinstance(request_property, ConsumerAttribute):213                        # A missing required consumer attribute should fail quickly as forbidden214                        raise errors.ClientErrorForbidden()215                    else:216                        # Otherwise collect missing properties and report them all together217                        missing_required_properties.append(request_property)218            except errors.ClientErrorMissingFields as mfe:  # TODO: seems unreachable219                extra_error_message += mfe.error_message  # pragma: nocover220            except (ValueError, errors.ClientErrorInvalidFieldValues) as ve:  # noqa221                # Collect invalid values and report them all together222                invalid_value_properties.append(request_property)  # pragma: nocover223        if missing_required_properties or extra_error_message:224            raise errors.ClientErrorMissingFields(225                [property.name for property in missing_required_properties],226                extra_message=extra_error_message,227            )228        if invalid_value_properties:229            raise errors.ClientErrorInvalidFieldValues(230                [request_property.name for request_property in invalid_value_properties]231            )232    def _validate_endpoint(self, endpoint):233        # Run standard validators234        try:235            if not (236                endpoint.is_authorized()237                and endpoint.is_permitted()238                and endpoint.is_valid()239            ):240                raise errors.ClientErrorForbidden(241                    additional_info=getattr(endpoint, "_validation_error_message", None)242                )243        except django.core.exceptions.ObjectDoesNotExist:244            raise errors.ClientErrorNotFound()245        # check ratelimit246        rate_limit_key = endpoint.rate_limit_key()247        if (rate_limit_key is not None) and rate_limit_exceeded(248            rate_limit_key, endpoint.rate_limit_period()249        ):250            raise errors.ClientErrorRequestThrottled()251class _EndpointRequestLifecycleManager(object):252    def __init__(self, endpoint_definition):253        super(_EndpointRequestLifecycleManager, self).__init__()254        self.endpoint_definition = endpoint_definition255        self.binder = EndpointBinder(endpoint_definition)256        self.endpoint_tasks = endpoint_definition.get_tasks()257    def bind_endpoint_to_request(self, request, *args, **kwargs):258        return self.binder.create_bound_endpoint(self, request, *args, **kwargs)259    def process_request_and_get_response(self, request, *args, **kwargs):260        bound_endpoint = self.bind_endpoint_to_request(request, *args, **kwargs)261        return bound_endpoint.get_response()262    def __str__(self):  # pragma: nocover263        return self.endpoint_definition.__name__264class BehavioralEndpointDefinitionRouter(object):265    def __init__(self, *endpoint_definitions):266        super(BehavioralEndpointDefinitionRouter, self).__init__()267        self.endpoint_definitions = endpoint_definitions268        self.endpoint_managers = [269            _EndpointRequestLifecycleManager(endpoint)270            for endpoint in endpoint_definitions271        ]272        self.endpoint_manager_names = "({0})".format(273            ",".join(map(lambda e: e.__name__, endpoint_definitions))274        )275    def bind_endpoint_to_request(self, request, *args, **kwargs):276        bound_endpoint = None277        for candidate_endpoint_manager in self.endpoint_managers:278            bound_endpoint = candidate_endpoint_manager.bind_endpoint_to_request(279                request, *args, **kwargs280            )281            if bound_endpoint.binding_exc_info is None:282                break283        return bound_endpoint284    def process_request_and_get_response(self, request, *args, **kwargs):285        try:286            bound_endpoint = self.bind_endpoint_to_request(request, *args, **kwargs)287            logger.info(288                "Processing request with handler %s",289                bound_endpoint.bound_endpoint.__class__.__name__,290            )291            return bound_endpoint.get_response()292        except errors.ApiError:293            raise294        except Exception as e:  # pragma: nocover295            raise errors.ServerError() from e296    def __call__(self, *args, **kwargs):297        return self.process_request_and_get_response(*args, **kwargs)298    def __str__(self):  # pragma: nocover299        return self.endpoint_manager_names300    @property301    def documentation(self):302        return [x.documentation() for x in self.endpoint_definitions]303class EndpointDefinitionMixin(metaclass=EndpointDefinitionMeta):304    pass305class BaseEndpointDefinition(metaclass=EndpointDefinitionMeta):306    @abc.abstractmethod307    def is_authorized(self):308        """Authorization check. Should be overridden by endpoint definition implementations.309        Returns:310            ``bool``: Whether or not the user should be able to access the resource. Defaults to ``False``.311        """312        return False313    def is_permitted(self):314        return True315    def is_valid(self):316        return True317    def rate_limit_key(self):318        """319        Should return a unique key that is used for rate-limiting requests to this endpoint.320        Return None if the request should not be rate-limited321        """322        return None323    def rate_limit_period(self):324        """325        number of seconds to enforce between requests with the same rate_limit_key326        """327        return 1328    @property329    def response_filter(self):330        filter_def_name = getattr(331            settings, "DECLARATIVE_ENDPOINT_DEFAULT_FILTERS", None332        )333        if filter_def_name:334            filter_def = locate_object(filter_def_name)335        else:336            filter_def = {}337        return filter_def338    @property339    def http_status(self):340        return http.client.OK341    @property342    @abc.abstractmethod343    def resource(self):344        """The instance of a resource. Should either be a ``dict`` or instance of a Django Model or QuerySet.345        This property *must* be implemented by all endpoint definitions.346        """347        raise NotImplementedError("Endpoints must implement self.resource property")348    @property349    def response(self):350        return self.resource351    @classmethod352    def get_endpoint_attributes(cls):353        endpoint_attributes = filter(354            lambda attribute: isinstance(attribute, EndpointAttribute),355            [getattr(cls, name) for name in dir(cls)],356        )357        return sorted(358            endpoint_attributes, key=lambda attribute: attribute.attribute_number359        )360    @classmethod361    def get_request_properties(cls):362        return list(363            filter(364                lambda attribute: isinstance(attribute, RequestProperty),365                cls.get_endpoint_attributes(),366            )367        )368    @classmethod369    def get_required_request_properties(cls):370        return list(371            filter(lambda property: property.required, cls.get_request_properties())372        )373    @classmethod374    def get_request_fields(cls):375        return list(376            filter(377                lambda property: isinstance(property, RequestField),378                cls.get_request_properties(),379            )380        )381    @classmethod382    def get_resource_fields(cls):383        return list(384            filter(385                lambda property: isinstance(property, ResourceField),386                cls.get_request_properties(),387            )388        )389    @classmethod390    def get_required_request_fields(cls):391        return list(392            filter(393                lambda property: isinstance(property, RequestField),394                cls.get_required_request_properties(),395            )396        )397    @classmethod398    def get_tasks(cls):399        endpoint_tasks = filter(400            lambda property: isinstance(property, EndpointTask),401            cls.get_endpoint_attributes(),402        )403        return sorted(endpoint_tasks, key=lambda task: task.priority)404    @classmethod405    def get_url_fields(cls):406        return list(407            filter(408                lambda property: isinstance(property, RequestUrlField),409                cls.get_endpoint_attributes(),410            )411        )412    @classmethod413    def documentation(cls):414        return {415            "class_name": cls.__name__,416            "fields": [p.documentation for p in cls.get_request_properties()],417        }418    @classmethod419    def get_adhoc_queries(cls):420        return [421            prop422            for prop in cls.get_endpoint_attributes()423            if isinstance(prop, RequestAdhocQuerySet)424        ]425class EndpointDefinition(BaseEndpointDefinition):426    """A base class to be used when defining endpoints.427    Base class to be used implementing endpoints that aren't necessarily tied to a model. Also implements428    basic consumer-based authentication.429    """430    request = RawRequestObjectProperty()431    _consumer_type = ConsumerAttribute(field_name="type", default="RW")432    is_read_only = False433    """ Used to determine accessibility for the current consumer.434    """435    def is_permitted(self):436        """Checks authorization for the current consumer....

test_application_boto3.py

Source:test_application_boto3.py

...153    endpoint_list.should.have.length_of(1)154    endpoint_list[0]["Attributes"]["CustomUserData"].should.equal("some data")155    endpoint_list[0]["EndpointArn"].should.equal(endpoint_arn)156@mock_sns157def test_get_endpoint_attributes():158    conn = boto3.client("sns", region_name="us-east-1")159    platform_application = conn.create_platform_application(160        Name="my-application", Platform="APNS", Attributes={}161    )162    application_arn = platform_application["PlatformApplicationArn"]163    endpoint = conn.create_platform_endpoint(164        PlatformApplicationArn=application_arn,165        Token="some_unique_id",166        CustomUserData="some user data",167        Attributes={"Enabled": "false", "CustomUserData": "some data"},168    )169    endpoint_arn = endpoint["EndpointArn"]170    attributes = conn.get_endpoint_attributes(EndpointArn=endpoint_arn)["Attributes"]171    attributes.should.equal(172        {"Token": "some_unique_id", "Enabled": "false", "CustomUserData": "some data"}173    )174@mock_sns175def test_get_non_existent_endpoint_attributes():176    conn = boto3.client("sns", region_name="us-east-1")177    endpoint_arn = "arn:aws:sns:us-east-1:123456789012:endpoint/APNS/my-application/c1f76c42-192a-4e75-b04f-a9268ce2abf3"178    with pytest.raises(conn.exceptions.NotFoundException) as excinfo:179        conn.get_endpoint_attributes(EndpointArn=endpoint_arn)180    error = excinfo.value.response["Error"]181    error["Type"].should.equal("Sender")182    error["Code"].should.equal("NotFound")183    error["Message"].should.equal("Endpoint does not exist")184@mock_sns185def test_get_missing_endpoint_attributes():186    conn = boto3.client("sns", region_name="us-east-1")187    conn.get_endpoint_attributes.when.called_with(188        EndpointArn="a-fake-arn"189    ).should.throw(ClientError)190@mock_sns191def test_set_endpoint_attributes():192    conn = boto3.client("sns", region_name="us-east-1")193    platform_application = conn.create_platform_application(194        Name="my-application", Platform="APNS", Attributes={}195    )196    application_arn = platform_application["PlatformApplicationArn"]197    endpoint = conn.create_platform_endpoint(198        PlatformApplicationArn=application_arn,199        Token="some_unique_id",200        CustomUserData="some user data",201        Attributes={"Enabled": "false", "CustomUserData": "some data"},202    )203    endpoint_arn = endpoint["EndpointArn"]204    conn.set_endpoint_attributes(205        EndpointArn=endpoint_arn, Attributes={"CustomUserData": "other data"}206    )207    attributes = conn.get_endpoint_attributes(EndpointArn=endpoint_arn)["Attributes"]208    attributes.should.equal(209        {"Token": "some_unique_id", "Enabled": "false", "CustomUserData": "other data"}210    )211@mock_sns212def test_publish_to_platform_endpoint():213    conn = boto3.client("sns", region_name="us-east-1")214    platform_application = conn.create_platform_application(215        Name="my-application", Platform="APNS", Attributes={}216    )217    application_arn = platform_application["PlatformApplicationArn"]218    endpoint = conn.create_platform_endpoint(219        PlatformApplicationArn=application_arn,220        Token="some_unique_id",221        CustomUserData="some user data",...

sns.py

Source:sns.py

...44def delete_topic(client=None, **kwargs):45    return client.delete_topic(**kwargs)46@sts_conn('sns')47@rate_limited()48def get_endpoint_attributes(client=None, **kwargs):49    return client.get_endpoint_attributes(**kwargs)['Attributes']50@sts_conn('sns')51@rate_limited()52def get_platform_application_attributes(client=None, **kwargs):53    return client.get_platform_application_attributes(**kwargs)['Attributes']54@sts_conn('sns')55@rate_limited()56def get_sms_attributes(client=None, **kwargs):57    return client.get_sms_attributes(**kwargs)['attributes']58@sts_conn('sns')59@rate_limited()60def get_subscription_attributes(client=None, **kwargs):61    return client.get_subscription_attributes(**kwargs)['Attributes']62@sts_conn('sns')63@rate_limited()...

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.