20565403496

Committed 29 Dec 2025 05:11AM UTC coverage: 84.103% (-2.8%) from 86.921%

Build # 20565403496

Build Type

Pull #13567

github

Committed by

web-flow

Commit Message

Merge 4816837a5 into 2417384aa

Pull Request Pull Request #13567: Update ASF APIs

Run Details

67166 of 79862 relevant lines covered (84.1%)

0.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.54

/localstack-core/localstack/services/sns/v2/models.py

import itertools
import time
from dataclasses import dataclass, field
from enum import StrEnum
from typing import Literal, TypedDict

from localstack.aws.api.sns import (
    Endpoint,
    MessageAttributeMap,
    PhoneNumber,
    PlatformApplication,
    PublishBatchRequestEntry,
    TopicAttributesMap,
    subscriptionARN,
    topicARN,
)
from localstack.services.stores import (
    AccountRegionBundle,
    BaseStore,
    CrossRegionAttribute,
    LocalAttribute,
)
from localstack.utils.objects import singleton_factory
from localstack.utils.strings import long_uid
from localstack.utils.tagging import TaggingService


class Topic(TypedDict, total=True):
    arn: str
    name: str
    attributes: TopicAttributesMap
    data_protection_policy: str
    subscriptions: list[str]


SnsProtocols = Literal[
    "http", "https", "email", "email-json", "sms", "sqs", "application", "lambda", "firehose"
]

SnsApplicationPlatforms = Literal[
    "APNS", "APNS_SANDBOX", "ADM", "FCM", "Baidu", "GCM", "MPNS", "WNS"
]


class EndpointAttributeNames(StrEnum):
    CUSTOM_USER_DATA = "CustomUserData"
    Token = "Token"
    ENABLED = "Enabled"


SMS_ATTRIBUTE_NAMES = [
    "DeliveryStatusIAMRole",
    "DeliveryStatusSuccessSamplingRate",
    "DefaultSenderID",
    "DefaultSMSType",
    "UsageReportS3Bucket",
]
SMS_TYPES = ["Promotional", "Transactional"]
SMS_DEFAULT_SENDER_REGEX = r"^(?=[A-Za-z0-9]{1,11}$)(?=.*[A-Za-z])[A-Za-z0-9]+$"
SnsMessageProtocols = Literal[SnsProtocols, SnsApplicationPlatforms]


class SnsSubscription(TypedDict, total=False):
    """
    In SNS, Subscription can be represented with only TopicArn, Endpoint, Protocol, SubscriptionArn and Owner, for
    example in ListSubscriptions. However, when getting a subscription with GetSubscriptionAttributes, it will return
    the Subscription object merged with its own attributes.
    This represents this merged object, for internal use and in GetSubscriptionAttributes
    https://docs.aws.amazon.com/cli/latest/reference/sns/get-subscription-attributes.html
    """

    TopicArn: topicARN
    Endpoint: str
    Protocol: SnsProtocols
    SubscriptionArn: subscriptionARN
    PendingConfirmation: Literal["true", "false"]
    Owner: str | None
    SubscriptionPrincipal: str | None
    FilterPolicy: str | None
    FilterPolicyScope: Literal["MessageAttributes", "MessageBody"]
    RawMessageDelivery: Literal["true", "false"]
    ConfirmationWasAuthenticated: Literal["true", "false"]
    SubscriptionRoleArn: str | None
    DeliveryPolicy: str | None


@singleton_factory
def global_sns_message_sequence():
    # creates a 20-digit number used as the start for the global sequence, adds 100 for it to be different from SQS's
    # mostly for testing purpose, both global sequence would be initialized at the same and be identical
    start = int(time.time() + 100) << 33
    # itertools.count is thread safe over the GIL since its getAndIncrement operation is a single python bytecode op
    return itertools.count(start)


def get_next_sequence_number():
    return next(global_sns_message_sequence())


class SnsMessageType(StrEnum):
    Notification = "Notification"
    SubscriptionConfirmation = "SubscriptionConfirmation"
    UnsubscribeConfirmation = "UnsubscribeConfirmation"


@dataclass
class SnsMessage:
    type: SnsMessageType
    message: (
        str | dict
    )  # can be Dict if after being JSON decoded for validation if structure is `json`
    message_attributes: MessageAttributeMap | None = None
    message_structure: str | None = None
    subject: str | None = None
    message_deduplication_id: str | None = None
    message_group_id: str | None = None
    token: str | None = None
    message_id: str = field(default_factory=long_uid)
    is_fifo: bool | None = False
    sequencer_number: str | None = None

    def __post_init__(self):
        if self.message_attributes is None:
            self.message_attributes = {}
        if self.is_fifo:
            self.sequencer_number = str(get_next_sequence_number())

    def message_content(self, protocol: SnsMessageProtocols) -> str:
        """
        Helper function to retrieve the message content for the right protocol if the StructureMessage is `json`
        See https://docs.aws.amazon.com/sns/latest/dg/sns-send-custom-platform-specific-payloads-mobile-devices.html
        https://docs.aws.amazon.com/sns/latest/dg/example_sns_Publish_section.html
        :param protocol:
        :return: message content as string
        """
        if self.message_structure == "json":
            return self.message.get(protocol, self.message.get("default"))

        return self.message

    @classmethod
    def from_batch_entry(cls, entry: PublishBatchRequestEntry, is_fifo=False) -> "SnsMessage":
        return cls(
            type=SnsMessageType.Notification,
            message=entry["Message"],
            subject=entry.get("Subject"),
            message_structure=entry.get("MessageStructure"),
            message_attributes=entry.get("MessageAttributes"),
            message_deduplication_id=entry.get("MessageDeduplicationId"),
            message_group_id=entry.get("MessageGroupId"),
            is_fifo=is_fifo,
        )


@dataclass
class PlatformEndpoint:
    platform_application_arn: str
    platform_endpoint: Endpoint


@dataclass
class PlatformApplicationDetails:
    platform_application: PlatformApplication
    # maps all Endpoints of the PlatformApplication, from their Token to their ARN
    platform_endpoints: dict[str, str]


class SnsStore(BaseStore):
    topics: dict[str, Topic] = LocalAttribute(default=dict)

    # maps subscription ARN to SnsSubscription
    subscriptions: dict[str, SnsSubscription] = LocalAttribute(default=dict)

    # filter policy are stored as JSON string in subscriptions, store the decoded result Dict
    subscription_filter_policy: dict[subscriptionARN, dict] = LocalAttribute(default=dict)

    # maps confirmation token to subscription ARN
    subscription_tokens: dict[str, str] = LocalAttribute(default=dict)

    # maps platform application arns to platform applications
    platform_applications: dict[str, PlatformApplicationDetails] = LocalAttribute(default=dict)

    # maps endpoint arns to platform endpoints
    platform_endpoints: dict[str, PlatformEndpoint] = LocalAttribute(default=dict)

    # cache of topic ARN to platform endpoint messages (used primarily for testing)
    platform_endpoint_messages: dict[str, list[dict]] = LocalAttribute(default=dict)

    # topic/subscription independent default values for sending sms messages
    sms_attributes: dict[str, str] = LocalAttribute(default=dict)

    # list of sent SMS messages
    sms_messages: list[dict] = LocalAttribute(default=list)

    TAGS: TaggingService = CrossRegionAttribute(default=TaggingService)

    PHONE_NUMBERS_OPTED_OUT: list[PhoneNumber] = CrossRegionAttribute(default=list)


sns_stores = AccountRegionBundle("sns", SnsStore)

1	import itertools	1✔
2	import time	1✔
3	from dataclasses import dataclass, field	1✔
4	from enum import StrEnum	1✔
5	from typing import Literal, TypedDict	1✔
6
7	from localstack.aws.api.sns import (	1✔
8	Endpoint,
9	MessageAttributeMap,
10	PhoneNumber,
11	PlatformApplication,
12	PublishBatchRequestEntry,
13	TopicAttributesMap,
14	subscriptionARN,
15	topicARN,
16	)
17	from localstack.services.stores import (	1✔
18	AccountRegionBundle,
19	BaseStore,
20	CrossRegionAttribute,
21	LocalAttribute,
22	)
23	from localstack.utils.objects import singleton_factory	1✔
24	from localstack.utils.strings import long_uid	1✔
25	from localstack.utils.tagging import TaggingService	1✔
26
27
28	class Topic(TypedDict, total=True):	1✔
29	arn: str	1✔
30	name: str	1✔
31	attributes: TopicAttributesMap	1✔
32	data_protection_policy: str	1✔
33	subscriptions: list[str]	1✔
34
35
36	SnsProtocols = Literal[	1✔
37	"http", "https", "email", "email-json", "sms", "sqs", "application", "lambda", "firehose"
38	]
39
40	SnsApplicationPlatforms = Literal[	1✔
41	"APNS", "APNS_SANDBOX", "ADM", "FCM", "Baidu", "GCM", "MPNS", "WNS"
42	]
43
44
45	class EndpointAttributeNames(StrEnum):	1✔
46	CUSTOM_USER_DATA = "CustomUserData"	1✔
47	Token = "Token"	1✔
48	ENABLED = "Enabled"	1✔
49
50
51	SMS_ATTRIBUTE_NAMES = [	1✔
52	"DeliveryStatusIAMRole",
53	"DeliveryStatusSuccessSamplingRate",
54	"DefaultSenderID",
55	"DefaultSMSType",
56	"UsageReportS3Bucket",
57	]
58	SMS_TYPES = ["Promotional", "Transactional"]	1✔
59	SMS_DEFAULT_SENDER_REGEX = r"^(?=[A-Za-z0-9]{1,11}$)(?=.*[A-Za-z])[A-Za-z0-9]+$"	1✔
60	SnsMessageProtocols = Literal[SnsProtocols, SnsApplicationPlatforms]	1✔
61
62
63	class SnsSubscription(TypedDict, total=False):	1✔
64	"""
65	In SNS, Subscription can be represented with only TopicArn, Endpoint, Protocol, SubscriptionArn and Owner, for
66	example in ListSubscriptions. However, when getting a subscription with GetSubscriptionAttributes, it will return
67	the Subscription object merged with its own attributes.
68	This represents this merged object, for internal use and in GetSubscriptionAttributes
69	https://docs.aws.amazon.com/cli/latest/reference/sns/get-subscription-attributes.html
70	"""
71
72	TopicArn: topicARN	1✔
73	Endpoint: str	1✔
74	Protocol: SnsProtocols	1✔
75	SubscriptionArn: subscriptionARN	1✔
76	PendingConfirmation: Literal["true", "false"]	1✔
77	Owner: str \| None	1✔
78	SubscriptionPrincipal: str \| None	1✔
79	FilterPolicy: str \| None	1✔
80	FilterPolicyScope: Literal["MessageAttributes", "MessageBody"]	1✔
81	RawMessageDelivery: Literal["true", "false"]	1✔
82	ConfirmationWasAuthenticated: Literal["true", "false"]	1✔
83	SubscriptionRoleArn: str \| None	1✔
84	DeliveryPolicy: str \| None	1✔
85
86
87	@singleton_factory	1✔
88	def global_sns_message_sequence():	1✔
89	# creates a 20-digit number used as the start for the global sequence, adds 100 for it to be different from SQS's
90	# mostly for testing purpose, both global sequence would be initialized at the same and be identical
91	start = int(time.time() + 100) << 33	×
92	# itertools.count is thread safe over the GIL since its getAndIncrement operation is a single python bytecode op
93	return itertools.count(start)	×
94
95
96	def get_next_sequence_number():	1✔
97	return next(global_sns_message_sequence())	×
98
99
100	class SnsMessageType(StrEnum):	1✔
101	Notification = "Notification"	1✔
102	SubscriptionConfirmation = "SubscriptionConfirmation"	1✔
103	UnsubscribeConfirmation = "UnsubscribeConfirmation"	1✔
104
105
106	@dataclass	1✔
107	class SnsMessage:	1✔
108	type: SnsMessageType	1✔
109	message: (	1✔
110	str \| dict
111	) # can be Dict if after being JSON decoded for validation if structure is `json`
112	message_attributes: MessageAttributeMap \| None = None	1✔
113	message_structure: str \| None = None	1✔
114	subject: str \| None = None	1✔
115	message_deduplication_id: str \| None = None	1✔
116	message_group_id: str \| None = None	1✔
117	token: str \| None = None	1✔
118	message_id: str = field(default_factory=long_uid)	1✔
119	is_fifo: bool \| None = False	1✔
120	sequencer_number: str \| None = None	1✔
121
122	def __post_init__(self):	1✔
123	if self.message_attributes is None:	×
124	self.message_attributes = {}	×
125	if self.is_fifo:	×
126	self.sequencer_number = str(get_next_sequence_number())	×
127
128	def message_content(self, protocol: SnsMessageProtocols) -> str:	1✔
129	"""
130	Helper function to retrieve the message content for the right protocol if the StructureMessage is `json`
131	See https://docs.aws.amazon.com/sns/latest/dg/sns-send-custom-platform-specific-payloads-mobile-devices.html
132	https://docs.aws.amazon.com/sns/latest/dg/example_sns_Publish_section.html
133	:param protocol:
134	:return: message content as string
135	"""
136	if self.message_structure == "json":	×
137	return self.message.get(protocol, self.message.get("default"))	×
138
139	return self.message	×
140
141	@classmethod	1✔
142	def from_batch_entry(cls, entry: PublishBatchRequestEntry, is_fifo=False) -> "SnsMessage":	1✔
143	return cls(	×
144	type=SnsMessageType.Notification,
145	message=entry["Message"],
146	subject=entry.get("Subject"),
147	message_structure=entry.get("MessageStructure"),
148	message_attributes=entry.get("MessageAttributes"),
149	message_deduplication_id=entry.get("MessageDeduplicationId"),
150	message_group_id=entry.get("MessageGroupId"),
151	is_fifo=is_fifo,
152	)
153
154
155	@dataclass	1✔
156	class PlatformEndpoint:	1✔
157	platform_application_arn: str	1✔
158	platform_endpoint: Endpoint	1✔
159
160
161	@dataclass	1✔
162	class PlatformApplicationDetails:	1✔
163	platform_application: PlatformApplication	1✔
164	# maps all Endpoints of the PlatformApplication, from their Token to their ARN
165	platform_endpoints: dict[str, str]	1✔
166
167
168	class SnsStore(BaseStore):	1✔
169	topics: dict[str, Topic] = LocalAttribute(default=dict)	1✔
170
171	# maps subscription ARN to SnsSubscription
172	subscriptions: dict[str, SnsSubscription] = LocalAttribute(default=dict)	1✔
173
174	# filter policy are stored as JSON string in subscriptions, store the decoded result Dict
175	subscription_filter_policy: dict[subscriptionARN, dict] = LocalAttribute(default=dict)	1✔
176
177	# maps confirmation token to subscription ARN
178	subscription_tokens: dict[str, str] = LocalAttribute(default=dict)	1✔
179
180	# maps platform application arns to platform applications
181	platform_applications: dict[str, PlatformApplicationDetails] = LocalAttribute(default=dict)	1✔
182
183	# maps endpoint arns to platform endpoints
184	platform_endpoints: dict[str, PlatformEndpoint] = LocalAttribute(default=dict)	1✔
185
186	# cache of topic ARN to platform endpoint messages (used primarily for testing)
187	platform_endpoint_messages: dict[str, list[dict]] = LocalAttribute(default=dict)	1✔
188
189	# topic/subscription independent default values for sending sms messages
190	sms_attributes: dict[str, str] = LocalAttribute(default=dict)	1✔
191
192	# list of sent SMS messages
193	sms_messages: list[dict] = LocalAttribute(default=list)	1✔
194
195	TAGS: TaggingService = CrossRegionAttribute(default=TaggingService)	1✔
196
197	PHONE_NUMBERS_OPTED_OUT: list[PhoneNumber] = CrossRegionAttribute(default=list)	1✔
198
199
200	sns_stores = AccountRegionBundle("sns", SnsStore)	1✔

localstack / localstack / 20565403496

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous