22709357475

Committed 05 Mar 2026 08:35AM UTC coverage: 59.732% (-27.2%) from 86.974%

Build # 22709357475

Build Type

Pull #13880

github

Committed by

web-flow

Commit Message

Merge 28fcab93c into 710618057

Pull Request Pull Request #13880: Firehose: Replace TaggingService

Coverage Stats

12 of 12 new or added lines in 2 files covered. (100.0%)

20464 existing lines in 510 files now uncovered.

45290 of 75822 relevant lines covered (59.73%)

0.6 hits per line

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

45.21

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

import datetime

from localstack.aws.api.cloudwatch import (
    AlarmHistoryItem,
    CompositeAlarm,
    DashboardBody,
    MetricAlarm,
    StateValue,
)
from localstack.services.stores import (
    AccountRegionBundle,
    BaseStore,
    LocalAttribute,
)
from localstack.utils.aws import arns
from localstack.utils.tagging import Tags


class LocalStackMetricAlarm:
    region: str
    account_id: str
    alarm: MetricAlarm

    def __init__(self, account_id: str, region: str, alarm: MetricAlarm):
        self.account_id = account_id
        self.region = region
        self.alarm = alarm
        # Tags are already stored as part of Tagging Service or RGTA plugin
        self.alarm.pop("Tags", None)
        self.set_default_attributes()

    def set_default_attributes(self):
        current_time = datetime.datetime.now(datetime.UTC)
        self.alarm["AlarmArn"] = arns.cloudwatch_alarm_arn(
            self.alarm["AlarmName"], account_id=self.account_id, region_name=self.region
        )
        self.alarm["AlarmConfigurationUpdatedTimestamp"] = current_time
        self.alarm.setdefault("ActionsEnabled", True)
        self.alarm.setdefault("OKActions", [])
        self.alarm.setdefault("AlarmActions", [])
        self.alarm.setdefault("InsufficientDataActions", [])
        self.alarm["StateValue"] = StateValue.INSUFFICIENT_DATA
        self.alarm["StateReason"] = "Unchecked: Initial alarm creation"
        self.alarm["StateUpdatedTimestamp"] = current_time
        self.alarm.setdefault("Dimensions", [])
        self.alarm["StateTransitionedTimestamp"] = current_time


class LocalStackCompositeAlarm:
    region: str
    account_id: str
    alarm: CompositeAlarm

    def __init__(self, account_id: str, region: str, alarm: CompositeAlarm):
        self.account_id = account_id
        self.region = region
        self.alarm = alarm
        # Tags are already stored as part of Tagging Service or RGTA plugin
        self.alarm.pop("Tags", None)
        self.set_default_attributes()

    def set_default_attributes(self):
        current_time = datetime.datetime.now(datetime.UTC)
        self.alarm["AlarmArn"] = arns.cloudwatch_alarm_arn(
            self.alarm["AlarmName"], account_id=self.account_id, region_name=self.region
        )
        self.alarm["AlarmConfigurationUpdatedTimestamp"] = current_time
        self.alarm.setdefault("ActionsEnabled", True)
        self.alarm.setdefault("OKActions", [])
        self.alarm.setdefault("AlarmActions", [])
        self.alarm.setdefault("InsufficientDataActions", [])
        self.alarm["StateValue"] = StateValue.INSUFFICIENT_DATA
        self.alarm["StateReason"] = "Unchecked: Initial alarm creation"
        self.alarm["StateUpdatedTimestamp"] = current_time
        self.alarm["StateTransitionedTimestamp"] = current_time


class LocalStackDashboard:
    region: str
    account_id: str
    dashboard_name: str
    dashboard_arn: str
    dashboard_body: DashboardBody
    last_modified: datetime.datetime
    size: int

    def __init__(
        self, account_id: str, region: str, dashboard_name: str, dashboard_body: DashboardBody
    ):
        self.account_id = account_id
        self.region = region
        self.dashboard_name = dashboard_name
        self.dashboard_arn = arns.cloudwatch_dashboard_arn(
            self.dashboard_name, account_id=self.account_id, region_name=self.region
        )
        self.dashboard_body = dashboard_body
        self.last_modified = datetime.datetime.now()
        self.size = 225  # TODO: calculate size


LocalStackAlarm = LocalStackMetricAlarm | LocalStackCompositeAlarm


class CloudWatchStore(BaseStore):
    # maps resource ARN to alarms
    alarms: dict[str, LocalStackAlarm] = LocalAttribute(default=dict)

    # Contains all the Alarm Histories. Per documentation, an alarm history is retained even if the alarm is deleted,
    # making it necessary to save this at store level
    histories: list[AlarmHistoryItem] = LocalAttribute(default=list)

    dashboards: dict[str, LocalStackDashboard] = LocalAttribute(default=dict)
    # Maps resource ARN to tags
    tags: Tags = LocalAttribute(default=Tags)


cloudwatch_stores = AccountRegionBundle("cloudwatch", CloudWatchStore)

1	import datetime	1✔
2
3	from localstack.aws.api.cloudwatch import (	1✔
4	AlarmHistoryItem,
5	CompositeAlarm,
6	DashboardBody,
7	MetricAlarm,
8	StateValue,
9	)
10	from localstack.services.stores import (	1✔
11	AccountRegionBundle,
12	BaseStore,
13	LocalAttribute,
14	)
15	from localstack.utils.aws import arns	1✔
16	from localstack.utils.tagging import Tags	1✔
17
18
19	class LocalStackMetricAlarm:	1✔
20	region: str	1✔
21	account_id: str	1✔
22	alarm: MetricAlarm	1✔
23
24	def __init__(self, account_id: str, region: str, alarm: MetricAlarm):	1✔
UNCOV 25	self.account_id = account_id	×
UNCOV 26	self.region = region	×
UNCOV 27	self.alarm = alarm	×
28	# Tags are already stored as part of Tagging Service or RGTA plugin
UNCOV 29	self.alarm.pop("Tags", None)	×
UNCOV 30	self.set_default_attributes()	×
31
32	def set_default_attributes(self):	1✔
UNCOV 33	current_time = datetime.datetime.now(datetime.UTC)	×
UNCOV 34	self.alarm["AlarmArn"] = arns.cloudwatch_alarm_arn(	×
35	self.alarm["AlarmName"], account_id=self.account_id, region_name=self.region
36	)
UNCOV 37	self.alarm["AlarmConfigurationUpdatedTimestamp"] = current_time	×
UNCOV 38	self.alarm.setdefault("ActionsEnabled", True)	×
UNCOV 39	self.alarm.setdefault("OKActions", [])	×
UNCOV 40	self.alarm.setdefault("AlarmActions", [])	×
UNCOV 41	self.alarm.setdefault("InsufficientDataActions", [])	×
UNCOV 42	self.alarm["StateValue"] = StateValue.INSUFFICIENT_DATA	×
UNCOV 43	self.alarm["StateReason"] = "Unchecked: Initial alarm creation"	×
UNCOV 44	self.alarm["StateUpdatedTimestamp"] = current_time	×
UNCOV 45	self.alarm.setdefault("Dimensions", [])	×
UNCOV 46	self.alarm["StateTransitionedTimestamp"] = current_time	×
47
48
49	class LocalStackCompositeAlarm:	1✔
50	region: str	1✔
51	account_id: str	1✔
52	alarm: CompositeAlarm	1✔
53
54	def __init__(self, account_id: str, region: str, alarm: CompositeAlarm):	1✔
UNCOV 55	self.account_id = account_id	×
UNCOV 56	self.region = region	×
UNCOV 57	self.alarm = alarm	×
58	# Tags are already stored as part of Tagging Service or RGTA plugin
UNCOV 59	self.alarm.pop("Tags", None)	×
UNCOV 60	self.set_default_attributes()	×
61
62	def set_default_attributes(self):	1✔
UNCOV 63	current_time = datetime.datetime.now(datetime.UTC)	×
UNCOV 64	self.alarm["AlarmArn"] = arns.cloudwatch_alarm_arn(	×
65	self.alarm["AlarmName"], account_id=self.account_id, region_name=self.region
66	)
UNCOV 67	self.alarm["AlarmConfigurationUpdatedTimestamp"] = current_time	×
UNCOV 68	self.alarm.setdefault("ActionsEnabled", True)	×
UNCOV 69	self.alarm.setdefault("OKActions", [])	×
UNCOV 70	self.alarm.setdefault("AlarmActions", [])	×
UNCOV 71	self.alarm.setdefault("InsufficientDataActions", [])	×
UNCOV 72	self.alarm["StateValue"] = StateValue.INSUFFICIENT_DATA	×
UNCOV 73	self.alarm["StateReason"] = "Unchecked: Initial alarm creation"	×
UNCOV 74	self.alarm["StateUpdatedTimestamp"] = current_time	×
UNCOV 75	self.alarm["StateTransitionedTimestamp"] = current_time	×
76
77
78	class LocalStackDashboard:	1✔
79	region: str	1✔
80	account_id: str	1✔
81	dashboard_name: str	1✔
82	dashboard_arn: str	1✔
83	dashboard_body: DashboardBody	1✔
84	last_modified: datetime.datetime	1✔
85	size: int	1✔
86
87	def __init__(	1✔
88	self, account_id: str, region: str, dashboard_name: str, dashboard_body: DashboardBody
89	):
UNCOV 90	self.account_id = account_id	×
UNCOV 91	self.region = region	×
UNCOV 92	self.dashboard_name = dashboard_name	×
UNCOV 93	self.dashboard_arn = arns.cloudwatch_dashboard_arn(	×
94	self.dashboard_name, account_id=self.account_id, region_name=self.region
95	)
UNCOV 96	self.dashboard_body = dashboard_body	×
UNCOV 97	self.last_modified = datetime.datetime.now()	×
UNCOV 98	self.size = 225 # TODO: calculate size	×
99
100
101	LocalStackAlarm = LocalStackMetricAlarm \| LocalStackCompositeAlarm	1✔
102
103
104	class CloudWatchStore(BaseStore):	1✔
105	# maps resource ARN to alarms
106	alarms: dict[str, LocalStackAlarm] = LocalAttribute(default=dict)	1✔
107
108	# Contains all the Alarm Histories. Per documentation, an alarm history is retained even if the alarm is deleted,
109	# making it necessary to save this at store level
110	histories: list[AlarmHistoryItem] = LocalAttribute(default=list)	1✔
111
112	dashboards: dict[str, LocalStackDashboard] = LocalAttribute(default=dict)	1✔
113	# Maps resource ARN to tags
114	tags: Tags = LocalAttribute(default=Tags)	1✔
115
116
117	cloudwatch_stores = AccountRegionBundle("cloudwatch", CloudWatchStore)	1✔

localstack / localstack / 22709357475

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