21697093787

Committed 04 Feb 2026 09:56PM UTC coverage: 86.962% (-0.004%) from 86.966%

Build # 21697093787

Build Type

push

github

Committed by

web-flow

Commit Message

improve system information sent in session and container_info (#13680)

Run Details

10 of 17 new or added lines in 2 files covered. (58.82%)

222 existing lines in 17 files now uncovered.

70560 of 81139 relevant lines covered (86.96%)

0.87 hits per line

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

97.34

/localstack-core/localstack/services/events/models.py

import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import Enum
from typing import Any, Literal, TypedDict

from localstack.aws.api import CommonServiceException
from localstack.aws.api.events import (
    ApiDestinationDescription,
    ApiDestinationHttpMethod,
    ApiDestinationInvocationRateLimitPerSecond,
    ApiDestinationName,
    ApiDestinationState,
    ArchiveDescription,
    ArchiveName,
    ArchiveState,
    Arn,
    ConnectionArn,
    ConnectionAuthorizationType,
    ConnectionDescription,
    ConnectionName,
    ConnectionState,
    ConnectivityResourceParameters,
    CreateConnectionAuthRequestParameters,
    CreatedBy,
    EventBusName,
    EventPattern,
    EventResourceList,
    EventSourceName,
    EventTime,
    HttpsEndpoint,
    ManagedBy,
    ReplayDescription,
    ReplayDestination,
    ReplayName,
    ReplayState,
    ReplayStateReason,
    RetentionDays,
    RoleArn,
    RuleDescription,
    RuleName,
    RuleState,
    ScheduleExpression,
    TagList,
    Target,
    TargetId,
    Timestamp,
)
from localstack.services.stores import (
    AccountRegionBundle,
    BaseStore,
    CrossRegionAttribute,
    LocalAttribute,
)
from localstack.utils.aws.arns import (
    event_bus_arn,
    events_api_destination_arn,
    events_archive_arn,
    events_connection_arn,
    events_replay_arn,
    events_rule_arn,
)
from localstack.utils.strings import short_uid
from localstack.utils.tagging import TaggingService

TargetDict = dict[TargetId, Target]


class ValidationException(CommonServiceException):
    def __init__(self, message: str):
        super().__init__("ValidationException", message, 400, True)


class InvalidEventPatternException(Exception):
    reason: str

    def __init__(self, reason=None, message=None) -> None:
        self.reason = reason
        self.message = message or f"Event pattern is not valid. Reason: {reason}"


FormattedEvent = TypedDict(  # functional syntax required due to name-name keys
    "FormattedEvent",
    {
        "version": str,
        "id": str,
        "detail-type": str | None,
        "source": EventSourceName | None,
        "account": str,
        "time": EventTime,
        "region": str,
        "resources": EventResourceList | None,
        "detail": dict[str, Any],
        "replay-name": ReplayName | None,
        "event-bus-name": EventBusName,
    },
)


FormattedEventDict = dict[str, FormattedEvent]
FormattedEventList = list[FormattedEvent]

type TransformedEvent = FormattedEvent | dict | str


class ResourceType(Enum):
    EVENT_BUS = "event_bus"
    RULE = "rule"


class Condition(TypedDict):
    Type: Literal["StringEquals"]
    Key: Literal["aws:PrincipalOrgID"]
    Value: str


class Statement(TypedDict):
    Sid: str
    Effect: str
    Principal: str | dict[str, str]
    Action: str
    Resource: str
    Condition: Condition


class ResourcePolicy(TypedDict):
    Version: str
    Statement: list[Statement]


@dataclass
class Rule:
    name: RuleName
    region: str
    account_id: str
    schedule_expression: ScheduleExpression | None = None
    event_pattern: EventPattern | None = None
    state: RuleState | None = None
    description: RuleDescription | None = None
    role_arn: RoleArn | None = None
    tags: TagList = field(default_factory=list)
    event_bus_name: EventBusName = "default"
    targets: TargetDict = field(default_factory=dict)
    managed_by: ManagedBy | None = None  # can only be set by AWS services
    created_by: CreatedBy = field(init=False)

    def __post_init__(self):
        self.created_by = self.account_id
        if self.tags is None:
            self.tags = []
        if self.targets is None:
            self.targets = {}
        if self.state is None:
            self.state = RuleState.ENABLED

    @property
    def arn(self) -> Arn:
        return events_rule_arn(self.name, self.account_id, self.region, self.event_bus_name)


RuleDict = dict[RuleName, Rule]


@dataclass
class Replay:
    name: str
    region: str
    account_id: str
    event_source_arn: Arn
    destination: ReplayDestination  # Event Bus Arn or Rule Arns
    event_start_time: Timestamp
    event_end_time: Timestamp
    description: ReplayDescription | None = None
    state: ReplayState | None = None
    state_reason: ReplayStateReason | None = None
    event_last_replayed_time: Timestamp | None = None
    replay_start_time: Timestamp | None = None
    replay_end_time: Timestamp | None = None

    @property
    def arn(self) -> Arn:
        return events_replay_arn(self.name, self.account_id, self.region)


ReplayDict = dict[ReplayName, Replay]


@dataclass
class Archive:
    name: ArchiveName
    region: str
    account_id: str
    event_source_arn: Arn
    description: ArchiveDescription | None = None
    event_pattern: EventPattern | None = None
    retention_days: RetentionDays | None = None
    state: ArchiveState = field(init=False, default=ArchiveState.DISABLED)
    creation_time: Timestamp = field(init=False, default_factory=lambda: datetime.now(UTC))
    events: FormattedEventDict = field(init=False, default_factory=dict)
    # TODO how to deal with updating this value?
    size_bytes: int = field(init=False, default=0)

    @property
    def arn(self) -> Arn:
        return events_archive_arn(self.name, self.account_id, self.region)

    @property
    def event_count(self) -> int:
        return len(self.events)


ArchiveDict = dict[ArchiveName, Archive]


@dataclass
class EventBus:
    name: EventBusName
    region: str
    account_id: str
    event_source_name: str | None = None
    description: str | None = None
    tags: TagList = field(default_factory=list)
    policy: ResourcePolicy | None = None
    rules: RuleDict = field(default_factory=dict)
    creation_time: Timestamp = field(init=False, default_factory=lambda: datetime.now(UTC))
    last_modified_time: Timestamp = field(init=False, default_factory=lambda: datetime.now(UTC))

    def __post_init__(self):
        if self.rules is None:
            self.rules = {}
        if self.tags is None:
            self.tags = []

    @property
    def arn(self) -> Arn:
        return event_bus_arn(self.name, self.account_id, self.region)


EventBusDict = dict[EventBusName, EventBus]


@dataclass
class Connection:
    name: ConnectionName
    region: str
    account_id: str
    authorization_type: ConnectionAuthorizationType
    auth_parameters: CreateConnectionAuthRequestParameters
    state: ConnectionState
    secret_arn: Arn
    description: ConnectionDescription | None = None
    invocation_connectivity_parameters: ConnectivityResourceParameters | None = None
    creation_time: Timestamp = field(init=False, default_factory=lambda: datetime.now(UTC))
    last_modified_time: Timestamp = field(init=False, default_factory=lambda: datetime.now(UTC))
    last_authorized_time: Timestamp = field(init=False, default_factory=lambda: datetime.now(UTC))
    tags: TagList = field(default_factory=list)
    id: str = str(uuid.uuid4())

    def __post_init__(self):
        if self.tags is None:
            self.tags = []

    @property
    def arn(self) -> Arn:
        return events_connection_arn(self.name, self.id, self.account_id, self.region)


ConnectionDict = dict[ConnectionName, Connection]


@dataclass
class ApiDestination:
    name: ApiDestinationName
    region: str
    account_id: str
    connection_arn: ConnectionArn
    invocation_endpoint: HttpsEndpoint
    http_method: ApiDestinationHttpMethod
    state: ApiDestinationState
    _invocation_rate_limit_per_second: ApiDestinationInvocationRateLimitPerSecond | None = None
    description: ApiDestinationDescription | None = None
    creation_time: Timestamp = field(init=False, default_factory=lambda: datetime.now(UTC))
    last_modified_time: Timestamp = field(init=False, default_factory=lambda: datetime.now(UTC))
    last_authorized_time: Timestamp = field(init=False, default_factory=lambda: datetime.now(UTC))
    tags: TagList = field(default_factory=list)
    id: str = str(short_uid())

    def __post_init__(self):
        if self.tags is None:
            self.tags = []

    @property
    def arn(self) -> Arn:
        return events_api_destination_arn(self.name, self.id, self.account_id, self.region)

    @property
    def invocation_rate_limit_per_second(self) -> int:
        return self._invocation_rate_limit_per_second or 300  # Default value

    @invocation_rate_limit_per_second.setter
    def invocation_rate_limit_per_second(
        self, value: ApiDestinationInvocationRateLimitPerSecond | None
    ):
        self._invocation_rate_limit_per_second = value


ApiDestinationDict = dict[ApiDestinationName, ApiDestination]


class EventsStore(BaseStore):
    # Map of eventbus names to eventbus objects. The name MUST be unique per account and region (works with AccountRegionBundle)
    event_buses: EventBusDict = LocalAttribute(default=dict)

    # Map of archive names to archive objects. The name MUST be unique per account and region (works with AccountRegionBundle)
    archives: ArchiveDict = LocalAttribute(default=dict)

    # Map of replay names to replay objects. The name MUST be unique per account and region (works with AccountRegionBundle)
    replays: ReplayDict = LocalAttribute(default=dict)

    # Map of connection names to connection objects.
    connections: ConnectionDict = LocalAttribute(default=dict)

    # Map of api destination names to api destination objects
    api_destinations: ApiDestinationDict = LocalAttribute(default=dict)

    # Maps resource ARN to tags
    TAGS: TaggingService = CrossRegionAttribute(default=TaggingService)


events_stores = AccountRegionBundle("events", EventsStore)

1	import uuid	1✔
2	from dataclasses import dataclass, field	1✔
3	from datetime import UTC, datetime	1✔
4	from enum import Enum	1✔
5	from typing import Any, Literal, TypedDict	1✔
6
7	from localstack.aws.api import CommonServiceException	1✔
8	from localstack.aws.api.events import (	1✔
9	ApiDestinationDescription,
10	ApiDestinationHttpMethod,
11	ApiDestinationInvocationRateLimitPerSecond,
12	ApiDestinationName,
13	ApiDestinationState,
14	ArchiveDescription,
15	ArchiveName,
16	ArchiveState,
17	Arn,
18	ConnectionArn,
19	ConnectionAuthorizationType,
20	ConnectionDescription,
21	ConnectionName,
22	ConnectionState,
23	ConnectivityResourceParameters,
24	CreateConnectionAuthRequestParameters,
25	CreatedBy,
26	EventBusName,
27	EventPattern,
28	EventResourceList,
29	EventSourceName,
30	EventTime,
31	HttpsEndpoint,
32	ManagedBy,
33	ReplayDescription,
34	ReplayDestination,
35	ReplayName,
36	ReplayState,
37	ReplayStateReason,
38	RetentionDays,
39	RoleArn,
40	RuleDescription,
41	RuleName,
42	RuleState,
43	ScheduleExpression,
44	TagList,
45	Target,
46	TargetId,
47	Timestamp,
48	)
49	from localstack.services.stores import (	1✔
50	AccountRegionBundle,
51	BaseStore,
52	CrossRegionAttribute,
53	LocalAttribute,
54	)
55	from localstack.utils.aws.arns import (	1✔
56	event_bus_arn,
57	events_api_destination_arn,
58	events_archive_arn,
59	events_connection_arn,
60	events_replay_arn,
61	events_rule_arn,
62	)
63	from localstack.utils.strings import short_uid	1✔
64	from localstack.utils.tagging import TaggingService	1✔
65
66	TargetDict = dict[TargetId, Target]	1✔
67
68
69	class ValidationException(CommonServiceException):	1✔
70	def __init__(self, message: str):	1✔
71	super().__init__("ValidationException", message, 400, True)	1✔
72
73
74	class InvalidEventPatternException(Exception):	1✔
75	reason: str	1✔
76
77	def __init__(self, reason=None, message=None) -> None:	1✔
78	self.reason = reason	×
79	self.message = message or f"Event pattern is not valid. Reason: {reason}"	×
80
81
82	FormattedEvent = TypedDict( # functional syntax required due to name-name keys	1✔
83	"FormattedEvent",
84	{
85	"version": str,
86	"id": str,
87	"detail-type": str \| None,
88	"source": EventSourceName \| None,
89	"account": str,
90	"time": EventTime,
91	"region": str,
92	"resources": EventResourceList \| None,
93	"detail": dict[str, Any],
94	"replay-name": ReplayName \| None,
95	"event-bus-name": EventBusName,
96	},
97	)
98
99
100	FormattedEventDict = dict[str, FormattedEvent]	1✔
101	FormattedEventList = list[FormattedEvent]	1✔
102
103	type TransformedEvent = FormattedEvent \| dict \| str	1✔
104
105
106	class ResourceType(Enum):	1✔
107	EVENT_BUS = "event_bus"	1✔
108	RULE = "rule"	1✔
109
110
111	class Condition(TypedDict):	1✔
112	Type: Literal["StringEquals"]	1✔
113	Key: Literal["aws:PrincipalOrgID"]	1✔
114	Value: str	1✔
115
116
117	class Statement(TypedDict):	1✔
118	Sid: str	1✔
119	Effect: str	1✔
120	Principal: str \| dict[str, str]	1✔
121	Action: str	1✔
122	Resource: str	1✔
123	Condition: Condition	1✔
124
125
126	class ResourcePolicy(TypedDict):	1✔
127	Version: str	1✔
128	Statement: list[Statement]	1✔
129
130
131	@dataclass	1✔
132	class Rule:	1✔
133	name: RuleName	1✔
134	region: str	1✔
135	account_id: str	1✔
136	schedule_expression: ScheduleExpression \| None = None	1✔
137	event_pattern: EventPattern \| None = None	1✔
138	state: RuleState \| None = None	1✔
139	description: RuleDescription \| None = None	1✔
140	role_arn: RoleArn \| None = None	1✔
141	tags: TagList = field(default_factory=list)	1✔
142	event_bus_name: EventBusName = "default"	1✔
143	targets: TargetDict = field(default_factory=dict)	1✔
144	managed_by: ManagedBy \| None = None # can only be set by AWS services	1✔
145	created_by: CreatedBy = field(init=False)	1✔
146
147	def __post_init__(self):	1✔
148	self.created_by = self.account_id	1✔
149	if self.tags is None:	1✔
150	self.tags = []	1✔
151	if self.targets is None:	1✔
152	self.targets = {}	1✔
153	if self.state is None:	1✔
154	self.state = RuleState.ENABLED	1✔
155
156	@property	1✔
157	def arn(self) -> Arn:	1✔
158	return events_rule_arn(self.name, self.account_id, self.region, self.event_bus_name)	1✔
159
160
161	RuleDict = dict[RuleName, Rule]	1✔
162
163
164	@dataclass	1✔
165	class Replay:	1✔
166	name: str	1✔
167	region: str	1✔
168	account_id: str	1✔
169	event_source_arn: Arn	1✔
170	destination: ReplayDestination # Event Bus Arn or Rule Arns	1✔
171	event_start_time: Timestamp	1✔
172	event_end_time: Timestamp	1✔
173	description: ReplayDescription \| None = None	1✔
174	state: ReplayState \| None = None	1✔
175	state_reason: ReplayStateReason \| None = None	1✔
176	event_last_replayed_time: Timestamp \| None = None	1✔
177	replay_start_time: Timestamp \| None = None	1✔
178	replay_end_time: Timestamp \| None = None	1✔
179
180	@property	1✔
181	def arn(self) -> Arn:	1✔
182	return events_replay_arn(self.name, self.account_id, self.region)	1✔
183
184
185	ReplayDict = dict[ReplayName, Replay]	1✔
186
187
188	@dataclass	1✔
189	class Archive:	1✔
190	name: ArchiveName	1✔
191	region: str	1✔
192	account_id: str	1✔
193	event_source_arn: Arn	1✔
194	description: ArchiveDescription \| None = None	1✔
195	event_pattern: EventPattern \| None = None	1✔
196	retention_days: RetentionDays \| None = None	1✔
197	state: ArchiveState = field(init=False, default=ArchiveState.DISABLED)	1✔
198	creation_time: Timestamp = field(init=False, default_factory=lambda: datetime.now(UTC))	1✔
199	events: FormattedEventDict = field(init=False, default_factory=dict)	1✔
200	# TODO how to deal with updating this value?
201	size_bytes: int = field(init=False, default=0)	1✔
202
203	@property	1✔
204	def arn(self) -> Arn:	1✔
205	return events_archive_arn(self.name, self.account_id, self.region)	1✔
206
207	@property	1✔
208	def event_count(self) -> int:	1✔
209	return len(self.events)	1✔
210
211
212	ArchiveDict = dict[ArchiveName, Archive]	1✔
213
214
215	@dataclass	1✔
216	class EventBus:	1✔
217	name: EventBusName	1✔
218	region: str	1✔
219	account_id: str	1✔
220	event_source_name: str \| None = None	1✔
221	description: str \| None = None	1✔
222	tags: TagList = field(default_factory=list)	1✔
223	policy: ResourcePolicy \| None = None	1✔
224	rules: RuleDict = field(default_factory=dict)	1✔
225	creation_time: Timestamp = field(init=False, default_factory=lambda: datetime.now(UTC))	1✔
226	last_modified_time: Timestamp = field(init=False, default_factory=lambda: datetime.now(UTC))	1✔
227
228	def __post_init__(self):	1✔
229	if self.rules is None:	1✔
230	self.rules = {}	1✔
231	if self.tags is None:	1✔
232	self.tags = []	1✔
233
234	@property	1✔
235	def arn(self) -> Arn:	1✔
236	return event_bus_arn(self.name, self.account_id, self.region)	1✔
237
238
239	EventBusDict = dict[EventBusName, EventBus]	1✔
240
241
242	@dataclass	1✔
243	class Connection:	1✔
244	name: ConnectionName	1✔
245	region: str	1✔
246	account_id: str	1✔
247	authorization_type: ConnectionAuthorizationType	1✔
248	auth_parameters: CreateConnectionAuthRequestParameters	1✔
249	state: ConnectionState	1✔
250	secret_arn: Arn	1✔
251	description: ConnectionDescription \| None = None	1✔
252	invocation_connectivity_parameters: ConnectivityResourceParameters \| None = None	1✔
253	creation_time: Timestamp = field(init=False, default_factory=lambda: datetime.now(UTC))	1✔
254	last_modified_time: Timestamp = field(init=False, default_factory=lambda: datetime.now(UTC))	1✔
255	last_authorized_time: Timestamp = field(init=False, default_factory=lambda: datetime.now(UTC))	1✔
256	tags: TagList = field(default_factory=list)	1✔
257	id: str = str(uuid.uuid4())	1✔
258
259	def __post_init__(self):	1✔
260	if self.tags is None:	1✔
UNCOV 261	self.tags = []	×
262
263	@property	1✔
264	def arn(self) -> Arn:	1✔
265	return events_connection_arn(self.name, self.id, self.account_id, self.region)	1✔
266
267
268	ConnectionDict = dict[ConnectionName, Connection]	1✔
269
270
271	@dataclass	1✔
272	class ApiDestination:	1✔
273	name: ApiDestinationName	1✔
274	region: str	1✔
275	account_id: str	1✔
276	connection_arn: ConnectionArn	1✔
277	invocation_endpoint: HttpsEndpoint	1✔
278	http_method: ApiDestinationHttpMethod	1✔
279	state: ApiDestinationState	1✔
280	_invocation_rate_limit_per_second: ApiDestinationInvocationRateLimitPerSecond \| None = None	1✔
281	description: ApiDestinationDescription \| None = None	1✔
282	creation_time: Timestamp = field(init=False, default_factory=lambda: datetime.now(UTC))	1✔
283	last_modified_time: Timestamp = field(init=False, default_factory=lambda: datetime.now(UTC))	1✔
284	last_authorized_time: Timestamp = field(init=False, default_factory=lambda: datetime.now(UTC))	1✔
285	tags: TagList = field(default_factory=list)	1✔
286	id: str = str(short_uid())	1✔
287
288	def __post_init__(self):	1✔
289	if self.tags is None:	1✔
UNCOV 290	self.tags = []	×
291
292	@property	1✔
293	def arn(self) -> Arn:	1✔
294	return events_api_destination_arn(self.name, self.id, self.account_id, self.region)	1✔
295
296	@property	1✔
297	def invocation_rate_limit_per_second(self) -> int:	1✔
298	return self._invocation_rate_limit_per_second or 300 # Default value	1✔
299
300	@invocation_rate_limit_per_second.setter	1✔
301	def invocation_rate_limit_per_second(	1✔
302	self, value: ApiDestinationInvocationRateLimitPerSecond \| None
303	):
UNCOV 304	self._invocation_rate_limit_per_second = value	×
305
306
307	ApiDestinationDict = dict[ApiDestinationName, ApiDestination]	1✔
308
309
310	class EventsStore(BaseStore):	1✔
311	# Map of eventbus names to eventbus objects. The name MUST be unique per account and region (works with AccountRegionBundle)
312	event_buses: EventBusDict = LocalAttribute(default=dict)	1✔
313
314	# Map of archive names to archive objects. The name MUST be unique per account and region (works with AccountRegionBundle)
315	archives: ArchiveDict = LocalAttribute(default=dict)	1✔
316
317	# Map of replay names to replay objects. The name MUST be unique per account and region (works with AccountRegionBundle)
318	replays: ReplayDict = LocalAttribute(default=dict)	1✔
319
320	# Map of connection names to connection objects.
321	connections: ConnectionDict = LocalAttribute(default=dict)	1✔
322
323	# Map of api destination names to api destination objects
324	api_destinations: ApiDestinationDict = LocalAttribute(default=dict)	1✔
325
326	# Maps resource ARN to tags
327	TAGS: TaggingService = CrossRegionAttribute(default=TaggingService)	1✔
328
329
330	events_stores = AccountRegionBundle("events", EventsStore)	1✔

localstack / localstack / 21697093787

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