ResourceGroupsTagging support for ELB (#7322)

This commit is contained in:
afuentes-19 2024-02-15 14:22:59 -05:00 committed by GitHub
parent 2e9b903ab6
commit 44487820d7
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 105 additions and 13 deletions

View File

@ -340,6 +340,24 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
"Tags": tags, "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 # TODO add these to the keys and values functions / combine functions
# ELB, resource type elasticloadbalancing:loadbalancer # ELB, resource type elasticloadbalancing:loadbalancer
if ( if (
@ -347,14 +365,14 @@ class ResourceGroupsTaggingAPIBackend(BaseBackend):
or "elasticloadbalancing" in resource_type_filters or "elasticloadbalancing" in resource_type_filters
or "elasticloadbalancing:loadbalancer" 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( tags = self.elbv2_backend.tagging_service.list_tags_for_resource(
elb.arn elbv2.arn
)["Tags"] )["Tags"]
if not tag_filter(tags): # Skip if no tags, or invalid filter if not tag_filter(tags): # Skip if no tags, or invalid filter
continue continue
yield {"ResourceARN": f"{elb.arn}", "Tags": tags} yield {"ResourceARN": f"{elbv2.arn}", "Tags": tags}
# ELB Target Group, resource type elasticloadbalancing:targetgroup # ELB Target Group, resource type elasticloadbalancing:targetgroup
if ( if (

View File

@ -15,7 +15,11 @@ alternative_service_names = {"lambda": "awslambda"}
def get_moto_implementation(service_name): def get_moto_implementation(service_name):
try: try:
backends = get_backend(service_name) 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 return backend, service_name
except ModuleNotFoundError: except ModuleNotFoundError:
return None, service_name return None, service_name
@ -24,9 +28,9 @@ def get_moto_implementation(service_name):
def get_module_name(o): def get_module_name(o):
klass = o.__class__ klass = o.__class__
module = klass.__module__ module = klass.__module__
if module == 'builtins': if module == "builtins":
return klass.__qualname__ # avoid outputs like 'builtins.str' return klass.__qualname__ # avoid outputs like 'builtins.str'
return module + '.' + klass.__qualname__ return module + "." + klass.__qualname__
def calculate_extended_implementation_coverage(): def calculate_extended_implementation_coverage():
@ -161,7 +165,9 @@ def write_implementation_coverage_to_file(coverage):
def write_implementation_coverage_to_docs(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 # rewrite the implementation coverage file with updated values
# try deleting the implementation coverage file # try deleting the implementation coverage file
try: try:
@ -179,7 +185,9 @@ def write_implementation_coverage_to_docs(coverage):
not_implemented = coverage.get(service_name)["not_implemented"] not_implemented = coverage.get(service_name)["not_implemented"]
operations = sorted(list(implemented.keys()) + 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(" ", "_") shorthand = service_name.replace(" ", "_")
with open(service_coverage_file, "w+") as file: with open(service_coverage_file, "w+") as file:
@ -202,7 +210,9 @@ def write_implementation_coverage_to_docs(coverage):
if coverage[service_name]["docs"]: if coverage[service_name]["docs"]:
# Only show auto-generated documentation if it exists # 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("\n\n")
file.write("|start-h3| Implemented features for this service |end-h3|\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".format(op))
file.write("\n") file.write("\n")
with open(implementation_coverage_file, "w+") as file: with open(implementation_coverage_file, "w+") as file:
file.write(".. _implemented_services:\n") file.write(".. _implemented_services:\n")
file.write("\n") file.write("\n")
@ -227,7 +236,9 @@ def write_implementation_coverage_to_docs(coverage):
file.write("Implemented Services\n") file.write("Implemented Services\n")
file.write("====================\n") file.write("====================\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("\n")
file.write(".. toctree::\n") file.write(".. toctree::\n")

View File

@ -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 S3bucket_path, as the functionality is covered in the S3 service
# Ignore neptune, as it shares a URL with RDS # Ignore neptune, as it shares a URL with RDS
# Ignore OpenSearch, as it shares a URL with ElasticSearch # 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(): def iter_backend_url_patterns():

View File

@ -757,3 +757,53 @@ def test_tag_resources_for_unknown_service():
missing_resources["arn:aws:service_x"]["ErrorCode"] missing_resources["arn:aws:service_x"]["ErrorCode"]
== "InternalServiceException" == "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"]
)