RDS: add restore db instance to point in time (#7203)

IMPLEMENTATION_COVERAGE.md

@@ -5807,7 +5807,7 @@
 - [ ] restore_db_cluster_to_point_in_time
 - [X] restore_db_instance_from_db_snapshot
 - [ ] restore_db_instance_from_s3
-- [ ] restore_db_instance_to_point_in_time
+- [X] restore_db_instance_to_point_in_time
 - [ ] revoke_db_security_group_ingress
 - [ ] start_activity_stream
 - [X] start_db_cluster

moto/rds/models.py

@@ -1797,6 +1797,30 @@ class RDSBackend(BaseBackend):
 
         return self.create_db_instance(new_instance_props)
 
+    def restore_db_instance_to_point_in_time(
+        self,
+        source_db_identifier: str,
+        target_db_identifier: str,
+        overrides: Dict[str, Any],
+    ) -> Database:
+        db_instance = self.describe_db_instances(
+            db_instance_identifier=source_db_identifier
+        )[0]
+        new_instance_props = copy.deepcopy(db_instance.__dict__)
+        if not db_instance.option_group_supplied:
+            # If the option group is not supplied originally, the 'option_group_name' will receive a default value
+            # Force this reconstruction, and prevent any validation on the default value
+            del new_instance_props["option_group_name"]
+
+        for key, value in overrides.items():
+            if value:
+                new_instance_props[key] = value
+
+        # set the new db instance identifier
+        new_instance_props["db_instance_identifier"] = target_db_identifier
+
+        return self.create_db_instance(new_instance_props)
+
     def stop_db_instance(
         self, db_instance_identifier: str, db_snapshot_identifier: Optional[str] = None
     ) -> Database:

moto/rds/responses.py

@@ -367,6 +367,17 @@ class RDSResponse(BaseResponse):
         template = self.response_template(RESTORE_INSTANCE_FROM_SNAPSHOT_TEMPLATE)
         return template.render(database=new_instance)
 
+    def restore_db_instance_to_point_in_time(self) -> str:
+        source_db_identifier = self._get_param("SourceDBInstanceIdentifier")
+        target_db_identifier = self._get_param("TargetDBInstanceIdentifier")
+
+        db_kwargs = self._get_db_kwargs()
+        new_instance = self.backend.restore_db_instance_to_point_in_time(
+            source_db_identifier, target_db_identifier, db_kwargs
+        )
+        template = self.response_template(RESTORE_INSTANCE_TO_POINT_IN_TIME_TEMPLATE)
+        return template.render(database=new_instance)
+
     def list_tags_for_resource(self) -> str:
         arn = self._get_param("ResourceName")
         template = self.response_template(LIST_TAGS_FOR_RESOURCE_TEMPLATE)
@@ -914,6 +925,16 @@ RESTORE_INSTANCE_FROM_SNAPSHOT_TEMPLATE = """<RestoreDBInstanceFromDBSnapshotRes
   </ResponseMetadata>
 </RestoreDBInstanceFromDBSnapshotResponse>"""
 
+
+RESTORE_INSTANCE_TO_POINT_IN_TIME_TEMPLATE = """<RestoreDBInstanceToPointInTimeResponse xmlns="http://rds.amazonaws.com/doc/2014-09-01/">
+  <RestoreDBInstanceToPointInTimeResult>
+  {{ database.to_xml() }}
+  </RestoreDBInstanceToPointInTimeResult>
+  <ResponseMetadata>
+    <RequestId>523e3218-afc7-11c3-90f5-f90431260ab4</RequestId>
+  </ResponseMetadata>
+</RestoreDBInstanceToPointInTimeResponse>"""
+
 CREATE_SNAPSHOT_TEMPLATE = """<CreateDBSnapshotResponse xmlns="http://rds.amazonaws.com/doc/2014-09-01/">
   <CreateDBSnapshotResult>
   {{ snapshot.to_xml() }}

tests/test_rds/test_rds.py

@@ -1214,6 +1214,51 @@ def test_restore_db_instance_from_db_snapshot():
     )
 
 
+@mock_rds
+def test_restore_db_instance_to_point_in_time():
+    conn = boto3.client("rds", region_name=DEFAULT_REGION)
+    conn.create_db_instance(
+        DBInstanceIdentifier="db-primary-1",
+        AllocatedStorage=10,
+        Engine="postgres",
+        DBName="staging-postgres",
+        DBInstanceClass="db.m1.small",
+        MasterUsername="root",
+        MasterUserPassword="hunter2",
+        DBSecurityGroups=["my_sg"],
+    )
+    assert len(conn.describe_db_instances()["DBInstances"]) == 1
+
+    # restore
+    new_instance = conn.restore_db_instance_to_point_in_time(
+        SourceDBInstanceIdentifier="db-primary-1",
+        TargetDBInstanceIdentifier="db-restore-1",
+    )["DBInstance"]
+    assert new_instance["DBInstanceIdentifier"] == "db-restore-1"
+    assert new_instance["DBInstanceClass"] == "db.m1.small"
+    assert new_instance["StorageType"] == "gp2"
+    assert new_instance["Engine"] == "postgres"
+    assert new_instance["DBName"] == "staging-postgres"
+    assert new_instance["DBParameterGroups"][0]["DBParameterGroupName"] == (
+        "default.postgres9.3"
+    )
+    assert new_instance["DBSecurityGroups"] == [
+        {"DBSecurityGroupName": "my_sg", "Status": "active"}
+    ]
+    assert new_instance["Endpoint"]["Port"] == 5432
+
+    # Verify it exists
+    assert len(conn.describe_db_instances()["DBInstances"]) == 2
+    assert (
+        len(
+            conn.describe_db_instances(DBInstanceIdentifier="db-restore-1")[
+                "DBInstances"
+            ]
+        )
+        == 1
+    )
+
+
 @mock_rds
 def test_restore_db_instance_from_db_snapshot_and_override_params():
     conn = boto3.client("rds", region_name=DEFAULT_REGION)