DSQL: implement get_cluster()

moto/dsql/exceptions.py

@@ -1 +1,12 @@
 """Exceptions raised by the dsql service."""
+
+from moto.core.exceptions import JsonRESTError
+
+
+class ValidationException(JsonRESTError):
+    """Tag validation failed."""
+
+    code = 400
+
+    def __init__(self, message: str):
+        super().__init__("ValidationException", message)

moto/dsql/models.py

@@ -10,6 +10,8 @@
 from moto.moto_api._internal.managed_state_model import ManagedState
 from moto.utilities.utils import get_partition
 
+from .exceptions import ValidationException
+
 
 class Cluster(BaseModel, ManagedState):
     """Model for an AuroraDSQL cluster."""
@@ -68,6 +70,17 @@ def create_cluster(
             tags,
             client_token,
         )
+        self.clusters[cluster.identifier] = cluster
+        return cluster
+
+    def get_cluster(
+        self,
+        identifier: str,
+    ) -> Cluster:
+        cluster = self.clusters.get(identifier)
+        if cluster is None:
+            raise ValidationException("invalid Cluster Id")
+
         return cluster
 
 

moto/dsql/responses.py

@@ -30,3 +30,9 @@ def create_cluster(self) -> str:
         )
 
         return json.dumps(dict(cluster.to_dict()))
+
+    def get_cluster(self) -> str:
+        identifier = self.path.split("/")[-1]
+        cluster = self.dsql_backend.get_cluster(identifier=identifier)
+
+        return json.dumps(dict(cluster.to_dict()))

moto/dsql/urls.py

@@ -8,4 +8,5 @@
 
 url_paths = {
     "{0}/cluster$": AuroraDSQLResponse.dispatch,
+    "{0}/cluster/(?P<identifier>[^/]+)$": AuroraDSQLResponse.dispatch,
 }

tests/test_dsql/test_dsql.py

@@ -3,15 +3,18 @@
 from datetime import datetime
 
 import boto3
+from botocore.exceptions import ClientError
 from dateutil.tz import tzutc
 from freezegun import freeze_time
 
 from moto import mock_aws, settings
 
+TEST_REGION = "us-east-1"
+
 
 @mock_aws
 def test_create_cluster():
-    client = boto3.client("dsql", region_name="us-east-1")
+    client = boto3.client("dsql", region_name=TEST_REGION)
     with freeze_time("2024-12-22 12:34:00"):
         resp = client.create_cluster()
 
@@ -22,3 +25,37 @@ def test_create_cluster():
     assert resp["status"] == "CREATING"
     if not settings.TEST_SERVER_MODE:
         assert resp["creationTime"] == datetime(2024, 12, 22, 12, 34, tzinfo=tzutc())
+
+
+@mock_aws
+def test_get_invalid_cluster():
+    client = boto3.client("dsql", region_name=TEST_REGION)
+
+    try:
+        client.get_cluster(identifier="invalid")
+    except ClientError as err:
+        assert err.response["Error"]["Code"] == "ValidationException"
+        assert err.response["Error"]["Message"] == "invalid Cluster Id"
+
+
+@mock_aws
+def test_get_cluster():
+    client = boto3.client("dsql", region_name=TEST_REGION)
+    with freeze_time("2024-12-22 12:34:00"):
+        resp = client.create_cluster()
+
+    identifier = resp["identifier"]
+
+    get_resp = client.get_cluster(identifier=identifier)
+
+    # TODO Add `witnessRegion` and `linkedClusterArns` when implement create-multi-region-clusters
+    assert get_resp["identifier"] == identifier
+    assert (
+        get_resp["arn"] == f"arn:aws:dsql:us-east-1:123456789012:cluster/{identifier}"
+    )
+    assert get_resp["deletionProtectionEnabled"] is True
+    assert get_resp["status"] == "CREATING"
+    if not settings.TEST_SERVER_MODE:
+        assert get_resp["creationTime"] == datetime(
+            2024, 12, 22, 12, 34, tzinfo=tzutc()
+        )