Repository: getmoto/moto

Issue:
RDS create_db_instance throws KeyError if cluster doesn't exist on 5.1.0
We have some code that just creates a database instance. It started throwing a `KeyError` on the 5.1.0 release when calling create_db_instance:

Our code:

```
        client = boto3.client("rds", region_name="us-west-2")

        client.create_db_instance(
            DBClusterIdentifier="cluster_id",
            DBInstanceClass="r6i.large",
            DBInstanceIdentifier="cluster_id-0",
            Engine="aurora-postgresql",
            Tags=[],
        )
```

Stack trace:

```
  File "test.py", line 15, in setUp
    client.create_db_instance(
<redacted retry middleware noise>
  File "/lib/python3.11/site-packages/moto/rds/responses.py", line 80, in create_db_instance
    database = self.backend.create_db_instance(db_kwargs)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/lib/python3.11/site-packages/moto/rds/models.py", line 2181, in create_db_instance
    database = DBInstanceClustered(backend=self, **db_kwargs)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/lib/python3.11/site-packages/moto/rds/models.py", line 1503, in __init__
    self.cluster = self.backend.clusters[db_cluster_identifier]
                   ~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^
KeyError: 'cluster_id'
```

Our code was obviously wrong, but earlier versions of moto actually worked with it for some reason. I would think this should throw a more useful exception, though, like whatever boto will throw when it encounters this situation of someone creating an instance on a nonexistent cluster.

Hints from issue comments:
@bpandola: Thanks for catching this.  I had intended to add this check with some recent changes I made, but obviously missed it.  I will raise a PR shortly to correct that.

Solution patch:
diff --git a/moto/rds/exceptions.py b/moto/rds/exceptions.py
index b8457254d..f1092c753 100644
--- a/moto/rds/exceptions.py
+++ b/moto/rds/exceptions.py
@@ -1,3 +1,6 @@
+from typing import Optional
+
+
 class RDSClientError(Exception):
     def __init__(self, code: str, message: str):
         super().__init__(message)
@@ -146,10 +149,10 @@ class InvalidDBClusterStateFault(RDSClientError):
 
 
 class DBClusterNotFoundError(RDSClientError):
-    def __init__(self, cluster_identifier: str):
-        super().__init__(
-            "DBClusterNotFoundFault", f"DBCluster {cluster_identifier} not found."
-        )
+    def __init__(self, cluster_identifier: str, message: Optional[str] = None):
+        if message is None:
+            message = f"DBCluster {cluster_identifier} not found."
+        super().__init__("DBClusterNotFoundFault", message)
 
 
 class DBClusterSnapshotNotFoundError(RDSClientError):
diff --git a/moto/rds/models.py b/moto/rds/models.py
index 7e9bfb7f1..54250aa9b 100644
--- a/moto/rds/models.py
+++ b/moto/rds/models.py
@@ -1509,6 +1509,11 @@ class DBInstanceClustered(DBInstance):
         self, db_cluster_identifier: str, promotion_tier: int = 1, **kwargs: Any
     ) -> None:
         super().__init__(db_cluster_identifier=db_cluster_identifier, **kwargs)
+        if db_cluster_identifier not in self.backend.clusters:
+            raise DBClusterNotFoundError(
+                db_cluster_identifier,
+                f"The source cluster could not be found or cannot be accessed: {db_cluster_identifier}",
+            )
         self.cluster = self.backend.clusters[db_cluster_identifier]
         self.db_cluster_identifier: str = db_cluster_identifier
         self.is_cluster_writer = True if not self.cluster.members else False


Test patch:
diff --git a/tests/test_rds/test_rds_clusters_with_instances.py b/tests/test_rds/test_rds_clusters_with_instances.py
index 91d25358a..4e6a24c93 100644
--- a/tests/test_rds/test_rds_clusters_with_instances.py
+++ b/tests/test_rds/test_rds_clusters_with_instances.py
@@ -128,3 +128,20 @@ def test_delete_db_cluster_fails_if_cluster_contains_db_instances():
         SkipFinalSnapshot=True,
     )["DBCluster"]
     assert cluster["DBClusterIdentifier"] == cluster_identifier
+
+
+@mock_aws
+def test_create_instance_with_non_existent_cluster_raises_error():
+    client = boto3.client("rds", "us-east-1")
+    cluster_identifier = "non-existent-cluster-id"
+    with pytest.raises(ClientError) as exc:
+        client.create_db_instance(
+            DBClusterIdentifier=cluster_identifier,
+            Engine="aurora-postgresql",
+            DBInstanceIdentifier="test-instance",
+            DBInstanceClass="db.t4g.medium",
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "DBClusterNotFoundFault"
+    assert cluster_identifier in err["Message"]
+    assert "could not be found" in err["Message"]