16820655284

Committed 07 Aug 2025 05:03PM UTC coverage: 86.841% (-0.05%) from 86.892%

Build # 16820655284

Build Type

push

github

Committed by

web-flow

Commit Message

CFNV2: support CDK bootstrap and deployment (#12967)

Coverage Stats

32 of 38 new or added lines in 5 files covered. (84.21%)

2013 existing lines in 125 files now uncovered.

66606 of 76699 relevant lines covered (86.84%)

0.87 hits per line

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

64.29

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

import localstack.services.route53resolver.utils
from localstack.aws.api.route53resolver import (
    FirewallConfig,
    FirewallDomainList,
    FirewallDomains,
    FirewallRule,
    FirewallRuleGroup,
    FirewallRuleGroupAssociation,
    ResolverQueryLogConfig,
    ResolverQueryLogConfigAssociation,
    ResolverQueryLogConfigStatus,
    ResourceNotFoundException,
)
from localstack.services.route53resolver.utils import get_firewall_config_id, validate_vpc
from localstack.services.stores import AccountRegionBundle, BaseStore, LocalAttribute


class Route53ResolverStore(BaseStore):
    firewall_configs: dict[str, FirewallConfig] = LocalAttribute(default=dict)
    firewall_domain_lists: dict[str, FirewallDomainList] = LocalAttribute(default=dict)
    firewall_domains: dict[str, FirewallDomains] = LocalAttribute(default=dict)
    firewall_rules: dict[str, FirewallRule] = LocalAttribute(default=dict)
    firewall_rule_groups: dict[str, FirewallRuleGroup] = LocalAttribute(default=dict)
    firewall_rule_group_associations: dict[str, FirewallRuleGroupAssociation] = LocalAttribute(
        default=dict
    )
    resolver_query_log_configs: dict[str, ResolverQueryLogConfig] = LocalAttribute(default=dict)
    resolver_query_log_config_associations: dict[str, ResolverQueryLogConfigAssociation] = (
        LocalAttribute(default=dict)
    )

    def get_firewall_rule_group(self, id):
        """returns firewall rule group with the given id if it exists"""

        firewall_rule_group = self.firewall_rule_groups.get(id)
        if not firewall_rule_group:
            raise ResourceNotFoundException(
                f"Can't find the resource with ID '{id}'. Trace Id: '{localstack.services.route53resolver.utils.get_trace_id()}'"
            )
        return firewall_rule_group

    def delete_firewall_rule_group(self, id):
        """deletes the firewall rule group with the given id"""
        # if firewall_rule_groups doesn't exist it will throw an error

        firewall_rule_group = self.get_firewall_rule_group(id)
        self.firewall_rule_groups.pop(id)
        return firewall_rule_group

    def get_firewall_rule_group_association(self, id):
        """returns firewall rule group association with the given id if it exists"""

        firewall_rule_group_association = self.firewall_rule_group_associations.get(id)
        if not firewall_rule_group_association:
            raise ResourceNotFoundException(
                f"[RSLVR-02025] Can't find the resource with ID '{id}'. Trace Id: '{localstack.services.route53resolver.utils.get_trace_id()}'"
            )
        return self.firewall_rule_group_associations.get(id)

    def delete_firewall_rule_group_association(self, id):
        """deletes the firewall rule group association with the given id"""
        # if firewall_rule_group_associations doesn't exist it will throw an error

        firewall_rule_group_associations = self.get_firewall_rule_group_association(id)
        self.firewall_rule_group_associations.pop(id)
        return firewall_rule_group_associations

    def get_firewall_domain(self, d):
        """returns firewall domain with the given id if it exists"""
        # firewall_domain can return none

        firewall_domain = self.firewall_domains.get(id)
        return firewall_domain

    def get_firewall_domain_list(self, id):
        """returns firewall domain list with the given id if it exists"""

        firewall_domain_list = self.firewall_domain_lists.get(id)
        if not firewall_domain_list:
            raise ResourceNotFoundException(
                f"Can't find the resource with ID '{id}'. Trace Id: '{localstack.services.route53resolver.utils.get_trace_id()}'"
            )
        return firewall_domain_list

    def delete_firewall_domain_list(self, id):
        """deletes the firewall domain list with the given id"""
        # if firewall_domain_lists doesn't exist it will throw an error

        firewall_domain_list = self.get_firewall_domain_list(id)
        self.firewall_domain_lists.pop(id)
        return firewall_domain_list

    def get_firewall_rule(self, firewall_rule_group_id, firewall_domain_list_id):
        """returns firewall rule with the given id if it exists"""

        firewall_rule = self.firewall_rules.get(firewall_rule_group_id, {}).get(
            firewall_domain_list_id
        )
        if not firewall_rule:
            raise ResourceNotFoundException(
                f"Can't find the resource with ID '{firewall_rule_group_id}'. Trace Id: '{localstack.services.route53resolver.utils.get_trace_id()}'"
            )
        return firewall_rule

    def delete_firewall_rule(self, firewall_rule_group_id, firewall_domain_list_id):
        """deletes the firewall rule with the given id"""
        # if firewall_rules doesn't exist it will throw an error

        firewall_rule = self.get_firewall_rule(firewall_rule_group_id, firewall_domain_list_id)
        self.firewall_rules.get(firewall_rule_group_id, {}).pop(firewall_domain_list_id)
        return firewall_rule

    def get_resolver_query_log_config(self, id):
        """returns resolver query log config with the given id if it exists"""

        resolver_query_log_config = self.resolver_query_log_configs.get(id)
        if not resolver_query_log_config:
            raise ResourceNotFoundException(
                f"[RSLVR-01601] The specified query logging configuration doesn't exist. Trace Id: '{localstack.services.route53resolver.utils.get_trace_id()}'"
            )
        return resolver_query_log_config

    def delete_resolver_query_log_config(self, id):
        """deletes the resolver query log config with the given id"""

        self.get_resolver_query_log_config(id)
        resolver_query_log_config = self.resolver_query_log_configs.pop(id)
        resolver_query_log_config["Status"] = ResolverQueryLogConfigStatus.DELETING
        return resolver_query_log_config

    def get_resolver_query_log_config_associations(self, id):
        """returns resolver query log config association with the given id if it exists"""

        resolver_query_log_config_association = self.resolver_query_log_config_associations.get(id)
        if not resolver_query_log_config_association:
            raise ResourceNotFoundException(
                f"[RSLVR-01601] The specified query logging configuration doesn't exist. Trace Id: '{localstack.services.route53resolver.utils.get_trace_id()}'"
            )
        return resolver_query_log_config_association

    def delete_resolver_query_log_config_associations(
        self, resolver_query_log_config_id, resource_id
    ):
        """deletes the resolver query log config association with the given id and vpc id"""

        association_id = None
        for association in self.resolver_query_log_config_associations.values():
            if not (
                association.get("ResolverQueryLogConfigId") == resolver_query_log_config_id
                and association.get("ResourceId") == resource_id
            ):
                raise ResourceNotFoundException(
                    f"[RSLVR-01602] The specified query logging configuration association doesn't exist. Trace Id: '{localstack.services.route53resolver.utils.get_trace_id()}'"
                )
            association["Status"] = "DELETING"
            association_id = association.get("Id")
        return self.resolver_query_log_config_associations.pop(association_id)

    def get_or_create_firewall_config(self, resource_id: str, region: str, account_id: str):
        """returns the firewall config with the given id if it exists or creates a new one"""

        validate_vpc(resource_id, region, account_id)
        firewall_config: FirewallConfig
        if self.firewall_configs.get(resource_id):
            firewall_config = self.firewall_configs[resource_id]
        else:
            id = get_firewall_config_id()
            firewall_config = FirewallConfig(
                Id=id,
                ResourceId=resource_id,
                OwnerId=account_id,
                FirewallFailOpen="DISABLED",
            )
            self.firewall_configs[resource_id] = firewall_config
        return firewall_config


route53resolver_stores = AccountRegionBundle("route53resolver", Route53ResolverStore)

1	import localstack.services.route53resolver.utils	1✔
2	from localstack.aws.api.route53resolver import (	1✔
3	FirewallConfig,
4	FirewallDomainList,
5	FirewallDomains,
6	FirewallRule,
7	FirewallRuleGroup,
8	FirewallRuleGroupAssociation,
9	ResolverQueryLogConfig,
10	ResolverQueryLogConfigAssociation,
11	ResolverQueryLogConfigStatus,
12	ResourceNotFoundException,
13	)
14	from localstack.services.route53resolver.utils import get_firewall_config_id, validate_vpc	1✔
15	from localstack.services.stores import AccountRegionBundle, BaseStore, LocalAttribute	1✔
16
17
18	class Route53ResolverStore(BaseStore):	1✔
19	firewall_configs: dict[str, FirewallConfig] = LocalAttribute(default=dict)	1✔
20	firewall_domain_lists: dict[str, FirewallDomainList] = LocalAttribute(default=dict)	1✔
21	firewall_domains: dict[str, FirewallDomains] = LocalAttribute(default=dict)	1✔
22	firewall_rules: dict[str, FirewallRule] = LocalAttribute(default=dict)	1✔
23	firewall_rule_groups: dict[str, FirewallRuleGroup] = LocalAttribute(default=dict)	1✔
24	firewall_rule_group_associations: dict[str, FirewallRuleGroupAssociation] = LocalAttribute(	1✔
25	default=dict
26	)
27	resolver_query_log_configs: dict[str, ResolverQueryLogConfig] = LocalAttribute(default=dict)	1✔
28	resolver_query_log_config_associations: dict[str, ResolverQueryLogConfigAssociation] = (	1✔
29	LocalAttribute(default=dict)
30	)
31
32	def get_firewall_rule_group(self, id):	1✔
33	"""returns firewall rule group with the given id if it exists"""
34
35	firewall_rule_group = self.firewall_rule_groups.get(id)	1✔
36	if not firewall_rule_group:	1✔
UNCOV 37	raise ResourceNotFoundException(	×
38	f"Can't find the resource with ID '{id}'. Trace Id: '{localstack.services.route53resolver.utils.get_trace_id()}'"
39	)
40	return firewall_rule_group	1✔
41
42	def delete_firewall_rule_group(self, id):	1✔
43	"""deletes the firewall rule group with the given id"""
44	# if firewall_rule_groups doesn't exist it will throw an error
45
46	firewall_rule_group = self.get_firewall_rule_group(id)	1✔
47	self.firewall_rule_groups.pop(id)	1✔
48	return firewall_rule_group	1✔
49
50	def get_firewall_rule_group_association(self, id):	1✔
51	"""returns firewall rule group association with the given id if it exists"""
52
UNCOV 53	firewall_rule_group_association = self.firewall_rule_group_associations.get(id)	×
UNCOV 54	if not firewall_rule_group_association:	×
55	raise ResourceNotFoundException(	×
56	f"[RSLVR-02025] Can't find the resource with ID '{id}'. Trace Id: '{localstack.services.route53resolver.utils.get_trace_id()}'"
57	)
UNCOV 58	return self.firewall_rule_group_associations.get(id)	×
59
60	def delete_firewall_rule_group_association(self, id):	1✔
61	"""deletes the firewall rule group association with the given id"""
62	# if firewall_rule_group_associations doesn't exist it will throw an error
63
UNCOV 64	firewall_rule_group_associations = self.get_firewall_rule_group_association(id)	×
UNCOV 65	self.firewall_rule_group_associations.pop(id)	×
66	return firewall_rule_group_associations	×
67
68	def get_firewall_domain(self, d):	1✔
69	"""returns firewall domain with the given id if it exists"""
70	# firewall_domain can return none
71
UNCOV 72	firewall_domain = self.firewall_domains.get(id)	×
UNCOV 73	return firewall_domain	×
74
75	def get_firewall_domain_list(self, id):	1✔
76	"""returns firewall domain list with the given id if it exists"""
77
78	firewall_domain_list = self.firewall_domain_lists.get(id)	1✔
79	if not firewall_domain_list:	1✔
UNCOV 80	raise ResourceNotFoundException(	×
81	f"Can't find the resource with ID '{id}'. Trace Id: '{localstack.services.route53resolver.utils.get_trace_id()}'"
82	)
83	return firewall_domain_list	1✔
84
85	def delete_firewall_domain_list(self, id):	1✔
86	"""deletes the firewall domain list with the given id"""
87	# if firewall_domain_lists doesn't exist it will throw an error
88
89	firewall_domain_list = self.get_firewall_domain_list(id)	1✔
90	self.firewall_domain_lists.pop(id)	1✔
91	return firewall_domain_list	1✔
92
93	def get_firewall_rule(self, firewall_rule_group_id, firewall_domain_list_id):	1✔
94	"""returns firewall rule with the given id if it exists"""
95
96	firewall_rule = self.firewall_rules.get(firewall_rule_group_id, {}).get(	1✔
97	firewall_domain_list_id
98	)
99	if not firewall_rule:	1✔
UNCOV 100	raise ResourceNotFoundException(	×
101	f"Can't find the resource with ID '{firewall_rule_group_id}'. Trace Id: '{localstack.services.route53resolver.utils.get_trace_id()}'"
102	)
103	return firewall_rule	1✔
104
105	def delete_firewall_rule(self, firewall_rule_group_id, firewall_domain_list_id):	1✔
106	"""deletes the firewall rule with the given id"""
107	# if firewall_rules doesn't exist it will throw an error
108
109	firewall_rule = self.get_firewall_rule(firewall_rule_group_id, firewall_domain_list_id)	1✔
110	self.firewall_rules.get(firewall_rule_group_id, {}).pop(firewall_domain_list_id)	1✔
111	return firewall_rule	1✔
112
113	def get_resolver_query_log_config(self, id):	1✔
114	"""returns resolver query log config with the given id if it exists"""
115
116	resolver_query_log_config = self.resolver_query_log_configs.get(id)	1✔
117	if not resolver_query_log_config:	1✔
118	raise ResourceNotFoundException(	1✔
119	f"[RSLVR-01601] The specified query logging configuration doesn't exist. Trace Id: '{localstack.services.route53resolver.utils.get_trace_id()}'"
120	)
121	return resolver_query_log_config	1✔
122
123	def delete_resolver_query_log_config(self, id):	1✔
124	"""deletes the resolver query log config with the given id"""
125
126	self.get_resolver_query_log_config(id)	1✔
127	resolver_query_log_config = self.resolver_query_log_configs.pop(id)	1✔
128	resolver_query_log_config["Status"] = ResolverQueryLogConfigStatus.DELETING	1✔
129	return resolver_query_log_config	1✔
130
131	def get_resolver_query_log_config_associations(self, id):	1✔
132	"""returns resolver query log config association with the given id if it exists"""
133
UNCOV 134	resolver_query_log_config_association = self.resolver_query_log_config_associations.get(id)	×
UNCOV 135	if not resolver_query_log_config_association:	×
136	raise ResourceNotFoundException(	×
137	f"[RSLVR-01601] The specified query logging configuration doesn't exist. Trace Id: '{localstack.services.route53resolver.utils.get_trace_id()}'"
138	)
UNCOV 139	return resolver_query_log_config_association	×
140
141	def delete_resolver_query_log_config_associations(	1✔
142	self, resolver_query_log_config_id, resource_id
143	):
144	"""deletes the resolver query log config association with the given id and vpc id"""
145
UNCOV 146	association_id = None	×
UNCOV 147	for association in self.resolver_query_log_config_associations.values():	×
148	if not (	×
149	association.get("ResolverQueryLogConfigId") == resolver_query_log_config_id
150	and association.get("ResourceId") == resource_id
151	):
UNCOV 152	raise ResourceNotFoundException(	×
153	f"[RSLVR-01602] The specified query logging configuration association doesn't exist. Trace Id: '{localstack.services.route53resolver.utils.get_trace_id()}'"
154	)
UNCOV 155	association["Status"] = "DELETING"	×
UNCOV 156	association_id = association.get("Id")	×
157	return self.resolver_query_log_config_associations.pop(association_id)	×
158
159	def get_or_create_firewall_config(self, resource_id: str, region: str, account_id: str):	1✔
160	"""returns the firewall config with the given id if it exists or creates a new one"""
161
UNCOV 162	validate_vpc(resource_id, region, account_id)	×
163	firewall_config: FirewallConfig
164	if self.firewall_configs.get(resource_id):	×
UNCOV 165	firewall_config = self.firewall_configs[resource_id]	×
166	else:
167	id = get_firewall_config_id()	×
UNCOV 168	firewall_config = FirewallConfig(	×
169	Id=id,
170	ResourceId=resource_id,
171	OwnerId=account_id,
172	FirewallFailOpen="DISABLED",
173	)
UNCOV 174	self.firewall_configs[resource_id] = firewall_config	×
UNCOV 175	return firewall_config	×
176
177
178	route53resolver_stores = AccountRegionBundle("route53resolver", Route53ResolverStore)	1✔

localstack / localstack / 16820655284

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