Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Techdebt: MyPy SSM (#6250)

Browse Source
This commit is contained in:
Bert Blommers 2023-04-24 10:26:25 +00:00 committed by GitHub
parent 2d7c38f64f
commit a32b721c79
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 509 additions and 391 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
moto/ssm/exceptions.py
View File

@ -4,138 +4,138 @@ from moto.core.exceptions import JsonRESTError
class InvalidFilterKey(JsonRESTError):
code = 400
def __init__(self, message):
def __init__(self, message: str):
super().__init__("InvalidFilterKey", message)
class InvalidFilterOption(JsonRESTError):
code = 400
def __init__(self, message):
def __init__(self, message: str):
super().__init__("InvalidFilterOption", message)
class InvalidFilterValue(JsonRESTError):
code = 400
def __init__(self, message):
def __init__(self, message: str):
super().__init__("InvalidFilterValue", message)
class InvalidResourceId(JsonRESTError):
code = 400
def __init__(self):
def __init__(self) -> None:
super().__init__("InvalidResourceId", "Invalid Resource Id")
class InvalidResourceType(JsonRESTError):
code = 400
def __init__(self):
def __init__(self) -> None:
super().__init__("InvalidResourceType", "Invalid Resource Type")
class ParameterNotFound(JsonRESTError):
code = 400
def __init__(self, message):
def __init__(self, message: str):
super().__init__("ParameterNotFound", message)
class ParameterVersionNotFound(JsonRESTError):
code = 400
def __init__(self, message):
def __init__(self, message: str):
super().__init__("ParameterVersionNotFound", message)
class ParameterVersionLabelLimitExceeded(JsonRESTError):
code = 400
def __init__(self, message):
def __init__(self, message: str):
super().__init__("ParameterVersionLabelLimitExceeded", message)
class ValidationException(JsonRESTError):
code = 400
def __init__(self, message):
def __init__(self, message: str):
super().__init__("ValidationException", message)
class DocumentAlreadyExists(JsonRESTError):
code = 400
def __init__(self, message):
def __init__(self, message: str):
super().__init__("DocumentAlreadyExists", message)
class DocumentPermissionLimit(JsonRESTError):
code = 400
def __init__(self, message):
def __init__(self, message: str):
super().__init__("DocumentPermissionLimit", message)
class InvalidPermissionType(JsonRESTError):
code = 400
def __init__(self, message):
def __init__(self, message: str):
super().__init__("InvalidPermissionType", message)
class InvalidDocument(JsonRESTError):
code = 400
def __init__(self, message):
def __init__(self, message: str):
super().__init__("InvalidDocument", message)
class InvalidDocumentOperation(JsonRESTError):
code = 400
def __init__(self, message):
def __init__(self, message: str):
super().__init__("InvalidDocumentOperation", message)
class AccessDeniedException(JsonRESTError):
code = 400
def __init__(self, message):
def __init__(self, message: str):
super().__init__("AccessDeniedException", message)
class InvalidDocumentContent(JsonRESTError):
code = 400
def __init__(self, message):
def __init__(self, message: str):
super().__init__("InvalidDocumentContent", message)
class InvalidDocumentVersion(JsonRESTError):
code = 400
def __init__(self, message):
def __init__(self, message: str):
super().__init__("InvalidDocumentVersion", message)
class DuplicateDocumentVersionName(JsonRESTError):
code = 400
def __init__(self, message):
def __init__(self, message: str):
super().__init__("DuplicateDocumentVersionName", message)
class DuplicateDocumentContent(JsonRESTError):
code = 400
def __init__(self, message):
def __init__(self, message: str):
super().__init__("DuplicateDocumentContent", message)
class ParameterMaxVersionLimitExceeded(JsonRESTError):
code = 400
def __init__(self, message):
def __init__(self, message: str):
super().__init__("ParameterMaxVersionLimitExceeded", message)

686
moto/ssm/models.py
View File

File diff suppressed because it is too large Load Diff

142
moto/ssm/responses.py
View File

@ -1,4 +1,5 @@
import json
from typing import Any, Dict, Tuple, Union
from moto.core.responses import BaseResponse
from .exceptions import ValidationException
@ -13,14 +14,7 @@ class SimpleSystemManagerResponse(BaseResponse):
def ssm_backend(self) -> SimpleSystemManagerBackend:
return ssm_backends[self.current_account][self.region]
@property
def request_params(self):
try:
return json.loads(self.body)
except ValueError:
return {}
def create_document(self):
def create_document(self) -> str:
content = self._get_param("Content")
requires = self._get_param("Requires")
attachments = self._get_param("Attachments")
@ -45,7 +39,7 @@ class SimpleSystemManagerResponse(BaseResponse):
return json.dumps({"DocumentDescription": result})
def delete_document(self):
def delete_document(self) -> str:
name = self._get_param("Name")
document_version = self._get_param("DocumentVersion")
version_name = self._get_param("VersionName")
@ -59,7 +53,7 @@ class SimpleSystemManagerResponse(BaseResponse):
return json.dumps({})
def get_document(self):
def get_document(self) -> str:
name = self._get_param("Name")
version_name = self._get_param("VersionName")
document_version = self._get_param("DocumentVersion")
@ -74,7 +68,7 @@ class SimpleSystemManagerResponse(BaseResponse):
return json.dumps(document)
def describe_document(self):
def describe_document(self) -> str:
name = self._get_param("Name")
document_version = self._get_param("DocumentVersion")
version_name = self._get_param("VersionName")
@ -85,7 +79,7 @@ class SimpleSystemManagerResponse(BaseResponse):
return json.dumps({"Document": result})
def update_document(self):
def update_document(self) -> str:
content = self._get_param("Content")
attachments = self._get_param("Attachments")
name = self._get_param("Name")
@ -106,7 +100,7 @@ class SimpleSystemManagerResponse(BaseResponse):
return json.dumps({"DocumentDescription": result})
def update_document_default_version(self):
def update_document_default_version(self) -> str:
name = self._get_param("Name")
document_version = self._get_param("DocumentVersion")
@ -115,7 +109,7 @@ class SimpleSystemManagerResponse(BaseResponse):
)
return json.dumps({"Description": result})
def list_documents(self):
def list_documents(self) -> str:
document_filter_list = self._get_param("DocumentFilterList")
filters = self._get_param("Filters")
max_results = self._get_param("MaxResults", 10)
@ -125,18 +119,18 @@ class SimpleSystemManagerResponse(BaseResponse):
document_filter_list=document_filter_list,
filters=filters,
max_results=max_results,
next_token=next_token,
token=next_token,
)
return json.dumps({"DocumentIdentifiers": documents, "NextToken": token})
def describe_document_permission(self):
def describe_document_permission(self) -> str:
name = self._get_param("Name")
result = self.ssm_backend.describe_document_permission(name=name)
return json.dumps(result)
def modify_document_permission(self):
def modify_document_permission(self) -> str:
account_ids_to_add = self._get_param("AccountIdsToAdd")
account_ids_to_remove = self._get_param("AccountIdsToRemove")
name = self._get_param("Name")
@ -150,11 +144,9 @@ class SimpleSystemManagerResponse(BaseResponse):
shared_document_version=shared_document_version,
permission_type=permission_type,
)
return "{}"
def _get_param(self, param_name, if_none=None):
return self.request_params.get(param_name, if_none)
def delete_parameter(self):
def delete_parameter(self) -> Union[str, Tuple[str, Dict[str, int]]]:
name = self._get_param("Name")
result = self.ssm_backend.delete_parameter(name)
if result is None:
@ -165,11 +157,11 @@ class SimpleSystemManagerResponse(BaseResponse):
return json.dumps(error), dict(status=400)
return json.dumps({})
def delete_parameters(self):
def delete_parameters(self) -> str:
names = self._get_param("Names")
result = self.ssm_backend.delete_parameters(names)
response = {"DeletedParameters": [], "InvalidParameters": []}
response: Dict[str, Any] = {"DeletedParameters": [], "InvalidParameters": []}
for name in names:
if name in result:
@ -178,7 +170,7 @@ class SimpleSystemManagerResponse(BaseResponse):
response["InvalidParameters"].append(name)
return json.dumps(response)
def get_parameter(self):
def get_parameter(self) -> Union[str, Tuple[str, Dict[str, int]]]:
name = self._get_param("Name")
with_decryption = self._get_param("WithDecryption")
@ -202,13 +194,13 @@ class SimpleSystemManagerResponse(BaseResponse):
response = {"Parameter": result.response_object(with_decryption, self.region)}
return json.dumps(response)
def get_parameters(self):
def get_parameters(self) -> str:
names = self._get_param("Names")
with_decryption = self._get_param("WithDecryption")
result = self.ssm_backend.get_parameters(names)
response = {"Parameters": [], "InvalidParameters": []}
response: Dict[str, Any] = {"Parameters": [], "InvalidParameters": []}
for name, parameter in result.items():
param_data = parameter.response_object(with_decryption, self.region)
@ -220,7 +212,7 @@ class SimpleSystemManagerResponse(BaseResponse):
response["InvalidParameters"].append(name)
return json.dumps(response)
def get_parameters_by_path(self):
def get_parameters_by_path(self) -> str:
path = self._get_param("Path")
with_decryption = self._get_param("WithDecryption")
recursive = self._get_param("Recursive", False)
@ -236,7 +228,7 @@ class SimpleSystemManagerResponse(BaseResponse):
max_results=max_results,
)
response = {"Parameters": [], "NextToken": next_token}
response: Dict[str, Any] = {"Parameters": [], "NextToken": next_token}
for parameter in result:
param_data = parameter.response_object(with_decryption, self.region)
@ -244,7 +236,7 @@ class SimpleSystemManagerResponse(BaseResponse):
return json.dumps(response)
def describe_parameters(self):
def describe_parameters(self) -> str:
page_size = 10
filters = self._get_param("Filters")
parameter_filters = self._get_param("ParameterFilters")
@ -257,7 +249,7 @@ class SimpleSystemManagerResponse(BaseResponse):
result = self.ssm_backend.describe_parameters(filters, parameter_filters)
response = {"Parameters": []}
response: Dict[str, Any] = {"Parameters": []}
end = token + page_size
for parameter in result[token:]:
@ -270,7 +262,7 @@ class SimpleSystemManagerResponse(BaseResponse):
return json.dumps(response)
def put_parameter(self):
def put_parameter(self) -> Union[str, Tuple[str, Dict[str, int]]]:
name = self._get_param("Name")
description = self._get_param("Description")
value = self._get_param("Value")
@ -303,7 +295,7 @@ class SimpleSystemManagerResponse(BaseResponse):
response = {"Version": result}
return json.dumps(response)
def get_parameter_history(self):
def get_parameter_history(self) -> Union[str, Tuple[str, Dict[str, int]]]:
name = self._get_param("Name")
with_decryption = self._get_param("WithDecryption")
next_token = self._get_param("NextToken")
@ -320,19 +312,19 @@ class SimpleSystemManagerResponse(BaseResponse):
}
return json.dumps(error), dict(status=400)
response = {"Parameters": []}
for parameter_version in result:
param_data = parameter_version.describe_response_object(
response = {
"Parameters": [
p_v.describe_response_object(
decrypt=with_decryption, include_labels=True
)
response["Parameters"].append(param_data)
if new_next_token is not None:
response["NextToken"] = new_next_token
for p_v in result
],
"NextToken": new_next_token,
}
return json.dumps(response)
def label_parameter_version(self):
def label_parameter_version(self) -> str:
name = self._get_param("Name")
version = self._get_param("ParameterVersion")
labels = self._get_param("Labels")
@ -344,7 +336,7 @@ class SimpleSystemManagerResponse(BaseResponse):
response = {"InvalidLabels": invalid_labels, "ParameterVersion": version}
return json.dumps(response)
def add_tags_to_resource(self):
def add_tags_to_resource(self) -> str:
resource_id = self._get_param("ResourceId")
resource_type = self._get_param("ResourceType")
tags = {t["Key"]: t["Value"] for t in self._get_param("Tags")}
@ -353,7 +345,7 @@ class SimpleSystemManagerResponse(BaseResponse):
)
return json.dumps({})
def remove_tags_from_resource(self):
def remove_tags_from_resource(self) -> str:
resource_id = self._get_param("ResourceId")
resource_type = self._get_param("ResourceType")
keys = self._get_param("TagKeys")
@ -362,7 +354,7 @@ class SimpleSystemManagerResponse(BaseResponse):
)
return json.dumps({})
def list_tags_for_resource(self):
def list_tags_for_resource(self) -> str:
resource_id = self._get_param("ResourceId")
resource_type = self._get_param("ResourceType")
tags = self.ssm_backend.list_tags_for_resource(
@ -372,21 +364,56 @@ class SimpleSystemManagerResponse(BaseResponse):
response = {"TagList": tag_list}
return json.dumps(response)
def send_command(self):
return json.dumps(self.ssm_backend.send_command(**self.request_params))
def list_commands(self):
return json.dumps(self.ssm_backend.list_commands(**self.request_params))
def get_command_invocation(self):
return json.dumps(
self.ssm_backend.get_command_invocation(**self.request_params)
def send_command(self) -> str:
comment = self._get_param("Comment", "")
document_name = self._get_param("DocumentName")
timeout_seconds = self._get_int_param("TimeoutSeconds")
instance_ids = self._get_param("InstanceIds", [])
max_concurrency = self._get_param("MaxConcurrency", "50")
max_errors = self._get_param("MaxErrors", "0")
notification_config = self._get_param("NotificationConfig")
output_s3_bucket_name = self._get_param("OutputS3BucketName", "")
output_s3_key_prefix = self._get_param("OutputS3KeyPrefix", "")
output_s3_region = self._get_param("OutputS3Region", "")
parameters = self._get_param("Parameters", {})
service_role_arn = self._get_param("ServiceRoleArn", "")
targets = self._get_param("Targets", [])
command = self.ssm_backend.send_command(
comment=comment,
document_name=document_name,
timeout_seconds=timeout_seconds,
instance_ids=instance_ids,
max_concurrency=max_concurrency,
max_errors=max_errors,
notification_config=notification_config,
output_s3_bucket_name=output_s3_bucket_name,
output_s3_key_prefix=output_s3_key_prefix,
output_s3_region=output_s3_region,
parameters=parameters,
service_role_arn=service_role_arn,
targets=targets,
)
return json.dumps({"Command": command.response_object()})
def create_maintenance_window(self):
def list_commands(self) -> str:
command_id = self._get_param("CommandId")
instance_id = self._get_param("InstanceId")
commands = self.ssm_backend.list_commands(command_id, instance_id)
response = {"Commands": [command.response_object() for command in commands]}
return json.dumps(response)
def get_command_invocation(self) -> str:
command_id = self._get_param("CommandId")
instance_id = self._get_param("InstanceId")
plugin_name = self._get_param("PluginName")
response = self.ssm_backend.get_command_invocation(
command_id, instance_id, plugin_name
)
return json.dumps(response)
def create_maintenance_window(self) -> str:
name = self._get_param("Name")
desc = self._get_param("Description", None)
enabled = self._get_bool_param("Enabled", True)
duration = self._get_int_param("Duration")
cutoff = self._get_int_param("Cutoff")
schedule = self._get_param("Schedule")
@ -397,7 +424,6 @@ class SimpleSystemManagerResponse(BaseResponse):
window_id = self.ssm_backend.create_maintenance_window(
name=name,
description=desc,
enabled=enabled,
duration=duration,
cutoff=cutoff,
schedule=schedule,
@ -408,12 +434,12 @@ class SimpleSystemManagerResponse(BaseResponse):
)
return json.dumps({"WindowId": window_id})
def get_maintenance_window(self):
def get_maintenance_window(self) -> str:
window_id = self._get_param("WindowId")
window = self.ssm_backend.get_maintenance_window(window_id)
return json.dumps(window.to_json())
def describe_maintenance_windows(self):
def describe_maintenance_windows(self) -> str:
filters = self._get_param("Filters", None)
windows = [
window.to_json()
@ -421,7 +447,7 @@ class SimpleSystemManagerResponse(BaseResponse):
]
return json.dumps({"WindowIdentities": windows})
def delete_maintenance_window(self):
def delete_maintenance_window(self) -> str:
window_id = self._get_param("WindowId")
self.ssm_backend.delete_maintenance_window(window_id)
return "{}"

17
moto/ssm/utils.py
View File

@ -1,16 +1,19 @@
def parameter_arn(account_id, region, parameter_name):
from typing import Any, Dict, List
def parameter_arn(account_id: str, region: str, parameter_name: str) -> str:
if parameter_name[0] == "/":
parameter_name = parameter_name[1:]
return f"arn:aws:ssm:{region}:{account_id}:parameter/{parameter_name}"
def convert_to_tree(parameters):
def convert_to_tree(parameters: List[Dict[str, Any]]) -> Dict[str, Any]:
"""
Convert input into a smaller, less redundant data set in tree form
Input: [{"Name": "/a/b/c", "Value": "af-south-1", ...}, ..]
Output: {"a": {"b": {"c": {"Value": af-south-1}, ..}, ..}, ..}
"""
tree_dict = {}
tree_dict: Dict[str, Any] = {}
for p in parameters:
current_level = tree_dict
for path in p["Name"].split("/"):
@ -23,18 +26,20 @@ def convert_to_tree(parameters):
return tree_dict
def convert_to_params(tree):
def convert_to_params(tree: Dict[str, Any]) -> List[Dict[str, Any]]:
"""
Inverse of 'convert_to_tree'
"""
def m(tree, params, current_path=""):
def m(
tree: Dict[str, Any], params: List[Dict[str, Any]], current_path: str = ""
) -> None:
for key, value in tree.items():
if key == "Value":
params.append({"Name": current_path, "Value": value})
else:
m(value, params, current_path + "/" + key)
params = []
params: List[Dict[str, Any]] = []
m(tree, params)
return params

2
setup.cfg
View File

@ -239,7 +239,7 @@ disable = W,C,R,E
enable = anomalous-backslash-in-string, arguments-renamed, dangerous-default-value, deprecated-module, function-redefined, import-self, redefined-builtin, redefined-outer-name, reimported, pointless-statement, super-with-arguments, unused-argument, unused-import, unused-variable, useless-else-on-loop, wildcard-import
[mypy]
files= moto/a*,moto/b*,moto/c*,moto/d*,moto/e*,moto/f*,moto/g*,moto/i*,moto/k*,moto/l*,moto/m*,moto/n*,moto/o*,moto/p*,moto/q*,moto/r*,moto/s3*,moto/sagemaker,moto/secretsmanager,moto/scheduler
files= moto/a*,moto/b*,moto/c*,moto/d*,moto/e*,moto/f*,moto/g*,moto/i*,moto/k*,moto/l*,moto/m*,moto/n*,moto/o*,moto/p*,moto/q*,moto/r*,moto/s3*,moto/sagemaker,moto/secretsmanager,moto/ssm,moto/scheduler
show_column_numbers=True
show_error_codes = True
disable_error_code=abstract

9
tests/test_ssm/test_ssm_boto3.py
View File

@ -1739,8 +1739,12 @@ def test_send_command():
before = datetime.datetime.now()
response = client.send_command(
Comment="some comment",
InstanceIds=["i-123456"],
DocumentName=ssm_document,
TimeoutSeconds=42,
MaxConcurrency="360",
MaxErrors="2",
Parameters=params,
OutputS3Region="us-east-2",
OutputS3BucketName="the-bucket",
@ -1749,6 +1753,7 @@ def test_send_command():
cmd = response["Command"]
cmd["CommandId"].should_not.be(None)
assert cmd["Comment"] == "some comment"
cmd["DocumentName"].should.equal(ssm_document)
cmd["Parameters"].should.equal(params)
@ -1759,6 +1764,10 @@ def test_send_command():
cmd["ExpiresAfter"].should.be.greater_than(before)
cmd["DeliveryTimedOutCount"].should.equal(0)
assert cmd["TimeoutSeconds"] == 42
assert cmd["MaxConcurrency"] == "360"
assert cmd["MaxErrors"] == "2"
# test sending a command without any optional parameters
response = client.send_command(DocumentName=ssm_document)