Repository: getmoto/moto
Pull request #7337 (2024-02-12 15:45:18)

added resourcegroupstaggingapi support for sns


Patch:
diff --git a/moto/resourcegroupstaggingapi/models.py b/moto/resourcegroupstaggingapi/models.py
index c54af77a2..db2002a96 100644
--- a/moto/resourcegroupstaggingapi/models.py
+++ b/moto/resourcegroupstaggingapi/models.py
@@ -19,6 +19,7 @@ from moto.moto_api._internal import mock_random
 from moto.rds.models import RDSBackend, rds_backends
 from moto.redshift.models import RedshiftBackend, redshift_backends
 from moto.s3.models import S3Backend, s3_backends
+from moto.sns.models import SNSBackend, sns_backends
 from moto.sqs.models import SQSBackend, sqs_backends
 from moto.utilities.tagging_service import TaggingService
 
@@ -97,6 +98,10 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
     def acm_backend(self) -> AWSCertificateManagerBackend:
         return acm_backends[self.account_id][self.region_name]
 
+    @property
+    def sns_backend(self) -> SNSBackend:
+        return sns_backends[self.account_id][self.region_name]
+
     @property
     def sqs_backend(self) -> SQSBackend:
         return sqs_backends[self.account_id][self.region_name]
@@ -340,6 +345,24 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
                     "Tags": tags,
                 }
 
+        # ELB (Classic Load Balancers)
+        if (
+            not resource_type_filters
+            or "elb" in resource_type_filters
+            or "elb:loadbalancer" in resource_type_filters
+        ):
+            for elb in self.elb_backend.load_balancers.values():
+                tags = format_tags(elb.tags)
+                if not tags or not tag_filter(
+                    tags
+                ):  # Skip if no tags, or invalid filter
+                    continue
+
+                yield {
+                    "ResourceARN": f"arn:aws:elasticloadbalancing:{self.region_name}:{self.account_id}:loadbalancer/{elb.name}",
+                    "Tags": tags,
+                }
+
         # TODO add these to the keys and values functions / combine functions
         # ELB, resource type elasticloadbalancing:loadbalancer
         if (
@@ -347,14 +370,14 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
             or "elasticloadbalancing" in resource_type_filters
             or "elasticloadbalancing:loadbalancer" in resource_type_filters
         ):
-            for elb in self.elbv2_backend.load_balancers.values():
+            for elbv2 in self.elbv2_backend.load_balancers.values():
                 tags = self.elbv2_backend.tagging_service.list_tags_for_resource(
-                    elb.arn
+                    elbv2.arn
                 )["Tags"]
                 if not tag_filter(tags):  # Skip if no tags, or invalid filter
                     continue
 
-                yield {"ResourceARN": f"{elb.arn}", "Tags": tags}
+                yield {"ResourceARN": f"{elbv2.arn}", "Tags": tags}
 
         # ELB Target Group, resource type elasticloadbalancing:targetgroup
         if (
@@ -474,6 +497,17 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
 
                 yield {"ResourceARN": f"{queue.queue_arn}", "Tags": tags}
 
+        # SNS
+        if not resource_type_filters or "sns" in resource_type_filters:
+
+            for topic in self.sns_backend.topics.values():
+                tags = format_tags(topic._tags)
+                if not tags or not tag_filter(
+                    tags
+                ):  # Skip if no tags, or invalid filter
+                    continue
+                yield {"ResourceARN": f"{topic.arn}", "Tags": tags}
+
         # VPC
         if (
             not resource_type_filters
diff --git a/scripts/implementation_coverage.py b/scripts/implementation_coverage.py
index 8e4ad0863..05adab0b8 100755
--- a/scripts/implementation_coverage.py
+++ b/scripts/implementation_coverage.py
@@ -15,7 +15,11 @@ alternative_service_names = {"lambda": "awslambda"}
 def get_moto_implementation(service_name):
     try:
         backends = get_backend(service_name)
-        backend = backends[0]["us-east-1"] if "us-east-1" in backends[0] else backends[0]["global"]
+        backend = (
+            backends[0]["us-east-1"]
+            if "us-east-1" in backends[0]
+            else backends[0]["global"]
+        )
         return backend, service_name
     except ModuleNotFoundError:
         return None, service_name
@@ -24,9 +28,9 @@ def get_moto_implementation(service_name):
 def get_module_name(o):
     klass = o.__class__
     module = klass.__module__
-    if module == 'builtins':
-        return klass.__qualname__ # avoid outputs like 'builtins.str'
-    return module + '.' + klass.__qualname__
+    if module == "builtins":
+        return klass.__qualname__  # avoid outputs like 'builtins.str'
+    return module + "." + klass.__qualname__
 
 
 def calculate_extended_implementation_coverage():
@@ -161,7 +165,9 @@ def write_implementation_coverage_to_file(coverage):
 
 
 def write_implementation_coverage_to_docs(coverage):
-    implementation_coverage_file = "{}/../docs/docs/services/index.rst".format(script_dir)
+    implementation_coverage_file = "{}/../docs/docs/services/index.rst".format(
+        script_dir
+    )
     # rewrite the implementation coverage file with updated values
     # try deleting the implementation coverage file
     try:
@@ -179,7 +185,9 @@ def write_implementation_coverage_to_docs(coverage):
         not_implemented = coverage.get(service_name)["not_implemented"]
         operations = sorted(list(implemented.keys()) + not_implemented)
 
-        service_coverage_file = "{}/../docs/docs/services/{}.rst".format(script_dir, service_name)
+        service_coverage_file = "{}/../docs/docs/services/{}.rst".format(
+            script_dir, service_name
+        )
         shorthand = service_name.replace(" ", "_")
 
         with open(service_coverage_file, "w+") as file:
@@ -202,7 +210,9 @@ def write_implementation_coverage_to_docs(coverage):
 
             if coverage[service_name]["docs"]:
                 # Only show auto-generated documentation if it exists
-                file.write(".. autoclass:: " + coverage[service_name].get("module_name"))
+                file.write(
+                    ".. autoclass:: " + coverage[service_name].get("module_name")
+                )
                 file.write("\n\n")
 
             file.write("|start-h3| Implemented features for this service |end-h3|\n\n")
@@ -217,7 +227,6 @@ def write_implementation_coverage_to_docs(coverage):
                     file.write("- [ ] {}\n".format(op))
             file.write("\n")
 
-
     with open(implementation_coverage_file, "w+") as file:
         file.write(".. _implemented_services:\n")
         file.write("\n")
@@ -227,7 +236,9 @@ def write_implementation_coverage_to_docs(coverage):
         file.write("Implemented Services\n")
         file.write("====================\n")
         file.write("\n")
-        file.write("Please see a list of all currently supported services. Each service will have a list of the endpoints that are implemented.\n")
+        file.write(
+            "Please see a list of all currently supported services. Each service will have a list of the endpoints that are implemented.\n"
+        )
 
         file.write("\n")
         file.write(".. toctree::\n")
diff --git a/scripts/update_backend_index.py b/scripts/update_backend_index.py
index e46ea9dc6..c956d61db 100755
--- a/scripts/update_backend_index.py
+++ b/scripts/update_backend_index.py
@@ -19,7 +19,20 @@ output_path = os.path.join(script_dir, "..", output_file)
 # Ignore S3bucket_path, as the functionality is covered in the S3 service
 # Ignore neptune, as it shares a URL with RDS
 # Ignore OpenSearch, as it shares a URL with ElasticSearch
-IGNORE_BACKENDS = ["moto_server", "moto_proxy", "apigatewayv2", "awslambda_simple", "batch_simple", "core", "dynamodb_v20111205", "packages", "utilities", "s3bucket_path", "neptune", "opensearch"]
+IGNORE_BACKENDS = [
+    "moto_server",
+    "moto_proxy",
+    "apigatewayv2",
+    "awslambda_simple",
+    "batch_simple",
+    "core",
+    "dynamodb_v20111205",
+    "packages",
+    "utilities",
+    "s3bucket_path",
+    "neptune",
+    "opensearch",
+]
 
 
 def iter_backend_url_patterns():


Test patch:
diff --git a/tests/test_resourcegroupstaggingapi/test_resourcegroupstaggingapi.py b/tests/test_resourcegroupstaggingapi/test_resourcegroupstaggingapi.py
index e01291123..4034621d9 100644
--- a/tests/test_resourcegroupstaggingapi/test_resourcegroupstaggingapi.py
+++ b/tests/test_resourcegroupstaggingapi/test_resourcegroupstaggingapi.py
@@ -745,6 +745,21 @@ def test_get_resources_sqs():
     assert {"Key": "Test", "Value": "1"} in resp["ResourceTagMappingList"][0]["Tags"]
 
 
+@mock_aws
+def test_get_resources_sns():
+    sns = boto3.client("sns", region_name="us-east-1")
+
+    sns.create_topic(Name="test", Tags=[{"Key": "Shape", "Value": "Square"}])
+
+    rtapi = boto3.client("resourcegroupstaggingapi", region_name="us-east-1")
+    resp = rtapi.get_resources(ResourceTypeFilters=["sns"])
+
+    assert len(resp["ResourceTagMappingList"]) == 1
+    assert {"Key": "Shape", "Value": "Square"} in resp["ResourceTagMappingList"][0][
+        "Tags"
+    ]
+
+
 @mock_aws
 def test_tag_resources_for_unknown_service():
     rtapi = boto3.client("resourcegroupstaggingapi", region_name="us-west-2")
@@ -757,3 +772,53 @@ def test_tag_resources_for_unknown_service():
         missing_resources["arn:aws:service_x"]["ErrorCode"]
         == "InternalServiceException"
     )
+
+
+@mock_aws
+def test_get_resources_elb():
+    client = boto3.client("elb", region_name="us-east-1")
+    lb_name = "my-lb"
+    lb2_name = "second-lb"
+    client.create_load_balancer(
+        LoadBalancerName=lb_name,
+        Listeners=[
+            {"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080},
+            {"Protocol": "http", "LoadBalancerPort": 81, "InstancePort": 9000},
+        ],
+        Scheme="internal",
+    )
+    client.add_tags(
+        LoadBalancerNames=[lb_name],
+        Tags=[
+            {"Key": "burger", "Value": "krabby patty"},
+        ],
+    )
+    client.create_load_balancer(
+        LoadBalancerName=lb2_name,
+        Listeners=[
+            {"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080},
+            {"Protocol": "http", "LoadBalancerPort": 81, "InstancePort": 9000},
+        ],
+        Scheme="internal",
+    )
+    client.add_tags(
+        LoadBalancerNames=[lb2_name],
+        Tags=[
+            {"Key": "color", "Value": "blue"},
+            {"Key": "shape", "Value": "square"},
+        ],
+    )
+    rgta_client = boto3.client("resourcegroupstaggingapi", region_name="us-east-1")
+    resources_no_filter = rgta_client.get_resources(
+        ResourceTypeFilters=["elb"],
+    )
+    assert len(resources_no_filter["ResourceTagMappingList"]) == 2
+
+    resources_burger_filter = rgta_client.get_resources(
+        TagFilters=[{"Key": "burger", "Values": ["krabby patty"]}]
+    )
+    assert len(resources_burger_filter["ResourceTagMappingList"]) == 1
+    assert (
+        f"arn:aws:elasticloadbalancing:us-east-1:123456789012:loadbalancer/{lb_name}"
+        == resources_burger_filter["ResourceTagMappingList"][0]["ResourceARN"]
+    )