[issue-4360] Fix InvalidResource Id,Type responses for ssm tagging methods (#4361)

This commit is contained in:
Jim King 2021-09-30 07:47:11 -04:00 committed by GitHub
parent f50cf51de7
commit 33e60a2d16
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 116 additions and 3 deletions

View File

@ -23,6 +23,24 @@ class InvalidFilterValue(JsonRESTError):
super(InvalidFilterValue, self).__init__("InvalidFilterValue", message)
class InvalidResourceId(JsonRESTError):
code = 400
def __init__(self):
super(InvalidResourceId, self).__init__(
"InvalidResourceId", "Invalid Resource Id"
)
class InvalidResourceType(JsonRESTError):
code = 400
def __init__(self):
super(InvalidResourceType, self).__init__(
"InvalidResourceType", "Invalid Resource Type"
)
class ParameterNotFound(JsonRESTError):
code = 400

View File

@ -38,6 +38,8 @@ from .exceptions import (
ParameterMaxVersionLimitExceeded,
DocumentPermissionLimit,
InvalidPermissionType,
InvalidResourceId,
InvalidResourceType,
)
@ -835,6 +837,7 @@ class SimpleSystemManagerBackend(BaseBackend):
documents.delete(*keys_to_delete)
if len(documents.versions) == 0:
self._resource_tags.get("Document", {}).pop(name, None)
del self._documents[name]
def get_document(self, name, document_version, version_name, document_format):
@ -1029,6 +1032,7 @@ class SimpleSystemManagerBackend(BaseBackend):
)
def delete_parameter(self, name):
self._resource_tags.get("Parameter", {}).pop(name, None)
return self._parameters.pop(name, None)
def delete_parameters(self, names):
@ -1037,6 +1041,7 @@ class SimpleSystemManagerBackend(BaseBackend):
try:
del self._parameters[name]
result.append(name)
self._resource_tags.get("Parameter", {}).pop(name, None)
except KeyError:
pass
return result
@ -1617,18 +1622,44 @@ class SimpleSystemManagerBackend(BaseBackend):
return version
def add_tags_to_resource(self, resource_type, resource_id, tags):
self._validate_resource_type_and_id(resource_type, resource_id)
for key, value in tags.items():
self._resource_tags[resource_type][resource_id][key] = value
def remove_tags_from_resource(self, resource_type, resource_id, keys):
self._validate_resource_type_and_id(resource_type, resource_id)
tags = self._resource_tags[resource_type][resource_id]
for key in keys:
if key in tags:
del tags[key]
def list_tags_for_resource(self, resource_type, resource_id):
self._validate_resource_type_and_id(resource_type, resource_id)
return self._resource_tags[resource_type][resource_id]
def _validate_resource_type_and_id(self, resource_type, resource_id):
if resource_type == "Parameter":
if resource_id not in self._parameters:
raise InvalidResourceId()
else:
return
elif resource_type == "Document":
if resource_id not in self._documents:
raise InvalidResourceId()
else:
return
elif resource_type not in (
# https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ssm.html#SSM.Client.remove_tags_from_resource
"ManagedInstance",
"MaintenanceWindow",
"PatchBaseline",
"OpsItem",
"OpsMetadata",
):
raise InvalidResourceType()
else:
raise InvalidResourceId()
def send_command(self, **kwargs):
command = Command(
comment=kwargs.get("Comment", ""),

View File

@ -1053,7 +1053,7 @@ def test_describe_parameters_tags():
@mock_ssm
def test_tags_in_list_tags_from_resource():
def test_tags_in_list_tags_from_resource_parameter():
client = boto3.client("ssm", region_name="us-east-1")
client.put_parameter(
@ -1066,9 +1066,32 @@ def test_tags_in_list_tags_from_resource():
tags = client.list_tags_for_resource(
ResourceId="/spam/eggs", ResourceType="Parameter"
)
assert tags.get("TagList") == [{"Key": "spam", "Value": "eggs"}]
client.delete_parameter(Name="/spam/eggs")
with pytest.raises(ClientError) as ex:
client.list_tags_for_resource(ResourceType="Parameter", ResourceId="/spam/eggs")
assert ex.value.response["Error"]["Code"] == "InvalidResourceId"
@mock_ssm
def test_tags_invalid_resource_id():
client = boto3.client("ssm", region_name="us-east-1")
with pytest.raises(ClientError) as ex:
client.list_tags_for_resource(ResourceType="Parameter", ResourceId="bar")
assert ex.value.response["Error"]["Code"] == "InvalidResourceId"
@mock_ssm
def test_tags_invalid_resource_type():
client = boto3.client("ssm", region_name="us-east-1")
with pytest.raises(ClientError) as ex:
client.list_tags_for_resource(ResourceType="foo", ResourceId="bar")
assert ex.value.response["Error"]["Code"] == "InvalidResourceType"
@mock_ssm
def test_get_parameter_invalid():
@ -1632,12 +1655,21 @@ def test_get_parameter_history_missing_parameter():
def test_add_remove_list_tags_for_resource():
client = boto3.client("ssm", region_name="us-east-1")
with pytest.raises(ClientError) as ce:
client.add_tags_to_resource(
ResourceId="test",
ResourceType="Parameter",
Tags=[{"Key": "test-key", "Value": "test-value"}],
)
assert ce.value.response["Error"]["Code"] == "InvalidResourceId"
client.put_parameter(Name="test", Value="value", Type="String")
client.add_tags_to_resource(
ResourceId="test",
ResourceType="Parameter",
Tags=[{"Key": "test-key", "Value": "test-value"}],
)
response = client.list_tags_for_resource(
ResourceId="test", ResourceType="Parameter"
)

View File

@ -10,6 +10,9 @@ import yaml
import hashlib
import copy
import pkgutil
import pytest
from botocore.exceptions import ClientError
from moto.core import ACCOUNT_ID
@ -767,3 +770,32 @@ def test_list_documents():
Filters=[{"Key": "TargetType", "Values": ["/AWS::EC2::Instance"]}]
)
len(response["DocumentIdentifiers"]).should.equal(1)
@mock_ssm
def test_tags_in_list_tags_from_resource_document():
template_file = _get_yaml_template()
json_doc = yaml.safe_load(template_file)
client = boto3.client("ssm", region_name="us-east-1")
client.create_document(
Content=json.dumps(json_doc),
Name="TestDocument",
DocumentType="Command",
DocumentFormat="JSON",
Tags=[{"Key": "spam", "Value": "ham"}],
)
tags = client.list_tags_for_resource(
ResourceId="TestDocument", ResourceType="Document"
)
assert tags.get("TagList") == [{"Key": "spam", "Value": "ham"}]
client.delete_document(Name="TestDocument")
with pytest.raises(ClientError) as ex:
client.list_tags_for_resource(
ResourceType="Document", ResourceId="TestDocument"
)
assert ex.value.response["Error"]["Code"] == "InvalidResourceId"