fix(route53): Allow upsert for resource records that already exist. (#5788)

moto/route53/exceptions.py

@@ -174,11 +174,11 @@ class DnsNameInvalidForZone(Route53ClientError):
         super().__init__("InvalidChangeBatch", error_msg)
 
 
-class ChangeSetAlreadyExists(Route53ClientError):
+class ResourceRecordAlreadyExists(Route53ClientError):
     code = 400
 
-    def __init__(self, action: str, name: str, _type: str):
+    def __init__(self, name: str, _type: str):
         super().__init__(
             "InvalidChangeBatch",
-            f"Tried to {action.lower()} resource record set [name='{name}', type='{_type}'] but it already exists",
+            f"Tried to create resource record set [name='{name}', type='{_type}'] but it already exists",
         )

moto/route53/models.py

@@ -19,7 +19,7 @@ from moto.route53.exceptions import (
     PublicZoneVPCAssociation,
     QueryLoggingConfigAlreadyExists,
     DnsNameInvalidForZone,
-    ChangeSetAlreadyExists,
+    ResourceRecordAlreadyExists,
     InvalidInput,
 )
 from moto.core import BaseBackend, BackendDict, BaseModel, CloudFormationModel
@@ -561,13 +561,11 @@ class Route53Backend(BaseBackend):
     def change_resource_record_sets(self, zoneid, change_list) -> None:
         the_zone = self.get_hosted_zone(zoneid)
 
-        for rr in change_list:
-            if rr in the_zone.rr_changes:
-                name = rr["ResourceRecordSet"]["Name"] + "."
-                _type = rr["ResourceRecordSet"]["Type"]
-                raise ChangeSetAlreadyExists(
-                    action=rr["Action"], name=name, _type=_type
-                )
+        for value in change_list:
+            if value["Action"] == "CREATE" and value in the_zone.rr_changes:
+                name = value["ResourceRecordSet"]["Name"] + "."
+                _type = value["ResourceRecordSet"]["Type"]
+                raise ResourceRecordAlreadyExists(name=name, _type=_type)
 
         for value in change_list:
             if value["Action"] == "DELETE":

tests/test_route53/test_route53.py

@@ -1169,7 +1169,7 @@ def test_change_resource_record_invalid_action_value():
 
 
 @mock_route53
-def test_change_resource_record_set_twice():
+def test_change_resource_record_set_create__should_fail_when_record_already_exists():
     ZONE = "cname.local"
     FQDN = f"test.{ZONE}"
     FQDN_TARGET = "develop.domain.com"
@@ -1195,8 +1195,54 @@ def test_change_resource_record_set_twice():
 
     with pytest.raises(ClientError) as exc:
         client.change_resource_record_sets(HostedZoneId=zone_id, ChangeBatch=changes)
+
     err = exc.value.response["Error"]
     err["Code"].should.equal("InvalidChangeBatch")
+    err["Message"].should.equal(
+        "Tried to create resource record set [name='test.cname.local.', type='CNAME'] but it already exists"
+    )
+
+
+@mock_route53
+def test_change_resource_record_set__should_create_record_when_using_upsert():
+    route53_client = boto3.client("route53", region_name="us-east-1")
+
+    hosted_zone = route53_client.create_hosted_zone(
+        Name="example.com", CallerReference="irrelevant"
+    )["HostedZone"]
+
+    resource_record = {
+        "Name": "test.example.com.",
+        "Type": "CNAME",
+        "TTL": 60,
+        "ResourceRecords": [{"Value": "www.test.example.com"}],
+    }
+
+    route53_client.change_resource_record_sets(
+        HostedZoneId=hosted_zone["Id"],
+        ChangeBatch={
+            "Changes": [{"Action": "UPSERT", "ResourceRecordSet": resource_record}],
+        },
+    )
+
+    response = route53_client.list_resource_record_sets(HostedZoneId=hosted_zone["Id"])
+
+    # The 1st and 2nd records are NS and SOA records, respectively.
+    len(response["ResourceRecordSets"]).should.equal(3)
+    response["ResourceRecordSets"][2].should.equal(resource_record)
+
+    # a subsequest UPSERT with the same ChangeBatch should succeed as well
+    route53_client.change_resource_record_sets(
+        HostedZoneId=hosted_zone["Id"],
+        ChangeBatch={
+            "Changes": [{"Action": "UPSERT", "ResourceRecordSet": resource_record}],
+        },
+    )
+    response = route53_client.list_resource_record_sets(HostedZoneId=hosted_zone["Id"])
+
+    # The 1st and 2nd records are NS and SOA records, respectively.
+    len(response["ResourceRecordSets"]).should.equal(3)
+    response["ResourceRecordSets"][2].should.equal(resource_record)
 
 
 @mock_route53