72 lines
2.8 KiB
Python
72 lines
2.8 KiB
Python
import unittest
|
|
|
|
import boto3
|
|
|
|
from moto import mock_rds, mock_resourcegroupstaggingapi
|
|
|
|
|
|
@mock_rds
|
|
@mock_resourcegroupstaggingapi
|
|
class TestRdsTagging(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.rds = boto3.client("rds", region_name="us-west-2")
|
|
self.rtapi = boto3.client("resourcegroupstaggingapi", region_name="us-west-2")
|
|
self.resources_tagged = []
|
|
self.resources_untagged = []
|
|
for i in range(3):
|
|
database = self.rds.create_db_instance(
|
|
DBInstanceIdentifier=f"db-instance-{i}",
|
|
Engine="postgres",
|
|
DBInstanceClass="db.m1.small",
|
|
CopyTagsToSnapshot=bool(i),
|
|
Tags=[{"Key": "test", "Value": f"value-{i}"}] if i else [],
|
|
).get("DBInstance")
|
|
snapshot = self.rds.create_db_snapshot(
|
|
DBInstanceIdentifier=database["DBInstanceIdentifier"],
|
|
DBSnapshotIdentifier=f"snapshot-{i}",
|
|
).get("DBSnapshot")
|
|
group = self.resources_tagged if i else self.resources_untagged
|
|
group.append(database["DBInstanceArn"])
|
|
group.append(snapshot["DBSnapshotArn"])
|
|
|
|
def test_get_resources_rds(self):
|
|
def assert_response(response, expected_count, resource_type=None):
|
|
results = response.get("ResourceTagMappingList", [])
|
|
assert len(results) == expected_count
|
|
for item in results:
|
|
arn = item["ResourceARN"]
|
|
assert arn in self.resources_tagged
|
|
assert arn not in self.resources_untagged
|
|
if resource_type:
|
|
assert f":{resource_type}:" in arn
|
|
|
|
resp = self.rtapi.get_resources(ResourceTypeFilters=["rds"])
|
|
assert_response(resp, 4)
|
|
resp = self.rtapi.get_resources(ResourceTypeFilters=["rds:db"])
|
|
assert_response(resp, 2, resource_type="db")
|
|
resp = self.rtapi.get_resources(ResourceTypeFilters=["rds:snapshot"])
|
|
assert_response(resp, 2, resource_type="snapshot")
|
|
resp = self.rtapi.get_resources(
|
|
TagFilters=[{"Key": "test", "Values": ["value-1"]}]
|
|
)
|
|
assert_response(resp, 2)
|
|
|
|
def test_tag_resources_rds(self):
|
|
# WHEN
|
|
# we tag resources
|
|
self.rtapi.tag_resources(
|
|
ResourceARNList=self.resources_tagged,
|
|
Tags={"key1": "value1", "key2": "value2"},
|
|
)
|
|
|
|
# THEN
|
|
# we can retrieve the tags using the RDS API
|
|
def get_tags(arn):
|
|
return self.rds.list_tags_for_resource(ResourceName=arn)["TagList"]
|
|
|
|
for arn in self.resources_tagged:
|
|
assert {"Key": "key1", "Value": "value1"} in get_tags(arn)
|
|
assert {"Key": "key2", "Value": "value2"} in get_tags(arn)
|
|
for arn in self.resources_untagged:
|
|
assert get_tags(arn) == []
|