KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use create_subscribe_url method in localstack

Best Python code snippet using localstack_python
provider.py
Source:provider.py
...510            if current_subscription["Protocol"] in ["http", "https"]:511                external_url = external_service_url("sns")512                subscription_token = short_uid()513                message_id = long_uid()514                subscription_url = create_subscribe_url(515                    external_url, current_subscription["TopicArn"], subscription_token516                )517                message = {518                    "Type": ["UnsubscribeConfirmation"],519                    "MessageId": [message_id],520                    "Token": [subscription_token],521                    "TopicArn": [current_subscription["TopicArn"]],522                    "Message": [523                        "You have chosen to deactivate subscription %s.\nTo cancel this operation and restore the subscription, visit the SubscribeURL included in this message."524                        % target_subscription_arn525                    ],526                    "SubscribeURL": [subscription_url],527                    "Timestamp": [datetime.datetime.utcnow().timestamp()],528                }529                headers = {530                    "x-amz-sns-message-type": "UnsubscribeConfirmation",531                    "x-amz-sns-message-id": message_id,532                    "x-amz-sns-topic-arn": current_subscription["TopicArn"],533                    "x-amz-sns-subscription-arn": target_subscription_arn,534                }535                publish_message(536                    current_subscription["TopicArn"],537                    message,538                    headers,539                    subscription_arn,540                    skip_checks=True,541                )542            return False543        for topic_arn, existing_subs in sns_backend.sns_subscriptions.items():544            sns_backend.sns_subscriptions[topic_arn] = [545                sub for sub in existing_subs if should_be_kept(sub, subscription_arn)546            ]547    def get_subscription_attributes(548        self, context: RequestContext, subscription_arn: subscriptionARN549    ) -> GetSubscriptionAttributesResponse:550        sub = get_subscription_by_arn(subscription_arn)551        if not sub:552            raise NotFoundException(f"Subscription with arn {subscription_arn} not found")553        return GetSubscriptionAttributesResponse(Attributes=sub)554    def list_subscriptions(555        self, context: RequestContext, next_token: nextToken = None556    ) -> ListSubscriptionsResponse:557        moto_response = call_moto(context)558        return ListSubscriptionsResponse(**moto_response)559    def publish(560        self,561        context: RequestContext,562        message: message,563        topic_arn: topicARN = None,564        target_arn: String = None,565        phone_number: String = None,566        subject: subject = None,567        message_structure: messageStructure = None,568        message_attributes: MessageAttributeMap = None,569        message_deduplication_id: String = None,570        message_group_id: String = None,571    ) -> PublishResponse:572        # We do not want the request to be forwarded to SNS backend573        if subject == "":574            raise InvalidParameterException("Invalid parameter: Subject")575        if not message or all(not m for m in message):576            raise InvalidParameterException("Invalid parameter: Empty message")577        if len(message) > MAXIMUM_MESSAGE_LENGTH:578            raise InvalidParameterException("Invalid parameter: Message too long")579        if topic_arn and ".fifo" in topic_arn:580            if not message_group_id:581                raise InvalidParameterException(582                    "The MessageGroupId parameter is required for FIFO topics",583                )584            moto_sns_backend = moto_sns_backends[context.region]585            if moto_sns_backend.get_topic(arn=topic_arn).content_based_deduplication == "false":586                if not message_deduplication_id:587                    raise InvalidParameterException(588                        "Invalid parameter: The topic should either have ContentBasedDeduplication enabled or MessageDeduplicationId provided explicitly",589                    )590        elif message_deduplication_id:591            # this is the first one to raise if both are set while the topic is not fifo592            raise InvalidParameterException(593                "Invalid parameter: MessageDeduplicationId Reason: The request includes MessageDeduplicationId parameter that is not valid for this topic type"594            )595        elif message_group_id:596            raise InvalidParameterException(597                "Invalid parameter: MessageGroupId Reason: The request includes MessageGroupId parameter that is not valid for this topic type"598            )599        sns_backend = SNSBackend.get()600        # No need to create a topic to send SMS or single push notifications with SNS601        # but we can't mock a sending so we only return that it went well602        if not phone_number and not target_arn:603            if topic_arn not in sns_backend.sns_subscriptions:604                raise NotFoundException(605                    "Topic does not exist",606                )607        # Legacy format to easily leverage existing publishing code608        # added parameters parsed by ASF. TODO: check/remove609        req_data = {610            "Action": ["Publish"],611            "TopicArn": [topic_arn],612            "TargetArn": target_arn,613            "Message": [message],614            "MessageAttributes": [message_attributes],615            "MessageDeduplicationId": [message_deduplication_id],616            "MessageGroupId": [message_group_id],617            "MessageStructure": [message_structure],618            "PhoneNumber": [phone_number],619            "Subject": [subject],620        }621        message_id = publish_message(622            topic_arn, req_data, context.request.headers, message_attributes=message_attributes623        )624        return PublishResponse(MessageId=message_id)625    def subscribe(626        self,627        context: RequestContext,628        topic_arn: topicARN,629        protocol: protocol,630        endpoint: endpoint = None,631        attributes: SubscriptionAttributesMap = None,632        return_subscription_arn: boolean = None,633    ) -> SubscribeResponse:634        if not endpoint:635            # TODO: check AWS behaviour (because endpoint is optional)636            raise NotFoundException("Endpoint not specified in subscription")637        if protocol not in SNS_PROTOCOLS:638            raise InvalidParameterException(639                f"Invalid parameter: Amazon SNS does not support this protocol string: {protocol}"640            )641        elif protocol in ["http", "https"] and not endpoint.startswith(f"{protocol}://"):642            raise InvalidParameterException(643                "Invalid parameter: Endpoint must match the specified protocol"644            )645        if ".fifo" in endpoint and ".fifo" not in topic_arn:646            raise InvalidParameterException(647                "FIFO SQS Queues can not be subscribed to standard SNS topics"648            )649        moto_response = call_moto(context)650        subscription_arn = moto_response.get("SubscriptionArn")651        filter_policy = moto_response.get("FilterPolicy")652        sns_backend = SNSBackend.get()653        topic_subs = sns_backend.sns_subscriptions[topic_arn] = (654            sns_backend.sns_subscriptions.get(topic_arn) or []655        )656        # An endpoint may only be subscribed to a topic once. Subsequent657        # subscribe calls do nothing (subscribe is idempotent).658        for existing_topic_subscription in topic_subs:659            if existing_topic_subscription.get("Endpoint") == endpoint:660                return SubscribeResponse(661                    SubscriptionArn=existing_topic_subscription["SubscriptionArn"]662                )663        subscription = {664            # http://docs.aws.amazon.com/cli/latest/reference/sns/get-subscription-attributes.html665            "TopicArn": topic_arn,666            "Endpoint": endpoint,667            "Protocol": protocol,668            "SubscriptionArn": subscription_arn,669            "FilterPolicy": filter_policy,670            "PendingConfirmation": "true",671        }672        if attributes:673            subscription.update(attributes)674        topic_subs.append(subscription)675        if subscription_arn not in sns_backend.subscription_status:676            sns_backend.subscription_status[subscription_arn] = {}677        subscription_token = short_uid()678        sns_backend.subscription_status[subscription_arn].update(679            {"TopicArn": topic_arn, "Token": subscription_token, "Status": "Not Subscribed"}680        )681        # Send out confirmation message for HTTP(S), fix for https://github.com/localstack/localstack/issues/881682        if protocol in ["http", "https"]:683            external_url = external_service_url("sns")684            subscription["UnsubscribeURL"] = create_unsubscribe_url(external_url, subscription_arn)685            confirmation = {686                "Type": ["SubscriptionConfirmation"],687                "Token": [subscription_token],688                "Message": [689                    f"You have chosen to subscribe to the topic {topic_arn}.\n"690                    + "To confirm the subscription, visit the SubscribeURL included in this message."691                ],692                "SubscribeURL": [create_subscribe_url(external_url, topic_arn, subscription_token)],693            }694            publish_message(topic_arn, confirmation, {}, subscription_arn, skip_checks=True)695        elif protocol in ["sqs", "lambda"]:696            # Auto-confirm sqs and lambda subscriptions for now697            # TODO: revisit for multi-account698            self.confirm_subscription(context, topic_arn, subscription_token)699        return SubscribeResponse(SubscriptionArn=subscription_arn)700    def tag_resource(701        self, context: RequestContext, resource_arn: AmazonResourceName, tags: TagList702    ) -> TagResourceResponse:703        # TODO: can this be used to tag any resource when using AWS?704        # each tag key must be unique705        # https://docs.aws.amazon.com/general/latest/gr/aws_tagging.html#tag-best-practices706        unique_tag_keys = {tag["Key"] for tag in tags}707        if len(unique_tag_keys) < len(tags):708            raise InvalidParameterException("Invalid parameter: Duplicated keys are not allowed.")709        call_moto(context)710        sns_backend = SNSBackend.get()711        existing_tags = sns_backend.sns_tags.get(resource_arn, [])712        def existing_tag_index(item):713            for idx, tag in enumerate(existing_tags):714                if item["Key"] == tag["Key"]:715                    return idx716            return None717        for item in tags:718            existing_index = existing_tag_index(item)719            if existing_index is None:720                existing_tags.append(item)721            else:722                existing_tags[existing_index] = item723        sns_backend.sns_tags[resource_arn] = existing_tags724        return TagResourceResponse()725    def delete_topic(self, context: RequestContext, topic_arn: topicARN) -> None:726        call_moto(context)727        sns_backend = SNSBackend.get()728        sns_backend.sns_subscriptions.pop(topic_arn, None)729        sns_backend.sns_tags.pop(topic_arn, None)730        event_publisher.fire_event(731            event_publisher.EVENT_SNS_DELETE_TOPIC,732            payload={"t": event_publisher.get_hash(topic_arn)},733        )734    def create_topic(735        self,736        context: RequestContext,737        name: topicName,738        attributes: TopicAttributesMap = None,739        tags: TagList = None,740    ) -> CreateTopicResponse:741        moto_response = call_moto(context)742        sns_backend = SNSBackend.get()743        topic_arn = moto_response["TopicArn"]744        tag_resource_success = extract_tags(topic_arn, tags, True, sns_backend)745        if not tag_resource_success:746            raise InvalidParameterException(747                "Invalid parameter: Tags Reason: Topic already exists with different tags"748            )749        if tags:750            self.tag_resource(context=context, resource_arn=topic_arn, tags=tags)751        sns_backend.sns_subscriptions[topic_arn] = (752            sns_backend.sns_subscriptions.get(topic_arn) or []753        )754        # publish event755        event_publisher.fire_event(756            event_publisher.EVENT_SNS_CREATE_TOPIC,757            payload={"t": event_publisher.get_hash(topic_arn)},758        )759        return CreateTopicResponse(TopicArn=topic_arn)760def message_to_subscribers(761    message_id,762    message,763    topic_arn,764    req_data,765    headers,766    subscription_arn=None,767    skip_checks=False,768    message_attributes=None,769):770    # AWS allows using TargetArn to publish to a topic, for backward compatibility771    if not topic_arn:772        topic_arn = req_data.get("TargetArn")773    sns_backend = SNSBackend.get()774    subscriptions = sns_backend.sns_subscriptions.get(topic_arn, [])775    async def wait_for_messages_sent():776        subs = [777            message_to_subscriber(778                message_id,779                message,780                topic_arn,781                req_data,782                headers,783                subscription_arn,784                skip_checks,785                sns_backend,786                subscriber,787                subscriptions,788                message_attributes,789            )790            for subscriber in list(subscriptions)791        ]792        if subs:793            await asyncio.wait(subs)794    asyncio.run(wait_for_messages_sent())795async def message_to_subscriber(796    message_id,797    message,798    topic_arn,799    req_data,800    headers,801    subscription_arn,802    skip_checks,803    sns_backend,804    subscriber,805    subscriptions,806    message_attributes,807):808    if subscription_arn not in [None, subscriber["SubscriptionArn"]]:809        return810    filter_policy = json.loads(subscriber.get("FilterPolicy") or "{}")811    if not skip_checks and not check_filter_policy(filter_policy, message_attributes):812        LOG.info(813            "SNS filter policy %s does not match attributes %s", filter_policy, message_attributes814        )815        return816    # todo: Message attributes are sent only when the message structure is String, not JSON.817    if subscriber["Protocol"] == "sms":818        event = {819            "topic_arn": topic_arn,820            "endpoint": subscriber["Endpoint"],821            "message_content": req_data["Message"][0],822        }823        sns_backend.sms_messages.append(event)824        LOG.info(825            "Delivering SMS message to %s: %s",826            subscriber["Endpoint"],827            req_data["Message"][0],828        )829        # MOCK DATA830        delivery = {831            "phoneCarrier": "Mock Carrier",832            "mnc": 270,833            "priceInUSD": 0.00645,834            "smsType": "Transactional",835            "mcc": 310,836            "providerResponse": "Message has been accepted by phone carrier",837            "dwellTimeMsUntilDeviceAck": 200,838        }839        store_delivery_log(subscriber, True, message, message_id, delivery)840        return841    elif subscriber["Protocol"] == "sqs":842        queue_url = None843        message_body = create_sns_message_body(subscriber, req_data, message_id, message_attributes)844        try:845            endpoint = subscriber["Endpoint"]846            if "sqs_queue_url" in subscriber:847                queue_url = subscriber.get("sqs_queue_url")848            elif "://" in endpoint:849                queue_url = endpoint850            else:851                queue_url = aws_stack.get_sqs_queue_url(endpoint)852                subscriber["sqs_queue_url"] = queue_url853            message_group_id = req_data.get("MessageGroupId", [""])[0]854            message_deduplication_id = req_data.get("MessageDeduplicationId", [""])[0]855            sqs_client = aws_stack.connect_to_service("sqs")856            kwargs = {}857            if message_group_id:858                kwargs["MessageGroupId"] = message_group_id859            if message_deduplication_id:860                kwargs["MessageDeduplicationId"] = message_deduplication_id861            sqs_client.send_message(862                QueueUrl=queue_url,863                MessageBody=message_body,864                MessageAttributes=create_sqs_message_attributes(subscriber, message_attributes),865                MessageSystemAttributes=create_sqs_system_attributes(headers),866                **kwargs,867            )868            store_delivery_log(subscriber, True, message, message_id)869        except Exception as exc:870            LOG.info("Unable to forward SNS message to SQS: %s %s", exc, traceback.format_exc())871            store_delivery_log(subscriber, False, message, message_id)872            sns_error_to_dead_letter_queue(subscriber, message_body, str(exc))873            if "NonExistentQueue" in str(exc):874                LOG.info(875                    'Removing non-existent queue "%s" subscribed to topic "%s"',876                    queue_url,877                    topic_arn,878                )879                subscriptions.remove(subscriber)880        return881    elif subscriber["Protocol"] == "lambda":882        try:883            external_url = external_service_url("sns")884            unsubscribe_url = create_unsubscribe_url(external_url, subscriber["SubscriptionArn"])885            response = lambda_api.process_sns_notification(886                subscriber["Endpoint"],887                topic_arn,888                subscriber["SubscriptionArn"],889                message,890                message_id,891                # see the format here892                # https://docs.aws.amazon.com/lambda/latest/dg/with-sns.html893                # issue with sdk to serialize the attribute inside lambda894                prepare_message_attributes(message_attributes),895                unsubscribe_url,896                subject=req_data.get("Subject")[0],897            )898            if response is not None:899                delivery = {900                    "statusCode": response.status_code,901                    "providerResponse": response.get_data(),902                }903                store_delivery_log(subscriber, True, message, message_id, delivery)904            # TODO: Check if it can be removed905            if isinstance(response, Response):906                response.raise_for_status()907            elif isinstance(response, FlaskResponse):908                if response.status_code >= 400:909                    raise Exception(910                        "Error response (code %s): %s" % (response.status_code, response.data)911                    )912        except Exception as exc:913            LOG.info(914                "Unable to run Lambda function on SNS message: %s %s", exc, traceback.format_exc()915            )916            store_delivery_log(subscriber, False, message, message_id)917            message_body = create_sns_message_body(918                subscriber, req_data, message_id, message_attributes919            )920            sns_error_to_dead_letter_queue(subscriber, message_body, str(exc))921        return922    elif subscriber["Protocol"] in ["http", "https"]:923        msg_type = (req_data.get("Type") or ["Notification"])[0]924        try:925            message_body = create_sns_message_body(926                subscriber, req_data, message_id, message_attributes927            )928        except Exception:929            return930        try:931            message_headers = {932                "Content-Type": "text/plain",933                # AWS headers according to934                # https://docs.aws.amazon.com/sns/latest/dg/sns-message-and-json-formats.html#http-header935                "x-amz-sns-message-type": msg_type,936                "x-amz-sns-message-id": message_id,937                "x-amz-sns-topic-arn": subscriber["TopicArn"],938                "User-Agent": "Amazon Simple Notification Service Agent",939            }940            if msg_type != "SubscriptionConfirmation":941                # while testing, never had those from AWS but the docs above states it should be there942                message_headers["x-amz-sns-subscription-arn"] = subscriber["SubscriptionArn"]943            # When raw message delivery is enabled, x-amz-sns-rawdelivery needs to be set to 'true'944            # indicating that the message has been published without JSON formatting.945            # https://docs.aws.amazon.com/sns/latest/dg/sns-large-payload-raw-message-delivery.html946            elif msg_type == "Notification" and is_raw_message_delivery(subscriber):947                message_headers["x-amz-sns-rawdelivery"] = "true"948            response = requests.post(949                subscriber["Endpoint"],950                headers=message_headers,951                data=message_body,952                verify=False,953            )954            delivery = {955                "statusCode": response.status_code,956                "providerResponse": response.content.decode("utf-8"),957            }958            store_delivery_log(subscriber, True, message, message_id, delivery)959            response.raise_for_status()960        except Exception as exc:961            LOG.info(962                "Received error on sending SNS message, putting to DLQ (if configured): %s", exc963            )964            store_delivery_log(subscriber, False, message, message_id)965            # AWS doesnt send to the DLQ if there's an error trying to deliver a UnsubscribeConfirmation msg966            if msg_type != "UnsubscribeConfirmation":967                sns_error_to_dead_letter_queue(subscriber, message_body, str(exc))968        return969    elif subscriber["Protocol"] == "application":970        try:971            sns_client = aws_stack.connect_to_service("sns")972            sns_client.publish(TargetArn=subscriber["Endpoint"], Message=message)973            store_delivery_log(subscriber, True, message, message_id)974        except Exception as exc:975            LOG.warning(976                "Unable to forward SNS message to SNS platform app: %s %s",977                exc,978                traceback.format_exc(),979            )980            store_delivery_log(subscriber, False, message, message_id)981            message_body = create_sns_message_body(subscriber, req_data, message_id)982            sns_error_to_dead_letter_queue(subscriber, message_body, str(exc))983        return984    elif subscriber["Protocol"] in ["email", "email-json"]:985        ses_client = aws_stack.connect_to_service("ses")986        if subscriber.get("Endpoint"):987            ses_client.verify_email_address(EmailAddress=subscriber.get("Endpoint"))988            ses_client.verify_email_address(EmailAddress="admin@localstack.com")989            ses_client.send_email(990                Source="admin@localstack.com",991                Message={992                    "Body": {993                        "Text": {994                            "Data": create_sns_message_body(995                                subscriber=subscriber, req_data=req_data, message_id=message_id996                            )997                            if subscriber["Protocol"] == "email-json"998                            else message999                        }1000                    },1001                    "Subject": {"Data": "SNS-Subscriber-Endpoint"},1002                },1003                Destination={"ToAddresses": [subscriber.get("Endpoint")]},1004            )1005            store_delivery_log(subscriber, True, message, message_id)1006    else:1007        LOG.warning('Unexpected protocol "%s" for SNS subscription', subscriber["Protocol"])1008def get_subscription_by_arn(sub_arn):1009    sns_backend = SNSBackend.get()1010    # TODO maintain separate map instead of traversing all items1011    for key, subscriptions in sns_backend.sns_subscriptions.items():1012        for sub in subscriptions:1013            if sub["SubscriptionArn"] == sub_arn:1014                return sub1015def create_sns_message_body(1016    subscriber, req_data, message_id=None, message_attributes: MessageAttributeMap = None1017) -> str:1018    message = req_data["Message"][0]1019    message_type = req_data.get("Type", ["Notification"])[0]1020    protocol = subscriber["Protocol"]1021    if req_data.get("MessageStructure") == ["json"]:1022        message = json.loads(message)1023        try:1024            message = message.get(protocol, message["default"])1025        except KeyError:1026            raise Exception("Unable to find 'default' key in message payload")1027    if message_type == "Notification" and is_raw_message_delivery(subscriber):1028        return message1029    external_url = external_service_url("sns")1030    data = {1031        "Type": message_type,1032        "MessageId": message_id,1033        "TopicArn": subscriber["TopicArn"],1034        "Message": message,1035        "Timestamp": timestamp_millis(),1036        "SignatureVersion": "1",1037        # TODO Add a more sophisticated solution with an actual signature1038        # Hardcoded1039        "Signature": "EXAMPLEpH+..",1040        "SigningCertURL": "https://sns.us-east-1.amazonaws.com/SimpleNotificationService-0000000000000000000000.pem",1041    }1042    if message_type == "Notification":1043        unsubscribe_url = create_unsubscribe_url(external_url, subscriber["SubscriptionArn"])1044        data["UnsubscribeURL"] = unsubscribe_url1045    for key in ["Subject", "SubscribeURL", "Token"]:1046        if req_data.get(key) and req_data[key][0]:1047            data[key] = req_data[key][0]1048    if message_attributes:1049        data["MessageAttributes"] = prepare_message_attributes(message_attributes)1050    return json.dumps(data)1051def _get_tags(topic_arn):1052    sns_backend = SNSBackend.get()1053    if topic_arn not in sns_backend.sns_tags:1054        sns_backend.sns_tags[topic_arn] = []1055    return sns_backend.sns_tags[topic_arn]1056def is_raw_message_delivery(susbcriber):1057    return susbcriber.get("RawMessageDelivery") in ("true", True, "True")1058def create_sqs_message_attributes(subscriber, attributes):1059    message_attributes = {}1060    if not is_raw_message_delivery(subscriber):1061        return message_attributes1062    for key, value in attributes.items():1063        # TODO: check if naming differs between ASF and QueryParameters, if not remove .get("Type") and .get("Value")1064        if value.get("Type") or value.get("DataType"):1065            tpe = value.get("Type") or value.get("DataType")1066            attribute = {"DataType": tpe}1067            if tpe == "Binary":1068                val = value.get("BinaryValue") or value.get("Value")1069                attribute["BinaryValue"] = base64.b64decode(to_bytes(val))1070                # base64 decoding might already have happened, in which decode fails.1071                # If decode fails, fallback to whatever is in there.1072                if not attribute["BinaryValue"]:1073                    attribute["BinaryValue"] = val1074            else:1075                val = value.get("StringValue") or value.get("Value", "")1076                attribute["StringValue"] = str(val)1077            message_attributes[key] = attribute1078    return message_attributes1079def prepare_message_attributes(message_attributes: MessageAttributeMap):1080    attributes = {}1081    if not message_attributes:1082        return attributes1083    # todo: Number type is not supported for Lambda subscriptions, passed as String1084    #  do conversion here1085    for attr_name, attr in message_attributes.items():1086        if attr.get("StringValue", None):1087            val = attr["StringValue"]1088        else:1089            # binary payload in base64 encoded by AWS, UTF-8 for JSON1090            # https://docs.aws.amazon.com/sns/latest/api/API_MessageAttributeValue.html1091            val = base64.b64encode(attr["BinaryValue"]).decode()1092        attributes[attr_name] = {1093            "Type": attr["DataType"],1094            "Value": val,1095        }1096    return attributes1097def create_subscribe_url(external_url, topic_arn, subscription_token):1098    return f"{external_url}/?Action=ConfirmSubscription&TopicArn={topic_arn}&Token={subscription_token}"1099def create_unsubscribe_url(external_url, subscription_arn):1100    return f"{external_url}/?Action=Unsubscribe&SubscriptionArn={subscription_arn}"1101def is_number(x):1102    try:1103        float(x)1104        return True1105    except ValueError:1106        return False1107def evaluate_numeric_condition(conditions, value):1108    if not is_number(value):1109        return False1110    for i in range(0, len(conditions), 2):1111        value = float(value)...
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.