Add list_tags_for_resource for Elasticache (#7463)

moto/elasticache/exceptions.py

@@ -82,3 +82,14 @@ def __init__(self, cache_cluster_id: str):
             "CacheClusterNotFound",
             message=f"Cache cluster {cache_cluster_id} not found.",
         )
+
+
+class InvalidARNFault(ElastiCacheException):
+
+    code = 400
+
+    def __init__(self, arn: str):
+        super().__init__(
+            "InvalidARNFault",
+            message=f"ARN {arn} is invalid.",
+        )

moto/elasticache/models.py

@@ -1,3 +1,4 @@
+from re import compile as re_compile
 from typing import Any, Dict, List, Optional, Tuple
 
 from moto.core.base_backend import BackendDict, BaseBackend
@@ -8,6 +9,7 @@
 from .exceptions import (
     CacheClusterAlreadyExists,
     CacheClusterNotFound,
+    InvalidARNFault,
     UserAlreadyExists,
     UserNotFound,
 )
@@ -78,7 +80,6 @@ def __init__(
     ):
         if tags is None:
             tags = []
-
         self.cache_cluster_id = cache_cluster_id
         self.az_mode = az_mode
         self.preferred_availability_zone = preferred_availability_zone
@@ -121,15 +122,23 @@ def __init__(
         self.cache_cluster_create_time = utcnow()
         self.auth_token_last_modified_date = utcnow()
         self.cache_cluster_status = "available"
-        self.arn = f"arn:aws:elasticache:{region_name}:{account_id}:{cache_cluster_id}"
+        self.arn = (
+            f"arn:aws:elasticache:{region_name}:{account_id}:cluster:{cache_cluster_id}"
+        )
         self.cache_node_id = str(mock_random.uuid4())
 
+    def get_tags(self) -> List[Dict[str, str]]:
+        return self.tags
+
 
 class ElastiCacheBackend(BaseBackend):
     """Implementation of ElastiCache APIs."""
 
     def __init__(self, region_name: str, account_id: str):
         super().__init__(region_name, account_id)
+        self.arn_regex = re_compile(
+            r"^arn:aws:elasticache:.*:[0-9]*:(cluster|snapshot):.*$"
+        )
         self.users = dict()
         self.users["default"] = User(
             account_id=self.account_id,
@@ -331,5 +340,19 @@ def delete_cache_cluster(self, cache_cluster_id: str) -> CacheCluster:
                 return cache_cluster
         raise CacheClusterNotFound(cache_cluster_id)
 
+    def list_tags_for_resource(self, arn: str) -> List[Dict[str, str]]:
+        if self.arn_regex.match(arn):
+            arn_breakdown = arn.split(":")
+            resource_type = arn_breakdown[len(arn_breakdown) - 2]
+            resource_name = arn_breakdown[len(arn_breakdown) - 1]
+            if resource_type == "cluster":
+                if resource_name in self.cache_clusters:
+                    return self.cache_clusters[resource_name].get_tags()
+            else:
+                return []
+        else:
+            raise InvalidARNFault(arn)
+        return []
+
 
 elasticache_backends = BackendDict(ElastiCacheBackend, "elasticache")

moto/elasticache/responses.py

@@ -67,7 +67,7 @@ def create_cache_cluster(self) -> str:
         cache_subnet_group_name = self._get_param("CacheSubnetGroupName")
         cache_security_group_names = self._get_param("CacheSecurityGroupNames")
         security_group_ids = self._get_param("SecurityGroupIds")
-        tags = self._get_param("Tags")
+        tags = (self._get_multi_param_dict("Tags") or {}).get("Tag", [])
         snapshot_arns = self._get_param("SnapshotArns")
         snapshot_name = self._get_param("SnapshotName")
         preferred_maintenance_window = self._get_param("PreferredMaintenanceWindow")
@@ -144,6 +144,12 @@ def delete_cache_cluster(self) -> str:
         template = self.response_template(DELETE_CACHE_CLUSTER_TEMPLATE)
         return template.render(cache_cluster=cache_cluster)
 
+    def list_tags_for_resource(self) -> str:
+        arn = self._get_param("ResourceName")
+        template = self.response_template(LIST_TAGS_FOR_RESOURCE_TEMPLATE)
+        tags = self.elasticache_backend.list_tags_for_resource(arn)
+        return template.render(tags=tags)
+
 
 USER_TEMPLATE = """<UserId>{{ user.id }}</UserId>
     <UserName>{{ user.name }}</UserName>
@@ -545,3 +551,19 @@ def delete_cache_cluster(self) -> str:
 </CacheCluster>
   </DeleteCacheClusterResult>
 </DeleteCacheClusterResponse>"""
+
+LIST_TAGS_FOR_RESOURCE_TEMPLATE = """<ListTagsForResourceResponse xmlns="http://elasticache.amazonaws.com/doc/2015-02-02/">
+  <ListTagsForResourceResult>
+    <TagList>
+    {%- for tag in tags -%}
+      <Tag>
+        <Key>{{ tag['Key'] }}</Key>
+        <Value>{{ tag['Value'] }}</Value>
+      </Tag>
+    {%- endfor -%}
+    </TagList>
+  </ListTagsForResourceResult>
+  <ResponseMetadata>
+    <RequestId>8c21ba39-a598-11e4-b688-194eaf8658fa</RequestId>
+  </ResponseMetadata>
+</ListTagsForResourceResponse>"""

tests/test_elasticache/test_elasticache.py

@@ -417,3 +417,26 @@ def test_delete_unknown_cache_cluster():
     err = exc.value.response["Error"]
     assert err["Code"] == "CacheClusterNotFound"
     assert err["Message"] == f"Cache cluster {cache_cluster_id_unknown} not found."
+
+
+@mock_aws
+def test_list_tags_cache_cluster():
+    conn = boto3.client("elasticache", region_name="ap-southeast-1")
+    result = conn.list_tags_for_resource(
+        ResourceName="arn:aws:elasticache:us-west-2:1234567890:cluster:foo"
+    )
+    assert result["TagList"] == []
+    test_instance = conn.create_cache_cluster(
+        CacheClusterId="test-cache-cluster",
+        Engine="memcached",
+        NumCacheNodes=2,
+        Tags=[{"Key": "foo", "Value": "bar"}, {"Key": "foo1", "Value": "bar1"}],
+        SecurityGroupIds=["sg-1234"],
+    )
+    post_create_result = conn.list_tags_for_resource(
+        ResourceName=test_instance["CacheCluster"]["ARN"],
+    )
+    assert post_create_result["TagList"] == [
+        {"Value": "bar", "Key": "foo"},
+        {"Value": "bar1", "Key": "foo1"},
+    ]