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

33.33

/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
UNCOV 35	firewall_rule_group = self.firewall_rule_groups.get(id)	×
UNCOV 36	if not firewall_rule_group:	×
37	raise ResourceNotFoundException(	×
38	f"Can't find the resource with ID '{id}'. Trace Id: '{localstack.services.route53resolver.utils.get_trace_id()}'"
39	)
UNCOV 40	return firewall_rule_group	×
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
UNCOV 46	firewall_rule_group = self.get_firewall_rule_group(id)	×
UNCOV 47	self.firewall_rule_groups.pop(id)	×
UNCOV 48	return firewall_rule_group	×
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
53	firewall_rule_group_association = self.firewall_rule_group_associations.get(id)	×
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	)
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
64	firewall_rule_group_associations = self.get_firewall_rule_group_association(id)	×
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
72	firewall_domain = self.firewall_domains.get(id)	×
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
UNCOV 78	firewall_domain_list = self.firewall_domain_lists.get(id)	×
UNCOV 79	if not firewall_domain_list:	×
80	raise ResourceNotFoundException(	×
81	f"Can't find the resource with ID '{id}'. Trace Id: '{localstack.services.route53resolver.utils.get_trace_id()}'"
82	)
UNCOV 83	return firewall_domain_list	×
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
UNCOV 89	firewall_domain_list = self.get_firewall_domain_list(id)	×
UNCOV 90	self.firewall_domain_lists.pop(id)	×
UNCOV 91	return firewall_domain_list	×
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
UNCOV 96	firewall_rule = self.firewall_rules.get(firewall_rule_group_id, {}).get(	×
97	firewall_domain_list_id
98	)
UNCOV 99	if not firewall_rule:	×
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	)
UNCOV 103	return firewall_rule	×
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
UNCOV 109	firewall_rule = self.get_firewall_rule(firewall_rule_group_id, firewall_domain_list_id)	×
UNCOV 110	self.firewall_rules.get(firewall_rule_group_id, {}).pop(firewall_domain_list_id)	×
UNCOV 111	return firewall_rule	×
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
UNCOV 116	resolver_query_log_config = self.resolver_query_log_configs.get(id)	×
UNCOV 117	if not resolver_query_log_config:	×
UNCOV 118	raise ResourceNotFoundException(	×
119	f"[RSLVR-01601] The specified query logging configuration doesn't exist. Trace Id: '{localstack.services.route53resolver.utils.get_trace_id()}'"
120	)
UNCOV 121	return resolver_query_log_config	×
122
123	def delete_resolver_query_log_config(self, id):	1✔
124	"""deletes the resolver query log config with the given id"""
125
UNCOV 126	self.get_resolver_query_log_config(id)	×
UNCOV 127	resolver_query_log_config = self.resolver_query_log_configs.pop(id)	×
UNCOV 128	resolver_query_log_config["Status"] = ResolverQueryLogConfigStatus.DELETING	×
UNCOV 129	return resolver_query_log_config	×
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
134	resolver_query_log_config_association = self.resolver_query_log_config_associations.get(id)	×
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	)
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
146	association_id = None	×
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	):
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	)
155	association["Status"] = "DELETING"	×
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
162	validate_vpc(resource_id, region, account_id)	×
163	firewall_config: FirewallConfig
164	if self.firewall_configs.get(resource_id):	×
165	firewall_config = self.firewall_configs[resource_id]	×
166	else:
167	id = get_firewall_config_id()	×
168	firewall_config = FirewallConfig(	×
169	Id=id,
170	ResourceId=resource_id,
171	OwnerId=account_id,
172	FirewallFailOpen="DISABLED",
173	)
174	self.firewall_configs[resource_id] = firewall_config	×
175	return firewall_config	×
176
177
178	route53resolver_stores = AccountRegionBundle("route53resolver", Route53ResolverStore)	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