8000 add resourcegroups validated tests by bentsku · Pull Request #9140 · localstack/localstack · GitHub
[go: up one dir, main page]
Skip to content

add resourcegroups validated tests #9140

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Sep 13, 2023
Merged

add resourcegroups validated tests #9140

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
45644da
validate existing test + add snapshot
bentsku Sep 13, 2023
ef8ffe1
implement ListGroupResources tests and SearchResources
bentsku Sep 13, 2023
bd84478
skipif instead of xfail
bentsku Sep 13, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
355 changes: 346 additions & 9 deletions tests/aws/services/resource_groups/test_resource_groups.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,15 +1,87 @@
import contextlib
import json
import os

import pytest
from botocore.exceptions import ClientError

from localstack.constants import TEST_AWS_REGION_NAME
from localstack.testing.aws.util import is_aws_cloud
from localstack.testing.pytest import markers
from localstack.utils.common import short_uid
from localstack.utils.sync import retry


@pytest.fixture
def resourcegroups_create_group(aws_client):
groups = []

def _create_group(**kwargs):
response = aws_client.resource_groups.create_group(**kwargs)
groups.append(response["Group"]["Name"])
return response

yield _create_group

for group_name in groups:
with contextlib.suppress(
ClientError, KeyError
): # adding KeyError to the list because Moto has a bug
aws_client.resource_groups.delete_group(GroupName=group_name)


@pytest.fixture
def sqs_create_queue_in_region(aws_client_factory):
queue_urls = {}

def factory(region, **kwargs):
if "QueueName" not in kwargs:
kwargs["QueueName"] = "test-queue-%s" % short_uid()
response = aws_client_factory(region_name=region).sqs.create_queue(**kwargs)
url = response["QueueUrl"]
queue_urls.setdefault(region, []).append(url)

return url

yield factory

# cleanup
for queues_region, queue_urls in queue_urls.items():
sqs_client = aws_client_factory(region_name=queues_region).sqs
for queue_url in queue_urls:
with contextlib.suppress(ClientError):
sqs_client.delete_queue(QueueUrl=queue_url)


@pytest.fixture(autouse=True)
def resource_groups_snapshot_transformers(snapshot):
snapshot.add_transformers_list(
[
snapshot.transform.key_value("Name"),
snapshot.transform.key_value("NextToken"),
]
)


@markers.snapshot.skip_snapshot_verify(
paths=[
"$..GroupArn", # Moto is always returning the region as `us-west-1`, seems to be hard-coded
"$..GroupConfiguration",
"$..NextToken",
]
)
class TestResourceGroups:
@markers.aws.unknown
def test_create_group(self, aws_client):
resource_group_client = aws_client.resource_groups
name = "resource_group-{}".format(short_uid())
response = resource_group_client.create_group(
"""
List of Services integrations with resourcegroups
see: https://docs.aws.amazon.com/ARG/latest/userguide/integrated-services-list.html
List of supported resources:
see: https://docs.aws.amazon.com/ARG/latest/userguide/supported-resources.html
"""

@markers.aws.validated
def test_create_group(self, aws_client, resourcegroups_create_group, snapshot):
name = f"resource_group-{short_uid()}"
response = resourcegroups_create_group(
Name=name,
Description="description",
ResourceQuery={
Expand All @@ -28,20 +100,285 @@ def test_create_group(self, aws_client):
},
Tags={"resource_group_tag_key": "resource_group_tag_value"},
)
snapshot.match("create-group", response)
assert name == response["Group"]["Name"]
assert "TAG_FILTERS_1_0" == response["ResourceQuery"]["Type"]
assert "resource_group_tag_value" == response["Tags"]["resource_group_tag_key"]

response = resource_group_client.get_group(GroupName=name)
response = aws_client.resource_groups.get_group(GroupName=name)
snapshot.match("get-group", response)
assert "description" == response["Group"]["Description"]

response = resource_group_client.list_groups()
response = aws_client.resource_groups.list_groups()
snapshot.match("list-groups", response)
assert 1 == len(response["GroupIdentifiers"])
assert 1 == len(response["Groups"])

response = resource_group_client.delete_group(GroupName=name)
response = aws_client.resource_groups.delete_group(GroupName=name)
snapshot.match("delete-group", response)
assert name == response["Group"]["Name"]

response = resource_group_client.list_groups()
response = aws_client.resource_groups.list_groups()
snapshot.match("list-groups-after-delete", response)
assert 0 == len(response["GroupIdentifiers"])
assert 0 == len(response["Groups"])

@markers.aws.validated
@pytest.mark.skipif(
condition=not is_aws_cloud(), reason="Not implemented in moto (ListGroupResources)"
)
def test_resource_groups_tag_query(
self, aws_client, snapshot, resourcegroups_create_group, s3_bucket, sqs_create_queue
):
snapshot.add_transformer(snapshot.transform.resource_name())
group_name = f"resource_group-{short_uid()}"
response = resourcegroups_create_group(
Name=group_name,
Description="test-tag-query",
ResourceQuery={
"Type": "TAG_FILTERS_1_0",
"Query": json.dumps(
{
"ResourceTypeFilters": ["AWS::AllSupported"],
"TagFilters": [
{
"Key": "Stage",
"Values": ["test-resource-group"],
}
],
}
),
},
Tags={"GroupTag": "GroupTag1"},
)
snapshot.match("create-group", response)

response = aws_client.resource_groups.list_group_resources(Group=group_name)
snapshot.match("list-group-resources-empty", response)

# create SQS queue
tagged_queue_url = sqs_create_queue()
# tag queue
tags = {"Stage": "test-resource-group"}
aws_client.sqs.tag_queue(QueueUrl=tagged_queue_url, Tags=tags)

aws_client.s3.put_bucket_tagging(
Bucket=s3_bucket, Tagging={"TagSet": [{"Key": "Stage", "Value": "test-resource-group"}]}
)

not_tagged_queue_url = sqs_create_queue()
tags = {"Stage": "not-part-resource-group"}
aws_client.sqs.tag_queue(QueueUrl=not_tagged_queue_url, Tags=tags)

response = aws_client.resource_groups.list_group_resources(Group=group_name)
snapshot.match("list-group-resources", response)

queue_tags = aws_client.sqs.list_queue_tags(QueueUrl=tagged_queue_url)
snapshot.match("get-queue-tags", queue_tags)

aws_client.sqs.untag_queue(QueueUrl=tagged_queue_url, TagKeys=["Stage"])

def _get_group_resources():
_response = aws_client.resource_groups.list_group_resources(Group=group_name)
assert len(response["Resources"]) == 1
return _response

response = retry(_get_group_resources, retries=3, sleep=1)
snapshot.match("list-group-resources-after-queue-removal", response)

@markers.aws.validated
@pytest.mark.skipif(
condition=not is_aws_cloud(), reason="Not implemented in moto (ListGroupResources)"
)
def test_resource_groups_different_region(
self, aws_client_factory, snapshot, resourcegroups_create_group, sqs_create_queue_in_region
):
"""Resource groups can only have resources from the same Region, the one of the group"""
region_1 = TEST_AWS_REGION_NAME
region_2 = "us-east-2"
resourcegroups_client = aws_client_factory(region_name=region_1).resource_groups
snapshot.add_transformer(snapshot.transform.resource_name())
group_name = f"resource_group-{short_uid()}"
response = resourcegroups_create_group(
Name=group_name,
Description="test-tag-query",
ResourceQuery={
"Type": "TAG_FILTERS_1_0",
"Query": json.dumps(
{
"ResourceTypeFilters": ["AWS::AllSupported"],
"TagFilters": [
{
"Key": "Stage",
"Values": ["test-resource-group"],
}
],
}
),
},
Tags={"GroupTag": "GroupTag1"},
)
snapshot.match("create-group", response)

# create 2 SQS queues in different regions with tags
tags = {"Stage": "test-resource-group"}
sqs_create_queue_in_region(region=region_1, tags=tags)
sqs_create_queue_in_region(region=region_2, tags=tags)

response = resourcegroups_client.list_group_resources(Group=group_name)
snapshot.match("list-group-resources", response)

@markers.aws.validated
@pytest.mark.skipif(
condition=not is_aws_cloud(), reason="Not implemented in moto (ListGroupResources)"
)
def test_resource_type_filters(
self, aws_client, snapshot, resourcegroups_create_group, s3_bucket, sqs_create_queue
):
"""Resource group can filter with a ResourceType, like `AWS::S3::Bucket`"""
snapshot.add_transformer(snapshot.transform.resource_name())
group_name = f"resource_group-{short_uid()}"
response = resourcegroups_create_group(
Name=group_name,
Description="test-tag-query",
ResourceQuery={
"Type": "TAG_FILTERS_1_0",
"Query": json.dumps(
{
"ResourceTypeFilters": ["AWS::S3::Bucket"],
"TagFilters": [
{
"Key": "Stage",
"Values": ["test-resource-group"],
}
],
}
),
},
Tags={"GroupTag": "GroupTag1"},
)
snapshot.match("create-group", response)

# create SQS queue with tags
sqs_create_queue(tags={"Stage": "test-resource-group"})

aws_client.s3.put_bucket_tagging(
Bucket=s3_bucket, Tagging={"TagSet": [{"Key": "Stage", "Value": "test-resource-group"}]}
)

response = aws_client.resource_groups.list_group_resources(Group=group_name)
snapshot.match("list-group-resources", response)

@markers.aws.validated
@pytest.mark.skipif(
condition=not is_aws_cloud(), reason="Not implemented in moto (ListGroupResources)"
)
def test_cloudformation_query(
self, aws_client, deploy_cfn_template, snapshot, resourcegroups_create_group
):
snapshot.add_transformers_list(
[
snapshot.transform.key_value("StackIdentifier"),
snapshot.transform.resource_name(),
]
)
stack = deploy_cfn_template(
template_path=os.path.join(
os.path.dirname(__file__), "../../templates/deploy_template_2.yaml"
),
parameters={"CompanyName": "MyCompany", "MyEmail1": "my@email.com"},
)
assert len(stack.outputs) == 3
topic_arn = stack.outputs["MyTopic"]

group_name = f"resource_group-{short_uid()}"
response = resourcegroups_create_group(
Name=group_name,
Description="test-cfn-query",
ResourceQuery={
"Type": "CLOUDFORMATION_STACK_1_0",
"Query": json.dumps(
{
"ResourceTypeFilters": ["AWS::AllSupported"],
"StackIdentifier": stack.stack_id,
}
),
},
)
snapshot.match("create-group", response)

response = aws_client.resource_groups.list_group_resources(Group=group_name)
snapshot.match("list-group-resources", response)

assert topic_arn in [
resource["ResourceArn"] for resource in response["ResourceIdentifiers"]
]

stack.destroy()

response = aws_client.resource_groups.list_group_resources(Group=group_name)
snapshot.match("list-group-resources-after-destroy", response)

with pytest.raises(ClientError) as e:
resourcegroups_create_group(
Name="going-to-fail",
Description="test-cfn-query",
ResourceQuery={
"Type": "CLOUDFORMATION_STACK_1_0",
"Query": json.dumps(
{
"ResourceTypeFilters": ["AWS::AllSupported"],
"StackIdentifier": stack.stack_id,
}
),
},
)
snapshot.match("create-group-with-delete-stack", e.value.response)

@markers.aws.validated
@pytest.mark.skipif(
condition=not is_aws_cloud(), reason="Not implemented in moto (SearchResources)"
)
def test_search_resources(self, aws_client, sqs_create_queue, snapshot):
snapshot.add_transformer(snapshot.transform.resource_name())
# create SQS queue with tags
queue_url = sqs_create_queue(tags={"Stage": "test-resource-group"})
queue_tags = aws_client.sqs.list_queue_tags(QueueUrl=queue_url)
snapshot.match("queue-tags", queue_tags)

def _get_resources(resource_types: list[str], expected: int):
_response = aws_client.resource_groups.search_resources(
ResourceQuery={
"Type": "TAG_FILTERS_1_0",
"Query": json.dumps(
{
"ResourceTypeFilters": resource_types,
"TagFilters": [
{
"Key": "Stage",
"Values": ["test-resource-group"],
}
],
}
),
}
)
assert len(_response["ResourceIdentifiers"]) == expected
return _response

retries = 10 if is_aws_cloud() else 3
sleep = 1 if is_aws_cloud() else 0.1

response = retry(
_get_resources,
resource_types=["AWS::AllSupported"],
expected=1,
retries=retries,
sleep=sleep,
)
snapshot.match("list-group-resources-sqs", response)

response = retry(
_get_resources, resource_types=["AWS::S3::Bucket"], expected=0, retries=1, sleep=1
)
snapshot.match("list-group-resources-s3", response)
Loading
0