21502801941

Committed 29 Jan 2026 06:53PM UTC coverage: 86.962% (+0.007%) from 86.955%

Build # 21502801941

Build Type

push

github

Committed by

web-flow

Commit Message

Route53: add `Update` support for `RecordSet` resource (#13627)

Run Details

26 of 28 new or added lines in 1 file covered. (92.86%)

163 existing lines in 4 files now uncovered.

70379 of 80931 relevant lines covered (86.96%)

0.87 hits per line

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

94.62

/localstack-core/localstack/services/route53/resource_providers/aws_route53_recordset.py

# LocalStack Resource Provider Scaffolding v2
from __future__ import annotations

from pathlib import Path
from typing import TYPE_CHECKING, TypedDict

if TYPE_CHECKING:
    from mypy_boto3_route53 import Route53Client
    from mypy_boto3_route53.type_defs import ResourceRecordSetTypeDef

import localstack.services.cloudformation.provider_utils as util
from localstack.services.cloudformation.resource_provider import (
    OperationStatus,
    ProgressEvent,
    ResourceProvider,
    ResourceRequest,
)


class Route53RecordSetProperties(TypedDict):
    Name: str | None
    Type: str | None
    AliasTarget: AliasTarget | None
    CidrRoutingConfig: CidrRoutingConfig | None
    Comment: str | None
    Failover: str | None
    GeoLocation: GeoLocation | None
    HealthCheckId: str | None
    HostedZoneId: str | None
    HostedZoneName: str | None
    Id: str | None
    MultiValueAnswer: bool | None
    Region: str | None
    ResourceRecords: list[str] | None
    SetIdentifier: str | None
    TTL: str | None
    Weight: int | None


class AliasTarget(TypedDict):
    DNSName: str | None
    HostedZoneId: str | None
    EvaluateTargetHealth: bool | None


class CidrRoutingConfig(TypedDict):
    CollectionId: str | None
    LocationName: str | None


class GeoLocation(TypedDict):
    ContinentCode: str | None
    CountryCode: str | None
    SubdivisionCode: str | None


REPEATED_INVOCATION = "repeated_invocation"


class Route53RecordSetProvider(ResourceProvider[Route53RecordSetProperties]):
    TYPE = "AWS::Route53::RecordSet"  # Autogenerated. Don't change
    SCHEMA = util.get_schema_path(Path(__file__))  # Autogenerated. Don't change

    def create(
        self,
        request: ResourceRequest[Route53RecordSetProperties],
    ) -> ProgressEvent[Route53RecordSetProperties]:
        """
        Create a new resource.

        Primary identifier fields:
          - /properties/Id

        Required properties:
          - Type
          - Name

        Create-only properties:
          - /properties/HostedZoneName
          - /properties/Name
          - /properties/HostedZoneId

        Read-only properties:
          - /properties/Id
        """
        model = request.desired_state
        route53 = request.aws_client_factory.route53

        if not model.get("HostedZoneId"):
            # if only name was provided for hosted zone
            hosted_zone_name = model.get("HostedZoneName")
            hosted_zone_id = self.get_hosted_zone_id_from_name(hosted_zone_name, route53)
            model["HostedZoneId"] = hosted_zone_id

        attrs = self._get_resource_record_set_from_model(model)

        route53.change_resource_record_sets(
            HostedZoneId=model["HostedZoneId"],
            ChangeBatch={
                "Changes": [
                    {
                        "Action": "UPSERT",
                        "ResourceRecordSet": attrs,
                    },
                ]
            },
        )
        # TODO: not 100% sure this behaves the same between alias and non-alias records
        model["Id"] = model["Name"]

        return ProgressEvent(
            status=OperationStatus.SUCCESS,
            resource_model=model,
        )

    def get_hosted_zone_id_from_name(self, hosted_zone_name: str, client: Route53Client):
        if not hosted_zone_name:
            raise Exception("Either HostedZoneId or HostedZoneName must be present.")

        zones = client.list_hosted_zones_by_name(DNSName=hosted_zone_name)["HostedZones"]
        if len(zones) != 1:
            raise Exception(f"Ambiguous HostedZoneName {hosted_zone_name} provided.")

        hosted_zone_id = zones[0]["Id"]
        return hosted_zone_id

    def read(
        self,
        request: ResourceRequest[Route53RecordSetProperties],
    ) -> ProgressEvent[Route53RecordSetProperties]:
        """
        Fetch resource information


        """
        raise NotImplementedError

    def delete(
        self,
        request: ResourceRequest[Route53RecordSetProperties],
    ) -> ProgressEvent[Route53RecordSetProperties]:
        """
        Delete a resource


        """
        model = request.previous_state
        route53 = request.aws_client_factory.route53

        resource_record_set = self._get_resource_record_set_from_model(model)
        route53.change_resource_record_sets(
            HostedZoneId=model["HostedZoneId"],
            ChangeBatch={
                "Changes": [
                    {
                        "Action": "DELETE",
                        "ResourceRecordSet": resource_record_set,
                    },
                ]
            },
        )
        return ProgressEvent(
            status=OperationStatus.SUCCESS,
            resource_model=model,
        )

    def update(
        self,
        request: ResourceRequest[Route53RecordSetProperties],
    ) -> ProgressEvent[Route53RecordSetProperties]:
        """
        Update a resource


        """
        model = request.desired_state
        prev_model = request.previous_state

        assert request.previous_state is not None

        route53 = request.aws_client_factory.route53
        changes = []

        if model.get("SetIdentifier") != prev_model.get("SetIdentifier"):
            prev_rrset = self._get_resource_record_set_from_model(prev_model)
            changes.append(
                {
                    "Action": "DELETE",
                    "ResourceRecordSet": prev_rrset,
                }
            )

        updated_rrset = self._get_resource_record_set_from_model(model)
        changes.append(
            {
                "Action": "UPSERT",
                "ResourceRecordSet": updated_rrset,
            }
        )

        route53.change_resource_record_sets(
            HostedZoneId=model["HostedZoneId"],
            ChangeBatch={"Changes": changes},
        )
        model["Id"] = model["Name"]

        return ProgressEvent(
            status=OperationStatus.SUCCESS,
            resource_model=model,
        )

    @staticmethod
    def _get_resource_record_set_from_model(
        model: Route53RecordSetProperties,
    ) -> ResourceRecordSetTypeDef:
        attr_names = [
            "Name",
            "Type",
            "SetIdentifier",
            "Weight",
            "Region",
            "GeoLocation",
            "Failover",
            "MultiValueAnswer",
            "TTL",
            "ResourceRecords",
            "AliasTarget",
            "HealthCheckId",
        ]
        attrs = util.select_attributes(model, attr_names)

        if "AliasTarget" in attrs:
            # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-route53-recordset-aliastarget.html
            if "EvaluateTargetHealth" not in attrs["AliasTarget"]:
                attrs["AliasTarget"]["EvaluateTargetHealth"] = False
        else:
            # TODO: CNAME & SOA only allow 1 record type. should we check that here?
            attrs["ResourceRecords"] = [{"Value": record} for record in attrs["ResourceRecords"]]

        if "TTL" in attrs:
            if isinstance(attrs["TTL"], str):
                attrs["TTL"] = int(attrs["TTL"])

        return attrs

1	# LocalStack Resource Provider Scaffolding v2
2	from __future__ import annotations	1✔
3
4	from pathlib import Path	1✔
5	from typing import TYPE_CHECKING, TypedDict	1✔
6
7	if TYPE_CHECKING:	1✔
8	from mypy_boto3_route53 import Route53Client	×
NEW 9	from mypy_boto3_route53.type_defs import ResourceRecordSetTypeDef	×
10
11	import localstack.services.cloudformation.provider_utils as util	1✔
12	from localstack.services.cloudformation.resource_provider import (	1✔
13	OperationStatus,
14	ProgressEvent,
15	ResourceProvider,
16	ResourceRequest,
17	)
18
19
20	class Route53RecordSetProperties(TypedDict):	1✔
21	Name: str \| None	1✔
22	Type: str \| None	1✔
23	AliasTarget: AliasTarget \| None	1✔
24	CidrRoutingConfig: CidrRoutingConfig \| None	1✔
25	Comment: str \| None	1✔
26	Failover: str \| None	1✔
27	GeoLocation: GeoLocation \| None	1✔
28	HealthCheckId: str \| None	1✔
29	HostedZoneId: str \| None	1✔
30	HostedZoneName: str \| None	1✔
31	Id: str \| None	1✔
32	MultiValueAnswer: bool \| None	1✔
33	Region: str \| None	1✔
34	ResourceRecords: list[str] \| None	1✔
35	SetIdentifier: str \| None	1✔
36	TTL: str \| None	1✔
37	Weight: int \| None	1✔
38
39
40	class AliasTarget(TypedDict):	1✔
41	DNSName: str \| None	1✔
42	HostedZoneId: str \| None	1✔
43	EvaluateTargetHealth: bool \| None	1✔
44
45
46	class CidrRoutingConfig(TypedDict):	1✔
47	CollectionId: str \| None	1✔
48	LocationName: str \| None	1✔
49
50
51	class GeoLocation(TypedDict):	1✔
52	ContinentCode: str \| None	1✔
53	CountryCode: str \| None	1✔
54	SubdivisionCode: str \| None	1✔
55
56
57	REPEATED_INVOCATION = "repeated_invocation"	1✔
58
59
60	class Route53RecordSetProvider(ResourceProvider[Route53RecordSetProperties]):	1✔
61	TYPE = "AWS::Route53::RecordSet" # Autogenerated. Don't change	1✔
62	SCHEMA = util.get_schema_path(Path(__file__)) # Autogenerated. Don't change	1✔
63
64	def create(	1✔
65	self,
66	request: ResourceRequest[Route53RecordSetProperties],
67	) -> ProgressEvent[Route53RecordSetProperties]:
68	"""
69	Create a new resource.
70
71	Primary identifier fields:
72	- /properties/Id
73
74	Required properties:
75	- Type
76	- Name
77
78	Create-only properties:
79	- /properties/HostedZoneName
80	- /properties/Name
81	- /properties/HostedZoneId
82
83	Read-only properties:
84	- /properties/Id
85	"""
86	model = request.desired_state	1✔
87	route53 = request.aws_client_factory.route53	1✔
88
89	if not model.get("HostedZoneId"):	1✔
90	# if only name was provided for hosted zone
91	hosted_zone_name = model.get("HostedZoneName")	1✔
92	hosted_zone_id = self.get_hosted_zone_id_from_name(hosted_zone_name, route53)	1✔
93	model["HostedZoneId"] = hosted_zone_id	1✔
94
95	attrs = self._get_resource_record_set_from_model(model)	1✔
96
97	route53.change_resource_record_sets(	1✔
98	HostedZoneId=model["HostedZoneId"],
99	ChangeBatch={
100	"Changes": [
101	{
102	"Action": "UPSERT",
103	"ResourceRecordSet": attrs,
104	},
105	]
106	},
107	)
108	# TODO: not 100% sure this behaves the same between alias and non-alias records
109	model["Id"] = model["Name"]	1✔
110
111	return ProgressEvent(	1✔
112	status=OperationStatus.SUCCESS,
113	resource_model=model,
114	)
115
116	def get_hosted_zone_id_from_name(self, hosted_zone_name: str, client: Route53Client):	1✔
117	if not hosted_zone_name:	1✔
118	raise Exception("Either HostedZoneId or HostedZoneName must be present.")	×
119
120	zones = client.list_hosted_zones_by_name(DNSName=hosted_zone_name)["HostedZones"]	1✔
121	if len(zones) != 1:	1✔
122	raise Exception(f"Ambiguous HostedZoneName {hosted_zone_name} provided.")	×
123
124	hosted_zone_id = zones[0]["Id"]	1✔
125	return hosted_zone_id	1✔
126
127	def read(	1✔
128	self,
129	request: ResourceRequest[Route53RecordSetProperties],
130	) -> ProgressEvent[Route53RecordSetProperties]:
131	"""
132	Fetch resource information
133
134
135	"""
136	raise NotImplementedError
137
138	def delete(	1✔
139	self,
140	request: ResourceRequest[Route53RecordSetProperties],
141	) -> ProgressEvent[Route53RecordSetProperties]:
142	"""
143	Delete a resource
144
145
146	"""
147	model = request.previous_state	1✔
148	route53 = request.aws_client_factory.route53	1✔
149
150	resource_record_set = self._get_resource_record_set_from_model(model)	1✔
151	route53.change_resource_record_sets(	1✔
152	HostedZoneId=model["HostedZoneId"],
153	ChangeBatch={
154	"Changes": [
155	{
156	"Action": "DELETE",
157	"ResourceRecordSet": resource_record_set,
158	},
159	]
160	},
161	)
162	return ProgressEvent(	1✔
163	status=OperationStatus.SUCCESS,
164	resource_model=model,
165	)
166
167	def update(	1✔
168	self,
169	request: ResourceRequest[Route53RecordSetProperties],
170	) -> ProgressEvent[Route53RecordSetProperties]:
171	"""
172	Update a resource
173
174
175	"""
176	model = request.desired_state	1✔
177	prev_model = request.previous_state	1✔
178
179	assert request.previous_state is not None	1✔
180
181	route53 = request.aws_client_factory.route53	1✔
182	changes = []	1✔
183
184	if model.get("SetIdentifier") != prev_model.get("SetIdentifier"):	1✔
185	prev_rrset = self._get_resource_record_set_from_model(prev_model)	1✔
186	changes.append(	1✔
187	{
188	"Action": "DELETE",
189	"ResourceRecordSet": prev_rrset,
190	}
191	)
192
193	updated_rrset = self._get_resource_record_set_from_model(model)	1✔
194	changes.append(	1✔
195	{
196	"Action": "UPSERT",
197	"ResourceRecordSet": updated_rrset,
198	}
199	)
200
201	route53.change_resource_record_sets(	1✔
202	HostedZoneId=model["HostedZoneId"],
203	ChangeBatch={"Changes": changes},
204	)
205	model["Id"] = model["Name"]	1✔
206
207	return ProgressEvent(	1✔
208	status=OperationStatus.SUCCESS,
209	resource_model=model,
210	)
211
212	@staticmethod	1✔
213	def _get_resource_record_set_from_model(	1✔
214	model: Route53RecordSetProperties,
215	) -> ResourceRecordSetTypeDef:
216	attr_names = [	1✔
217	"Name",
218	"Type",
219	"SetIdentifier",
220	"Weight",
221	"Region",
222	"GeoLocation",
223	"Failover",
224	"MultiValueAnswer",
225	"TTL",
226	"ResourceRecords",
227	"AliasTarget",
228	"HealthCheckId",
229	]
230	attrs = util.select_attributes(model, attr_names)	1✔
231
232	if "AliasTarget" in attrs:	1✔
233	# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-route53-recordset-aliastarget.html
234	if "EvaluateTargetHealth" not in attrs["AliasTarget"]:	1✔
NEW 235	attrs["AliasTarget"]["EvaluateTargetHealth"] = False	×
236	else:
237	# TODO: CNAME & SOA only allow 1 record type. should we check that here?
238	attrs["ResourceRecords"] = [{"Value": record} for record in attrs["ResourceRecords"]]	1✔
239
240	if "TTL" in attrs:	1✔
241	if isinstance(attrs["TTL"], str):	1✔
242	attrs["TTL"] = int(attrs["TTL"])	1✔
243
244	return attrs	1✔

localstack / localstack / 21502801941

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