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

black linting

Browse Source
This commit is contained in:
Alex Bainbridge 2020-06-30 12:43:42 -04:00
parent 82825787db
commit c9b38e25b8
4 changed files with 576 additions and 296 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
moto/ssm/exceptions.py
View File

@ -73,7 +73,9 @@ class InvalidDocumentOperation(JsonRESTError):
code = 400 code = 400
def __init__(self, message): def __init__(self, message):
super(InvalidDocumentOperation, self).__init__("InvalidDocumentOperation", message) super(InvalidDocumentOperation, self).__init__(
"InvalidDocumentOperation", message
)
class InvalidDocumentContent(JsonRESTError): class InvalidDocumentContent(JsonRESTError):
@ -94,12 +96,15 @@ class DuplicateDocumentVersionName(JsonRESTError):
code = 400 code = 400
def __init__(self, message): def __init__(self, message):
super(DuplicateDocumentVersionName, self).__init__("DuplicateDocumentVersionName", message) super(DuplicateDocumentVersionName, self).__init__(
"DuplicateDocumentVersionName", message
)
class DuplicateDocumentContent(JsonRESTError): class DuplicateDocumentContent(JsonRESTError):
code = 400 code = 400
def __init__(self, message): def __init__(self, message):
super(DuplicateDocumentContent, self).__init__("DuplicateDocumentContent", message) super(DuplicateDocumentContent, self).__init__(
"DuplicateDocumentContent", message
)

261
moto/ssm/models.py
View File

@ -31,7 +31,7 @@ from .exceptions import (
InvalidDocumentContent, InvalidDocumentContent,
InvalidDocumentVersion, InvalidDocumentVersion,
DuplicateDocumentVersionName, DuplicateDocumentVersionName,
DuplicateDocumentContent DuplicateDocumentContent,
) )
@ -123,7 +123,11 @@ def generate_ssm_doc_param_list(parameters):
final_dict["Type"] = param_info["type"] final_dict["Type"] = param_info["type"]
final_dict["Description"] = param_info["description"] final_dict["Description"] = param_info["description"]
if param_info["type"] == "StringList" or param_info["type"] == "StringMap" or param_info["type"] == "MapList": if (
param_info["type"] == "StringList"
or param_info["type"] == "StringMap"
or param_info["type"] == "MapList"
):
final_dict["DefaultValue"] = json.dumps(param_info["default"]) final_dict["DefaultValue"] = json.dumps(param_info["default"])
else: else:
final_dict["DefaultValue"] = str(param_info["default"]) final_dict["DefaultValue"] = str(param_info["default"])
@ -134,8 +138,19 @@ def generate_ssm_doc_param_list(parameters):
class Document(BaseModel): class Document(BaseModel):
def __init__(self, name, version_name, content, document_type, document_format, requires, attachments, def __init__(
target_type, tags, document_version="1"): self,
name,
version_name,
content,
document_type,
document_format,
requires,
attachments,
target_type,
tags,
document_version="1",
):
self.name = name self.name = name
self.version_name = version_name self.version_name = version_name
self.content = content self.content = content
@ -155,14 +170,18 @@ class Document(BaseModel):
try: try:
content_json = json.loads(content) content_json = json.loads(content)
except json.decoder.JSONDecodeError: except json.decoder.JSONDecodeError:
raise InvalidDocumentContent("The content for the document is not valid.") raise InvalidDocumentContent(
"The content for the document is not valid."
)
elif document_format == "YAML": elif document_format == "YAML":
try: try:
content_json = yaml.safe_load(content) content_json = yaml.safe_load(content)
except yaml.YAMLError: except yaml.YAMLError:
raise InvalidDocumentContent("The content for the document is not valid.") raise InvalidDocumentContent(
"The content for the document is not valid."
)
else: else:
raise ValidationException(f'Invalid document format {document_format}') raise ValidationException(f"Invalid document format {document_format}")
self.content_json = content_json self.content_json = content_json
@ -171,11 +190,17 @@ class Document(BaseModel):
self.description = content_json.get("description") self.description = content_json.get("description")
self.outputs = content_json.get("outputs") self.outputs = content_json.get("outputs")
self.files = content_json.get("files") self.files = content_json.get("files")
# TODO add platformType # TODO add platformType (requires mapping the ssm actions to OS's this isn't well documented)
self.platform_types = ["Not Implemented (moto)"] self.platform_types = ["Not Implemented (moto)"]
self.parameter_list = generate_ssm_doc_param_list(content_json.get("parameters")) self.parameter_list = generate_ssm_doc_param_list(
content_json.get("parameters")
)
if self.schema_version == "0.3" or self.schema_version == "2.0" or self.schema_version == "2.2": if (
self.schema_version == "0.3"
or self.schema_version == "2.0"
or self.schema_version == "2.2"
):
self.mainSteps = content_json["mainSteps"] self.mainSteps = content_json["mainSteps"]
elif self.schema_version == "1.2": elif self.schema_version == "1.2":
self.runtimeConfig = content_json.get("runtimeConfig") self.runtimeConfig = content_json.get("runtimeConfig")
@ -184,8 +209,6 @@ class Document(BaseModel):
raise InvalidDocumentContent("The content for the document is not valid.") raise InvalidDocumentContent("The content for the document is not valid.")
class Command(BaseModel): class Command(BaseModel):
def __init__( def __init__(
self, self,
@ -356,14 +379,23 @@ class Command(BaseModel):
def _validate_document_format(document_format): def _validate_document_format(document_format):
aws_doc_formats = ["JSON", "YAML"] aws_doc_formats = ["JSON", "YAML"]
if document_format not in aws_doc_formats: if document_format not in aws_doc_formats:
raise ValidationException(f'Invalid document format {document_format}') raise ValidationException(f"Invalid document format {document_format}")
def _validate_document_info(content, name, document_type, document_format, strict=True): def _validate_document_info(content, name, document_type, document_format, strict=True):
aws_ssm_name_regex = r'^[a-zA-Z0-9_\-.]{3,128}$' aws_ssm_name_regex = r"^[a-zA-Z0-9_\-.]{3,128}$"
aws_name_reject_list = ["aws-", "amazon", "amzn"] aws_name_reject_list = ["aws-", "amazon", "amzn"]
aws_doc_types = ["Command", "Policy", "Automation", "Session", "Package", "ApplicationConfiguration", aws_doc_types = [
"ApplicationConfigurationSchema", "DeploymentStrategy", "ChangeCalendar"] "Command",
"Policy",
"Automation",
"Session",
"Package",
"ApplicationConfiguration",
"ApplicationConfigurationSchema",
"DeploymentStrategy",
"ChangeCalendar",
]
_validate_document_format(document_format) _validate_document_format(document_format)
@ -371,14 +403,14 @@ def _validate_document_info(content, name, document_type, document_format, stric
raise ValidationException("Content is required") raise ValidationException("Content is required")
if list(filter(name.startswith, aws_name_reject_list)): if list(filter(name.startswith, aws_name_reject_list)):
raise ValidationException(f'Invalid document name {name}') raise ValidationException(f"Invalid document name {name}")
ssm_name_pattern = re.compile(aws_ssm_name_regex) ssm_name_pattern = re.compile(aws_ssm_name_regex)
if not ssm_name_pattern.match(name): if not ssm_name_pattern.match(name):
raise ValidationException(f'Invalid document name {name}') raise ValidationException(f"Invalid document name {name}")
if strict and document_type not in aws_doc_types: if strict and document_type not in aws_doc_types:
# Update document doesn't use document type # Update document doesn't use document type
raise ValidationException(f'Invalid document type {document_type}') raise ValidationException(f"Invalid document type {document_type}")
def _document_filter_equal_comparator(keyed_value, filter): def _document_filter_equal_comparator(keyed_value, filter):
@ -397,7 +429,9 @@ def _document_filter_list_includes_comparator(keyed_value_list, filter):
def _document_filter_match(filters, ssm_doc): def _document_filter_match(filters, ssm_doc):
for filter in filters: for filter in filters:
if filter["Key"] == "Name" and not _document_filter_equal_comparator(ssm_doc.name, filter): if filter["Key"] == "Name" and not _document_filter_equal_comparator(
ssm_doc.name, filter
):
return False return False
elif filter["Key"] == "Owner": elif filter["Key"] == "Owner":
@ -409,14 +443,21 @@ def _document_filter_match(filters, ssm_doc):
if not _document_filter_equal_comparator(ssm_doc.owner, filter): if not _document_filter_equal_comparator(ssm_doc.owner, filter):
return False return False
elif filter["Key"] == "PlatformTypes" and not \ elif filter[
_document_filter_list_includes_comparator(ssm_doc.platform_types, filter): "Key"
] == "PlatformTypes" and not _document_filter_list_includes_comparator(
ssm_doc.platform_types, filter
):
return False return False
elif filter["Key"] == "DocumentType" and not _document_filter_equal_comparator(ssm_doc.document_type, filter): elif filter["Key"] == "DocumentType" and not _document_filter_equal_comparator(
ssm_doc.document_type, filter
):
return False return False
elif filter["Key"] == "TargetType" and not _document_filter_equal_comparator(ssm_doc.target_type, filter): elif filter["Key"] == "TargetType" and not _document_filter_equal_comparator(
ssm_doc.target_type, filter
):
return False return False
return True return True
@ -440,10 +481,10 @@ class SimpleSystemManagerBackend(BaseBackend):
def _generate_document_description(self, document): def _generate_document_description(self, document):
latest = self._documents[document.name]['latest_version'] latest = self._documents[document.name]["latest_version"]
default_version = self._documents[document.name]["default_version"] default_version = self._documents[document.name]["default_version"]
base = { base = {
"Hash": hashlib.sha256(document.content.encode('utf-8')).hexdigest(), "Hash": hashlib.sha256(document.content.encode("utf-8")).hexdigest(),
"HashType": "Sha256", "HashType": "Sha256",
"Name": document.name, "Name": document.name,
"Owner": document.owner, "Owner": document.owner,
@ -457,7 +498,7 @@ class SimpleSystemManagerBackend(BaseBackend):
"SchemaVersion": document.schema_version, "SchemaVersion": document.schema_version,
"LatestVersion": latest, "LatestVersion": latest,
"DefaultVersion": default_version, "DefaultVersion": default_version,
"DocumentFormat": document.document_format "DocumentFormat": document.document_format,
} }
if document.version_name: if document.version_name:
base["VersionName"] = document.version_name base["VersionName"] = document.version_name
@ -475,7 +516,7 @@ class SimpleSystemManagerBackend(BaseBackend):
"Status": ssm_document.status, "Status": ssm_document.status,
"Content": ssm_document.content, "Content": ssm_document.content,
"DocumentType": ssm_document.document_type, "DocumentType": ssm_document.document_type,
"DocumentFormat": document_format "DocumentFormat": document_format,
} }
if document_format == "JSON": if document_format == "JSON":
@ -483,7 +524,7 @@ class SimpleSystemManagerBackend(BaseBackend):
elif document_format == "YAML": elif document_format == "YAML":
base["Content"] = yaml.dump(ssm_document.content_json) base["Content"] = yaml.dump(ssm_document.content_json)
else: else:
raise ValidationException(f'Invalid document format {document_format}') raise ValidationException(f"Invalid document format {document_format}")
if ssm_document.version_name: if ssm_document.version_name:
base["VersionName"] = ssm_document.version_name base["VersionName"] = ssm_document.version_name
@ -501,7 +542,7 @@ class SimpleSystemManagerBackend(BaseBackend):
"DocumentVersion": ssm_document.document_version, "DocumentVersion": ssm_document.document_version,
"DocumentType": ssm_document.document_type, "DocumentType": ssm_document.document_type,
"SchemaVersion": ssm_document.schema_version, "SchemaVersion": ssm_document.schema_version,
"DocumentFormat": ssm_document.document_format "DocumentFormat": ssm_document.document_format,
} }
if ssm_document.version_name: if ssm_document.version_name:
base["VersionName"] = ssm_document.version_name base["VersionName"] = ssm_document.version_name
@ -516,24 +557,44 @@ class SimpleSystemManagerBackend(BaseBackend):
return base return base
def create_document(self, content, requires, attachments, name, version_name, document_type, document_format, def create_document(
target_type, tags): self,
ssm_document = Document(name=name, version_name=version_name, content=content, document_type=document_type, content,
document_format=document_format, requires=requires, attachments=attachments, requires,
target_type=target_type, tags=tags) attachments,
name,
version_name,
document_type,
document_format,
target_type,
tags,
):
ssm_document = Document(
name=name,
version_name=version_name,
content=content,
document_type=document_type,
document_format=document_format,
requires=requires,
attachments=attachments,
target_type=target_type,
tags=tags,
)
_validate_document_info(content=content, name=name, document_type=document_type, _validate_document_info(
document_format=document_format) content=content,
name=name,
document_type=document_type,
document_format=document_format,
)
if self._documents.get(ssm_document.name): if self._documents.get(ssm_document.name):
raise DocumentAlreadyExists(f"The specified document already exists.") raise DocumentAlreadyExists(f"The specified document already exists.")
self._documents[ssm_document.name] = { self._documents[ssm_document.name] = {
"documents": { "documents": {ssm_document.document_version: ssm_document},
ssm_document.document_version: ssm_document
},
"default_version": ssm_document.document_version, "default_version": ssm_document.document_version,
"latest_version": ssm_document.document_version "latest_version": ssm_document.document_version,
} }
return self._generate_document_description(ssm_document) return self._generate_document_description(ssm_document)
@ -545,20 +606,34 @@ class SimpleSystemManagerBackend(BaseBackend):
if documents: if documents:
default_version = self._documents[name]["default_version"] default_version = self._documents[name]["default_version"]
if documents[default_version].document_type == "ApplicationConfigurationSchema" and not force: if (
raise InvalidDocumentOperation("You attempted to delete a document while it is still shared. " documents[default_version].document_type
"You must stop sharing the document before you can delete it.") == "ApplicationConfigurationSchema"
and not force
):
raise InvalidDocumentOperation(
"You attempted to delete a document while it is still shared. "
"You must stop sharing the document before you can delete it."
)
if document_version and document_version == default_version: if document_version and document_version == default_version:
raise InvalidDocumentOperation("Default version of the document can't be deleted.") raise InvalidDocumentOperation(
"Default version of the document can't be deleted."
)
if document_version or version_name: if document_version or version_name:
# We delete only a specific version # We delete only a specific version
delete_doc = self._find_document(name, document_version, version_name) delete_doc = self._find_document(name, document_version, version_name)
# we can't delete only the default version # we can't delete only the default version
if delete_doc and delete_doc.document_version == default_version and len(documents) != 1: if (
raise InvalidDocumentOperation("Default version of the document can't be deleted.") delete_doc
and delete_doc.document_version == default_version
and len(documents) != 1
):
raise InvalidDocumentOperation(
"Default version of the document can't be deleted."
)
if delete_doc: if delete_doc:
keys_to_delete.add(delete_doc.document_version) keys_to_delete.add(delete_doc.document_version)
@ -571,8 +646,6 @@ class SimpleSystemManagerBackend(BaseBackend):
for key in keys_to_delete: for key in keys_to_delete:
del self._documents[name]["documents"][key] del self._documents[name]["documents"][key]
keys = self._documents[name]["documents"].keys()
if len(self._documents[name]["documents"].keys()) == 0: if len(self._documents[name]["documents"].keys()) == 0:
del self._documents[name] del self._documents[name]
else: else:
@ -586,7 +659,9 @@ class SimpleSystemManagerBackend(BaseBackend):
else: else:
raise InvalidDocument("The specified document does not exist.") raise InvalidDocument("The specified document does not exist.")
def _find_document(self, name, document_version=None, version_name=None, strict=True): def _find_document(
self, name, document_version=None, version_name=None, strict=True
):
if not self._documents.get(name): if not self._documents.get(name):
raise InvalidDocument(f"The specified document does not exist.") raise InvalidDocument(f"The specified document does not exist.")
@ -595,12 +670,15 @@ class SimpleSystemManagerBackend(BaseBackend):
if not version_name and not document_version: if not version_name and not document_version:
# Retrieve default version # Retrieve default version
default_version = self._documents[name]['default_version'] default_version = self._documents[name]["default_version"]
ssm_document = documents.get(default_version) ssm_document = documents.get(default_version)
elif version_name and document_version: elif version_name and document_version:
for doc_version, document in documents.items(): for doc_version, document in documents.items():
if doc_version == document_version and document.version_name == version_name: if (
doc_version == document_version
and document.version_name == version_name
):
ssm_document = document ssm_document = document
break break
@ -642,32 +720,68 @@ class SimpleSystemManagerBackend(BaseBackend):
return base return base
def update_document(self, content, attachments, name, version_name, document_version, document_format, target_type): def update_document(
_validate_document_info(content=content, name=name, document_type=None, document_format=document_format, self,
strict=False) content,
attachments,
name,
version_name,
document_version,
document_format,
target_type,
):
_validate_document_info(
content=content,
name=name,
document_type=None,
document_format=document_format,
strict=False,
)
if not self._documents.get(name): if not self._documents.get(name):
raise InvalidDocument("The specified document does not exist.") raise InvalidDocument("The specified document does not exist.")
if self._documents[name]['latest_version'] != document_version and document_version != "$LATEST": if (
raise InvalidDocumentVersion("The document version is not valid or does not exist.") self._documents[name]["latest_version"] != document_version
if version_name and self._find_document(name, version_name=version_name, strict=False): and document_version != "$LATEST"
raise DuplicateDocumentVersionName(f"The specified version name is a duplicate.") ):
raise InvalidDocumentVersion(
"The document version is not valid or does not exist."
)
if version_name and self._find_document(
name, version_name=version_name, strict=False
):
raise DuplicateDocumentVersionName(
f"The specified version name is a duplicate."
)
old_ssm_document = self._find_document(name) old_ssm_document = self._find_document(name)
new_ssm_document = Document(name=name, version_name=version_name, content=content, new_ssm_document = Document(
document_type=old_ssm_document.document_type, document_format=document_format, name=name,
requires=old_ssm_document.requires, attachments=attachments, version_name=version_name,
target_type=target_type, tags=old_ssm_document.tags, content=content,
document_version=str(int(self._documents[name]['latest_version']) + 1)) document_type=old_ssm_document.document_type,
document_format=document_format,
requires=old_ssm_document.requires,
attachments=attachments,
target_type=target_type,
tags=old_ssm_document.tags,
document_version=str(int(self._documents[name]["latest_version"]) + 1),
)
for doc_version, document in self._documents[name]['documents'].items(): for doc_version, document in self._documents[name]["documents"].items():
if document.content == new_ssm_document.content: if document.content == new_ssm_document.content:
raise DuplicateDocumentContent("The content of the association document matches another document. " raise DuplicateDocumentContent(
"Change the content of the document and try again.") "The content of the association document matches another document. "
"Change the content of the document and try again."
)
self._documents[name]["latest_version"] = str(int(self._documents[name]["latest_version"]) + 1) self._documents[name]["latest_version"] = str(
self._documents[name]["documents"][new_ssm_document.document_version] = new_ssm_document int(self._documents[name]["latest_version"]) + 1
)
self._documents[name]["documents"][
new_ssm_document.document_version
] = new_ssm_document
return self._generate_document_description(new_ssm_document) return self._generate_document_description(new_ssm_document)
@ -675,7 +789,9 @@ class SimpleSystemManagerBackend(BaseBackend):
ssm_document = self._find_document(name, document_version, version_name) ssm_document = self._find_document(name, document_version, version_name)
return self._generate_document_description(ssm_document) return self._generate_document_description(ssm_document)
def list_documents(self, document_filter_list, filters, max_results=10, next_token="0"): def list_documents(
self, document_filter_list, filters, max_results=10, next_token="0"
):
if document_filter_list: if document_filter_list:
raise ValidationException( raise ValidationException(
"DocumentFilterList is deprecated. Instead use Filters." "DocumentFilterList is deprecated. Instead use Filters."
@ -690,13 +806,12 @@ class SimpleSystemManagerBackend(BaseBackend):
# There's still more to go so we need a next token # There's still more to go so we need a next token
return results, str(next_token + len(results)) return results, str(next_token + len(results))
if dummy_token_tracker < next_token: if dummy_token_tracker < next_token:
dummy_token_tracker = dummy_token_tracker + 1 dummy_token_tracker = dummy_token_tracker + 1
continue continue
default_version = document_bundle['default_version'] default_version = document_bundle["default_version"]
ssm_doc = self._documents[document_name]['documents'][default_version] ssm_doc = self._documents[document_name]["documents"][default_version]
if filters and not _document_filter_match(filters, ssm_doc): if filters and not _document_filter_match(filters, ssm_doc):
# If we have filters enabled, and we don't match them, # If we have filters enabled, and we don't match them,
continue continue

80
moto/ssm/responses.py
View File

@ -28,21 +28,31 @@ class SimpleSystemManagerResponse(BaseResponse):
target_type = self._get_param("TargetType") target_type = self._get_param("TargetType")
tags = self._get_param("Tags") tags = self._get_param("Tags")
result = self.ssm_backend.create_document(content=content, requires=requires, attachments=attachments, result = self.ssm_backend.create_document(
name=name, version_name=version_name, document_type=document_type, content=content,
document_format=document_format, target_type=target_type, tags=tags) requires=requires,
attachments=attachments,
name=name,
version_name=version_name,
document_type=document_type,
document_format=document_format,
target_type=target_type,
tags=tags,
)
return json.dumps({ return json.dumps({"DocumentDescription": result})
'DocumentDescription': result
})
def delete_document(self): def delete_document(self):
name = self._get_param("Name") name = self._get_param("Name")
document_version = self._get_param("DocumentVersion") document_version = self._get_param("DocumentVersion")
version_name = self._get_param("VersionName") version_name = self._get_param("VersionName")
force = self._get_param("Force", False) force = self._get_param("Force", False)
self.ssm_backend.delete_document(name=name, document_version=document_version, self.ssm_backend.delete_document(
version_name=version_name, force=force) name=name,
document_version=document_version,
version_name=version_name,
force=force,
)
return json.dumps({}) return json.dumps({})
@ -52,8 +62,12 @@ class SimpleSystemManagerResponse(BaseResponse):
document_version = self._get_param("DocumentVersion") document_version = self._get_param("DocumentVersion")
document_format = self._get_param("DocumentFormat", "JSON") document_format = self._get_param("DocumentFormat", "JSON")
document = self.ssm_backend.get_document(name=name, document_version=document_version, document = self.ssm_backend.get_document(
document_format=document_format, version_name=version_name) name=name,
document_version=document_version,
document_format=document_format,
version_name=version_name,
)
return json.dumps(document) return json.dumps(document)
@ -62,12 +76,11 @@ class SimpleSystemManagerResponse(BaseResponse):
document_version = self._get_param("DocumentVersion") document_version = self._get_param("DocumentVersion")
version_name = self._get_param("VersionName") version_name = self._get_param("VersionName")
result = self.ssm_backend.describe_document(name=name, document_version=document_version, result = self.ssm_backend.describe_document(
version_name=version_name) name=name, document_version=document_version, version_name=version_name
)
return json.dumps({ return json.dumps({"Document": result})
'Document': result
})
def update_document(self): def update_document(self):
content = self._get_param("Content") content = self._get_param("Content")
@ -78,22 +91,26 @@ class SimpleSystemManagerResponse(BaseResponse):
document_format = self._get_param("DocumentFormat", "JSON") document_format = self._get_param("DocumentFormat", "JSON")
target_type = self._get_param("TargetType") target_type = self._get_param("TargetType")
result = self.ssm_backend.update_document(content=content, attachments=attachments, name=name, result = self.ssm_backend.update_document(
version_name=version_name, document_version=document_version, content=content,
document_format=document_format, target_type=target_type) attachments=attachments,
name=name,
version_name=version_name,
document_version=document_version,
document_format=document_format,
target_type=target_type,
)
return json.dumps({ return json.dumps({"DocumentDescription": result})
'DocumentDescription': result
})
def update_document_default_version(self): def update_document_default_version(self):
name = self._get_param("Name") name = self._get_param("Name")
document_version = self._get_param("DocumentVersion") document_version = self._get_param("DocumentVersion")
result = self.ssm_backend.update_document_default_version(name=name, document_version=document_version) result = self.ssm_backend.update_document_default_version(
return json.dumps({ name=name, document_version=document_version
'Description': result )
}) return json.dumps({"Description": result})
def list_documents(self): def list_documents(self):
document_filter_list = self._get_param("DocumentFilterList") document_filter_list = self._get_param("DocumentFilterList")
@ -101,13 +118,14 @@ class SimpleSystemManagerResponse(BaseResponse):
max_results = self._get_param("MaxResults", 10) max_results = self._get_param("MaxResults", 10)
next_token = self._get_param("NextToken", "0") next_token = self._get_param("NextToken", "0")
documents, token = self.ssm_backend.list_documents(document_filter_list=document_filter_list, filters=filters, documents, token = self.ssm_backend.list_documents(
max_results=max_results, next_token=next_token) document_filter_list=document_filter_list,
filters=filters,
max_results=max_results,
next_token=next_token,
)
return json.dumps({ return json.dumps({"DocumentIdentifiers": documents, "NextToken": token})
"DocumentIdentifiers": documents,
"NextToken": token
})
def _get_param(self, param, default=None): def _get_param(self, param, default=None):
return self.request_params.get(param, default) return self.request_params.get(param, default)

420
tests/test_ssm/test_ssm_docs.py
View File

@ -21,18 +21,29 @@ from moto import mock_ssm, mock_cloudformation
def _get_yaml_template(): def _get_yaml_template():
template_path = '/'.join(['test_ssm', 'test_templates', 'good.yaml']) template_path = "/".join(["test_ssm", "test_templates", "good.yaml"])
resource_path = pkg_resources.resource_string('tests', template_path) resource_path = pkg_resources.resource_string("tests", template_path)
return resource_path return resource_path
def _validate_document_description(doc_name, doc_description, json_doc, expected_document_version, def _validate_document_description(
expected_latest_version, expected_default_version, expected_format): doc_name,
doc_description,
json_doc,
expected_document_version,
expected_latest_version,
expected_default_version,
expected_format,
):
if expected_format == "JSON": if expected_format == "JSON":
doc_description["Hash"].should.equal(hashlib.sha256(json.dumps(json_doc).encode('utf-8')).hexdigest()) doc_description["Hash"].should.equal(
hashlib.sha256(json.dumps(json_doc).encode("utf-8")).hexdigest()
)
else: else:
doc_description["Hash"].should.equal(hashlib.sha256(yaml.dump(json_doc).encode('utf-8')).hexdigest()) doc_description["Hash"].should.equal(
hashlib.sha256(yaml.dump(json_doc).encode("utf-8")).hexdigest()
)
doc_description["HashType"].should.equal("Sha256") doc_description["HashType"].should.equal("Sha256")
doc_description["Name"].should.equal(doc_name) doc_description["Name"].should.equal(doc_name)
@ -63,7 +74,7 @@ def _validate_document_description(doc_name, doc_description, json_doc, expected
doc_description["Parameters"][3]["Name"].should.equal("Parameter4") doc_description["Parameters"][3]["Name"].should.equal("Parameter4")
doc_description["Parameters"][3]["Type"].should.equal("StringList") doc_description["Parameters"][3]["Type"].should.equal("StringList")
doc_description["Parameters"][3]["Description"].should.equal("A string list") doc_description["Parameters"][3]["Description"].should.equal("A string list")
doc_description["Parameters"][3]["DefaultValue"].should.equal("[\"abc\", \"def\"]") doc_description["Parameters"][3]["DefaultValue"].should.equal('["abc", "def"]')
doc_description["Parameters"][4]["Name"].should.equal("Parameter5") doc_description["Parameters"][4]["Name"].should.equal("Parameter5")
doc_description["Parameters"][4]["Type"].should.equal("StringMap") doc_description["Parameters"][4]["Type"].should.equal("StringMap")
@ -74,22 +85,32 @@ def _validate_document_description(doc_name, doc_description, json_doc, expected
if expected_format == "JSON": if expected_format == "JSON":
# We have to replace single quotes from the response to package it back up # We have to replace single quotes from the response to package it back up
json.loads(doc_description["Parameters"][4]["DefaultValue"]).should.equal( json.loads(doc_description["Parameters"][4]["DefaultValue"]).should.equal(
{'NotificationArn': '$dependency.topicArn', {
'NotificationEvents': ['Failed'], "NotificationArn": "$dependency.topicArn",
'NotificationType': 'Command'}) "NotificationEvents": ["Failed"],
"NotificationType": "Command",
}
)
json.loads(doc_description["Parameters"][5]["DefaultValue"]).should.equal( json.loads(doc_description["Parameters"][5]["DefaultValue"]).should.equal(
[{'DeviceName': '/dev/sda1', 'Ebs': {'VolumeSize': '50'}}, [
{'DeviceName': '/dev/sdm', 'Ebs': {'VolumeSize': '100'}}] {"DeviceName": "/dev/sda1", "Ebs": {"VolumeSize": "50"}},
{"DeviceName": "/dev/sdm", "Ebs": {"VolumeSize": "100"}},
]
) )
else: else:
yaml.safe_load(doc_description["Parameters"][4]["DefaultValue"]).should.equal( yaml.safe_load(doc_description["Parameters"][4]["DefaultValue"]).should.equal(
{'NotificationArn': '$dependency.topicArn', {
'NotificationEvents': ['Failed'], "NotificationArn": "$dependency.topicArn",
'NotificationType': 'Command'}) "NotificationEvents": ["Failed"],
"NotificationType": "Command",
}
)
yaml.safe_load(doc_description["Parameters"][5]["DefaultValue"]).should.equal( yaml.safe_load(doc_description["Parameters"][5]["DefaultValue"]).should.equal(
[{'DeviceName': '/dev/sda1', 'Ebs': {'VolumeSize': '50'}}, [
{'DeviceName': '/dev/sdm', 'Ebs': {'VolumeSize': '100'}}] {"DeviceName": "/dev/sda1", "Ebs": {"VolumeSize": "50"}},
{"DeviceName": "/dev/sdm", "Ebs": {"VolumeSize": "100"}},
]
) )
doc_description["DocumentType"].should.equal("Command") doc_description["DocumentType"].should.equal("Command")
@ -98,7 +119,10 @@ def _validate_document_description(doc_name, doc_description, json_doc, expected
doc_description["DefaultVersion"].should.equal(expected_default_version) doc_description["DefaultVersion"].should.equal(expected_default_version)
doc_description["DocumentFormat"].should.equal(expected_format) doc_description["DocumentFormat"].should.equal(expected_format)
def _get_doc_validator(response, version_name, doc_version, json_doc_content, document_format):
def _get_doc_validator(
response, version_name, doc_version, json_doc_content, document_format
):
response["Name"].should.equal("TestDocument3") response["Name"].should.equal("TestDocument3")
if version_name: if version_name:
response["VersionName"].should.equal(version_name) response["VersionName"].should.equal(version_name)
@ -111,6 +135,7 @@ def _get_doc_validator(response, version_name, doc_version, json_doc_content, do
response["DocumentType"].should.equal("Command") response["DocumentType"].should.equal("Command")
response["DocumentFormat"].should.equal(document_format) response["DocumentFormat"].should.equal(document_format)
# Done # Done
@mock_ssm @mock_ssm
def test_create_document(): def test_create_document():
@ -120,27 +145,45 @@ def test_create_document():
client = boto3.client("ssm", region_name="us-east-1") client = boto3.client("ssm", region_name="us-east-1")
response = client.create_document( response = client.create_document(
Content=yaml.dump(json_doc), Name="TestDocument", DocumentType="Command", DocumentFormat="YAML" Content=yaml.dump(json_doc),
Name="TestDocument",
DocumentType="Command",
DocumentFormat="YAML",
) )
doc_description = response["DocumentDescription"] doc_description = response["DocumentDescription"]
_validate_document_description("TestDocument", doc_description, json_doc, "1", "1", "1", "YAML") _validate_document_description(
"TestDocument", doc_description, json_doc, "1", "1", "1", "YAML"
)
response = client.create_document( response = client.create_document(
Content=json.dumps(json_doc), Name="TestDocument2", DocumentType="Command", DocumentFormat="JSON" Content=json.dumps(json_doc),
Name="TestDocument2",
DocumentType="Command",
DocumentFormat="JSON",
) )
doc_description = response["DocumentDescription"] doc_description = response["DocumentDescription"]
_validate_document_description("TestDocument2", doc_description, json_doc, "1", "1", "1", "JSON") _validate_document_description(
"TestDocument2", doc_description, json_doc, "1", "1", "1", "JSON"
)
response = client.create_document( response = client.create_document(
Content=json.dumps(json_doc), Name="TestDocument3", DocumentType="Command", DocumentFormat="JSON", Content=json.dumps(json_doc),
VersionName="Base", TargetType="/AWS::EC2::Instance", Tags=[{'Key': 'testing', 'Value': 'testingValue'}] Name="TestDocument3",
DocumentType="Command",
DocumentFormat="JSON",
VersionName="Base",
TargetType="/AWS::EC2::Instance",
Tags=[{"Key": "testing", "Value": "testingValue"}],
) )
doc_description = response["DocumentDescription"] doc_description = response["DocumentDescription"]
doc_description["VersionName"].should.equal("Base") doc_description["VersionName"].should.equal("Base")
doc_description["TargetType"].should.equal("/AWS::EC2::Instance") doc_description["TargetType"].should.equal("/AWS::EC2::Instance")
doc_description["Tags"].should.equal([{'Key': 'testing', 'Value': 'testingValue'}]) doc_description["Tags"].should.equal([{"Key": "testing", "Value": "testingValue"}])
_validate_document_description(
"TestDocument3", doc_description, json_doc, "1", "1", "1", "JSON"
)
_validate_document_description("TestDocument3", doc_description, json_doc, "1", "1", "1", "JSON")
# Done # Done
@mock_ssm @mock_ssm
@ -155,18 +198,26 @@ def test_get_document():
raise RuntimeError("Should fail") raise RuntimeError("Should fail")
except botocore.exceptions.ClientError as err: except botocore.exceptions.ClientError as err:
err.operation_name.should.equal("GetDocument") err.operation_name.should.equal("GetDocument")
err.response["Error"]["Message"].should.equal("The specified document does not exist.") err.response["Error"]["Message"].should.equal(
"The specified document does not exist."
)
client.create_document( client.create_document(
Content=yaml.dump(json_doc), Name="TestDocument3", DocumentType="Command", DocumentFormat="YAML", Content=yaml.dump(json_doc),
VersionName="Base" Name="TestDocument3",
DocumentType="Command",
DocumentFormat="YAML",
VersionName="Base",
) )
new_json_doc = copy.copy(json_doc) new_json_doc = copy.copy(json_doc)
new_json_doc['description'] = "a new description" new_json_doc["description"] = "a new description"
client.update_document( client.update_document(
Content=json.dumps(new_json_doc), Name="TestDocument3", DocumentVersion="$LATEST", VersionName="NewBase" Content=json.dumps(new_json_doc),
Name="TestDocument3",
DocumentVersion="$LATEST",
VersionName="NewBase",
) )
response = client.get_document(Name="TestDocument3") response = client.get_document(Name="TestDocument3")
@ -190,32 +241,38 @@ def test_get_document():
response = client.get_document(Name="TestDocument3", VersionName="NewBase") response = client.get_document(Name="TestDocument3", VersionName="NewBase")
_get_doc_validator(response, "NewBase", "2", new_json_doc, "JSON") _get_doc_validator(response, "NewBase", "2", new_json_doc, "JSON")
response = client.get_document(Name="TestDocument3", VersionName="NewBase", DocumentVersion="2") response = client.get_document(
Name="TestDocument3", VersionName="NewBase", DocumentVersion="2"
)
_get_doc_validator(response, "NewBase", "2", new_json_doc, "JSON") _get_doc_validator(response, "NewBase", "2", new_json_doc, "JSON")
try: try:
response = client.get_document(Name="TestDocument3", VersionName="BadName", DocumentVersion="2") response = client.get_document(
Name="TestDocument3", VersionName="BadName", DocumentVersion="2"
)
raise RuntimeError("Should fail") raise RuntimeError("Should fail")
except botocore.exceptions.ClientError as err: except botocore.exceptions.ClientError as err:
err.operation_name.should.equal("GetDocument") err.operation_name.should.equal("GetDocument")
err.response["Error"]["Message"].should.equal("The specified document does not exist.") err.response["Error"]["Message"].should.equal(
"The specified document does not exist."
)
try: try:
response = client.get_document(Name="TestDocument3", DocumentVersion="3") response = client.get_document(Name="TestDocument3", DocumentVersion="3")
raise RuntimeError("Should fail") raise RuntimeError("Should fail")
except botocore.exceptions.ClientError as err: except botocore.exceptions.ClientError as err:
err.operation_name.should.equal("GetDocument") err.operation_name.should.equal("GetDocument")
err.response["Error"]["Message"].should.equal("The specified document does not exist.") err.response["Error"]["Message"].should.equal(
"The specified document does not exist."
)
# Updating default should update normal get # Updating default should update normal get
client.update_document_default_version( client.update_document_default_version(Name="TestDocument3", DocumentVersion="2")
Name="TestDocument3",
DocumentVersion="2"
)
response = client.get_document(Name="TestDocument3", DocumentFormat="JSON") response = client.get_document(Name="TestDocument3", DocumentFormat="JSON")
_get_doc_validator(response, "NewBase", "2", new_json_doc, "JSON") _get_doc_validator(response, "NewBase", "2", new_json_doc, "JSON")
@mock_ssm @mock_ssm
def test_delete_document(): def test_delete_document():
template_file = _get_yaml_template() template_file = _get_yaml_template()
@ -227,12 +284,18 @@ def test_delete_document():
raise RuntimeError("Should fail") raise RuntimeError("Should fail")
except botocore.exceptions.ClientError as err: except botocore.exceptions.ClientError as err:
err.operation_name.should.equal("DeleteDocument") err.operation_name.should.equal("DeleteDocument")
err.response["Error"]["Message"].should.equal("The specified document does not exist.") err.response["Error"]["Message"].should.equal(
"The specified document does not exist."
)
# Test simple # Test simple
client.create_document( client.create_document(
Content=yaml.dump(json_doc), Name="TestDocument3", DocumentType="Command", DocumentFormat="YAML", Content=yaml.dump(json_doc),
VersionName="Base", TargetType="/AWS::EC2::Instance" Name="TestDocument3",
DocumentType="Command",
DocumentFormat="YAML",
VersionName="Base",
TargetType="/AWS::EC2::Instance",
) )
client.delete_document(Name="TestDocument3") client.delete_document(Name="TestDocument3")
@ -241,51 +304,68 @@ def test_delete_document():
raise RuntimeError("Should fail") raise RuntimeError("Should fail")
except botocore.exceptions.ClientError as err: except botocore.exceptions.ClientError as err:
err.operation_name.should.equal("GetDocument") err.operation_name.should.equal("GetDocument")
err.response["Error"]["Message"].should.equal("The specified document does not exist.") err.response["Error"]["Message"].should.equal(
"The specified document does not exist."
)
# Delete default version with other version is bad # Delete default version with other version is bad
client.create_document( client.create_document(
Content=yaml.dump(json_doc), Name="TestDocument3", DocumentType="Command", DocumentFormat="YAML", Content=yaml.dump(json_doc),
VersionName="Base", TargetType="/AWS::EC2::Instance" Name="TestDocument3",
DocumentType="Command",
DocumentFormat="YAML",
VersionName="Base",
TargetType="/AWS::EC2::Instance",
) )
new_json_doc = copy.copy(json_doc) new_json_doc = copy.copy(json_doc)
new_json_doc['description'] = "a new description" new_json_doc["description"] = "a new description"
client.update_document( client.update_document(
Content=json.dumps(new_json_doc), Name="TestDocument3", DocumentVersion="$LATEST", VersionName="NewBase" Content=json.dumps(new_json_doc),
Name="TestDocument3",
DocumentVersion="$LATEST",
VersionName="NewBase",
) )
new_json_doc['description'] = "a new description2" new_json_doc["description"] = "a new description2"
client.update_document( client.update_document(
Content=json.dumps(new_json_doc), Name="TestDocument3", DocumentVersion="$LATEST" Content=json.dumps(new_json_doc),
Name="TestDocument3",
DocumentVersion="$LATEST",
) )
new_json_doc['description'] = "a new description3" new_json_doc["description"] = "a new description3"
client.update_document( client.update_document(
Content=json.dumps(new_json_doc), Name="TestDocument3", DocumentVersion="$LATEST" Content=json.dumps(new_json_doc),
Name="TestDocument3",
DocumentVersion="$LATEST",
) )
new_json_doc['description'] = "a new description4" new_json_doc["description"] = "a new description4"
client.update_document( client.update_document(
Content=json.dumps(new_json_doc), Name="TestDocument3", DocumentVersion="$LATEST" Content=json.dumps(new_json_doc),
Name="TestDocument3",
DocumentVersion="$LATEST",
) )
try: try:
client.delete_document(Name="TestDocument3", DocumentVersion="1") client.delete_document(Name="TestDocument3", DocumentVersion="1")
raise RuntimeError("Should fail") raise RuntimeError("Should fail")
except botocore.exceptions.ClientError as err: except botocore.exceptions.ClientError as err:
err.operation_name.should.equal("DeleteDocument") err.operation_name.should.equal("DeleteDocument")
err.response["Error"]["Message"].should.equal("Default version of the document can't be deleted.") err.response["Error"]["Message"].should.equal(
"Default version of the document can't be deleted."
)
try: try:
client.delete_document(Name="TestDocument3", VersionName="Base") client.delete_document(Name="TestDocument3", VersionName="Base")
raise RuntimeError("Should fail") raise RuntimeError("Should fail")
except botocore.exceptions.ClientError as err: except botocore.exceptions.ClientError as err:
err.operation_name.should.equal("DeleteDocument") err.operation_name.should.equal("DeleteDocument")
err.response["Error"]["Message"].should.equal("Default version of the document can't be deleted.") err.response["Error"]["Message"].should.equal(
"Default version of the document can't be deleted."
)
# Make sure no ill side effects # Make sure no ill side effects
response = client.get_document(Name="TestDocument3") response = client.get_document(Name="TestDocument3")
@ -311,24 +391,31 @@ def test_delete_document():
raise RuntimeError("Should fail") raise RuntimeError("Should fail")
except botocore.exceptions.ClientError as err: except botocore.exceptions.ClientError as err:
err.operation_name.should.equal("GetDocument") err.operation_name.should.equal("GetDocument")
err.response["Error"]["Message"].should.equal("The specified document does not exist.") err.response["Error"]["Message"].should.equal(
"The specified document does not exist."
)
try: try:
client.get_document(Name="TestDocument3", DocumentVersion="3") client.get_document(Name="TestDocument3", DocumentVersion="3")
raise RuntimeError("Should fail") raise RuntimeError("Should fail")
except botocore.exceptions.ClientError as err: except botocore.exceptions.ClientError as err:
err.operation_name.should.equal("GetDocument") err.operation_name.should.equal("GetDocument")
err.response["Error"]["Message"].should.equal("The specified document does not exist.") err.response["Error"]["Message"].should.equal(
"The specified document does not exist."
)
try: try:
client.get_document(Name="TestDocument3", DocumentVersion="4") client.get_document(Name="TestDocument3", DocumentVersion="4")
raise RuntimeError("Should fail") raise RuntimeError("Should fail")
except botocore.exceptions.ClientError as err: except botocore.exceptions.ClientError as err:
err.operation_name.should.equal("GetDocument") err.operation_name.should.equal("GetDocument")
err.response["Error"]["Message"].should.equal("The specified document does not exist.") err.response["Error"]["Message"].should.equal(
"The specified document does not exist."
)
response = client.list_documents() response = client.list_documents()
len(response['DocumentIdentifiers']).should.equal(0) len(response["DocumentIdentifiers"]).should.equal(0)
# Done # Done
@mock_ssm @mock_ssm
@ -342,46 +429,55 @@ def test_update_document_default_version():
raise RuntimeError("Should fail") raise RuntimeError("Should fail")
except botocore.exceptions.ClientError as err: except botocore.exceptions.ClientError as err:
err.operation_name.should.equal("UpdateDocumentDefaultVersion") err.operation_name.should.equal("UpdateDocumentDefaultVersion")
err.response["Error"]["Message"].should.equal("The specified document does not exist.") err.response["Error"]["Message"].should.equal(
"The specified document does not exist."
)
client.create_document( client.create_document(
Content=json.dumps(json_doc), Name="TestDocument", DocumentType="Command", VersionName="Base" Content=json.dumps(json_doc),
Name="TestDocument",
DocumentType="Command",
VersionName="Base",
) )
json_doc['description'] = "a new description" json_doc["description"] = "a new description"
client.update_document( client.update_document(
Content=json.dumps(json_doc), Name="TestDocument", DocumentVersion="$LATEST", Content=json.dumps(json_doc),
DocumentFormat="JSON" Name="TestDocument",
DocumentVersion="$LATEST",
DocumentFormat="JSON",
) )
json_doc['description'] = "a new description2" json_doc["description"] = "a new description2"
client.update_document( client.update_document(
Content=json.dumps(json_doc), Name="TestDocument", DocumentVersion="$LATEST" Content=json.dumps(json_doc), Name="TestDocument", DocumentVersion="$LATEST"
) )
response = client.update_document_default_version( response = client.update_document_default_version(
Name="TestDocument", Name="TestDocument", DocumentVersion="2"
DocumentVersion="2"
) )
response["Description"]["Name"].should.equal("TestDocument") response["Description"]["Name"].should.equal("TestDocument")
response["Description"]["DefaultVersion"].should.equal("2") response["Description"]["DefaultVersion"].should.equal("2")
json_doc['description'] = "a new description3" json_doc["description"] = "a new description3"
client.update_document( client.update_document(
Content=json.dumps(json_doc), Name="TestDocument", DocumentVersion="$LATEST", VersionName="NewBase" Content=json.dumps(json_doc),
Name="TestDocument",
DocumentVersion="$LATEST",
VersionName="NewBase",
) )
response = client.update_document_default_version( response = client.update_document_default_version(
Name="TestDocument", Name="TestDocument", DocumentVersion="4"
DocumentVersion="4"
) )
response["Description"]["Name"].should.equal("TestDocument") response["Description"]["Name"].should.equal("TestDocument")
response["Description"]["DefaultVersion"].should.equal("4") response["Description"]["DefaultVersion"].should.equal("4")
response["Description"]["DefaultVersionName"].should.equal("NewBase") response["Description"]["DefaultVersionName"].should.equal("NewBase")
# Done # Done
@mock_ssm @mock_ssm
def test_update_document(): def test_update_document():
@ -391,54 +487,80 @@ def test_update_document():
client = boto3.client("ssm", region_name="us-east-1") client = boto3.client("ssm", region_name="us-east-1")
try: try:
client.update_document(Name="DNE", Content=json.dumps(json_doc), DocumentVersion="1", DocumentFormat="JSON") client.update_document(
Name="DNE",
Content=json.dumps(json_doc),
DocumentVersion="1",
DocumentFormat="JSON",
)
raise RuntimeError("Should fail") raise RuntimeError("Should fail")
except botocore.exceptions.ClientError as err: except botocore.exceptions.ClientError as err:
err.operation_name.should.equal("UpdateDocument") err.operation_name.should.equal("UpdateDocument")
err.response["Error"]["Message"].should.equal("The specified document does not exist.") err.response["Error"]["Message"].should.equal(
"The specified document does not exist."
)
client.create_document( client.create_document(
Content=json.dumps(json_doc), Name="TestDocument", DocumentType="Command", DocumentFormat="JSON", Content=json.dumps(json_doc),
VersionName="Base" Name="TestDocument",
DocumentType="Command",
DocumentFormat="JSON",
VersionName="Base",
) )
# Duplicate content throws an error # Duplicate content throws an error
try: try:
client.update_document( client.update_document(
Content=json.dumps(json_doc), Name="TestDocument", DocumentVersion="1", DocumentFormat="JSON" Content=json.dumps(json_doc),
Name="TestDocument",
DocumentVersion="1",
DocumentFormat="JSON",
) )
raise RuntimeError("Should fail") raise RuntimeError("Should fail")
except botocore.exceptions.ClientError as err: except botocore.exceptions.ClientError as err:
err.operation_name.should.equal("UpdateDocument") err.operation_name.should.equal("UpdateDocument")
err.response["Error"]["Message"].should.equal("The content of the association document matches another " err.response["Error"]["Message"].should.equal(
"document. Change the content of the document and try again.") "The content of the association document matches another "
"document. Change the content of the document and try again."
)
json_doc['description'] = "a new description" json_doc["description"] = "a new description"
# Duplicate version name # Duplicate version name
try: try:
client.update_document( client.update_document(
Content=json.dumps(json_doc), Name="TestDocument", DocumentVersion="1", DocumentFormat="JSON", Content=json.dumps(json_doc),
VersionName="Base" Name="TestDocument",
DocumentVersion="1",
DocumentFormat="JSON",
VersionName="Base",
) )
raise RuntimeError("Should fail") raise RuntimeError("Should fail")
except botocore.exceptions.ClientError as err: except botocore.exceptions.ClientError as err:
err.operation_name.should.equal("UpdateDocument") err.operation_name.should.equal("UpdateDocument")
err.response["Error"]["Message"].should.equal("The specified version name is a duplicate.") err.response["Error"]["Message"].should.equal(
"The specified version name is a duplicate."
)
response = client.update_document( response = client.update_document(
Content=json.dumps(json_doc), Name="TestDocument", VersionName="Base2", DocumentVersion="1", Content=json.dumps(json_doc),
DocumentFormat="JSON" Name="TestDocument",
VersionName="Base2",
DocumentVersion="1",
DocumentFormat="JSON",
) )
response["DocumentDescription"]["Description"].should.equal("a new description") response["DocumentDescription"]["Description"].should.equal("a new description")
response["DocumentDescription"]["DocumentVersion"].should.equal("2") response["DocumentDescription"]["DocumentVersion"].should.equal("2")
response["DocumentDescription"]["LatestVersion"].should.equal("2") response["DocumentDescription"]["LatestVersion"].should.equal("2")
response["DocumentDescription"]["DefaultVersion"].should.equal("1") response["DocumentDescription"]["DefaultVersion"].should.equal("1")
json_doc['description'] = "a new description2" json_doc["description"] = "a new description2"
response = client.update_document( response = client.update_document(
Content=json.dumps(json_doc), Name="TestDocument", DocumentVersion="$LATEST", Content=json.dumps(json_doc),
DocumentFormat="JSON", VersionName="NewBase" Name="TestDocument",
DocumentVersion="$LATEST",
DocumentFormat="JSON",
VersionName="NewBase",
) )
response["DocumentDescription"]["Description"].should.equal("a new description2") response["DocumentDescription"]["Description"].should.equal("a new description2")
response["DocumentDescription"]["DocumentVersion"].should.equal("3") response["DocumentDescription"]["DocumentVersion"].should.equal("3")
@ -446,6 +568,7 @@ def test_update_document():
response["DocumentDescription"]["DefaultVersion"].should.equal("1") response["DocumentDescription"]["DefaultVersion"].should.equal("1")
response["DocumentDescription"]["VersionName"].should.equal("NewBase") response["DocumentDescription"]["VersionName"].should.equal("NewBase")
# Done # Done
@mock_ssm @mock_ssm
def test_describe_document(): def test_describe_document():
@ -458,26 +581,38 @@ def test_describe_document():
raise RuntimeError("Should fail") raise RuntimeError("Should fail")
except botocore.exceptions.ClientError as err: except botocore.exceptions.ClientError as err:
err.operation_name.should.equal("DescribeDocument") err.operation_name.should.equal("DescribeDocument")
err.response["Error"]["Message"].should.equal("The specified document does not exist.") err.response["Error"]["Message"].should.equal(
"The specified document does not exist."
)
client.create_document( client.create_document(
Content=yaml.dump(json_doc), Name="TestDocument", DocumentType="Command", DocumentFormat="YAML", Content=yaml.dump(json_doc),
VersionName="Base", TargetType="/AWS::EC2::Instance", Tags=[{'Key': 'testing', 'Value': 'testingValue'}] Name="TestDocument",
DocumentType="Command",
DocumentFormat="YAML",
VersionName="Base",
TargetType="/AWS::EC2::Instance",
Tags=[{"Key": "testing", "Value": "testingValue"}],
) )
response = client.describe_document(Name="TestDocument") response = client.describe_document(Name="TestDocument")
doc_description=response['Document'] doc_description = response["Document"]
_validate_document_description("TestDocument", doc_description, json_doc, "1", "1", "1", "YAML") _validate_document_description(
"TestDocument", doc_description, json_doc, "1", "1", "1", "YAML"
)
# Adding update to check for issues # Adding update to check for issues
new_json_doc = copy.copy(json_doc) new_json_doc = copy.copy(json_doc)
new_json_doc['description'] = "a new description2" new_json_doc["description"] = "a new description2"
client.update_document( client.update_document(
Content=json.dumps(new_json_doc), Name="TestDocument", DocumentVersion="$LATEST" Content=json.dumps(new_json_doc), Name="TestDocument", DocumentVersion="$LATEST"
) )
response = client.describe_document(Name="TestDocument") response = client.describe_document(Name="TestDocument")
doc_description = response['Document'] doc_description = response["Document"]
_validate_document_description("TestDocument", doc_description, json_doc, "1", "2", "1", "YAML") _validate_document_description(
"TestDocument", doc_description, json_doc, "1", "2", "1", "YAML"
)
# Done # Done
@mock_ssm @mock_ssm
@ -488,70 +623,77 @@ def test_list_documents():
client = boto3.client("ssm", region_name="us-east-1") client = boto3.client("ssm", region_name="us-east-1")
client.create_document( client.create_document(
Content=json.dumps(json_doc), Name="TestDocument", DocumentType="Command", DocumentFormat="JSON" Content=json.dumps(json_doc),
Name="TestDocument",
DocumentType="Command",
DocumentFormat="JSON",
) )
client.create_document( client.create_document(
Content=json.dumps(json_doc), Name="TestDocument2", DocumentType="Command", DocumentFormat="JSON" Content=json.dumps(json_doc),
Name="TestDocument2",
DocumentType="Command",
DocumentFormat="JSON",
) )
client.create_document( client.create_document(
Content=json.dumps(json_doc), Name="TestDocument3", DocumentType="Command", DocumentFormat="JSON" Content=json.dumps(json_doc),
Name="TestDocument3",
DocumentType="Command",
DocumentFormat="JSON",
) )
response = client.list_documents() response = client.list_documents()
len(response['DocumentIdentifiers']).should.equal(3) len(response["DocumentIdentifiers"]).should.equal(3)
response['DocumentIdentifiers'][0]["Name"].should.equal("TestDocument") response["DocumentIdentifiers"][0]["Name"].should.equal("TestDocument")
response['DocumentIdentifiers'][1]["Name"].should.equal("TestDocument2") response["DocumentIdentifiers"][1]["Name"].should.equal("TestDocument2")
response['DocumentIdentifiers'][2]["Name"].should.equal("TestDocument3") response["DocumentIdentifiers"][2]["Name"].should.equal("TestDocument3")
response['NextToken'].should.equal("") response["NextToken"].should.equal("")
response = client.list_documents(MaxResults=1) response = client.list_documents(MaxResults=1)
len(response['DocumentIdentifiers']).should.equal(1) len(response["DocumentIdentifiers"]).should.equal(1)
response['DocumentIdentifiers'][0]["Name"].should.equal("TestDocument") response["DocumentIdentifiers"][0]["Name"].should.equal("TestDocument")
response['DocumentIdentifiers'][0]["DocumentVersion"].should.equal("1") response["DocumentIdentifiers"][0]["DocumentVersion"].should.equal("1")
response['NextToken'].should.equal("1") response["NextToken"].should.equal("1")
response = client.list_documents(MaxResults=1, NextToken=response['NextToken']) response = client.list_documents(MaxResults=1, NextToken=response["NextToken"])
len(response['DocumentIdentifiers']).should.equal(1) len(response["DocumentIdentifiers"]).should.equal(1)
response['DocumentIdentifiers'][0]["Name"].should.equal("TestDocument2") response["DocumentIdentifiers"][0]["Name"].should.equal("TestDocument2")
response['DocumentIdentifiers'][0]["DocumentVersion"].should.equal("1") response["DocumentIdentifiers"][0]["DocumentVersion"].should.equal("1")
response['NextToken'].should.equal("2") response["NextToken"].should.equal("2")
response = client.list_documents(MaxResults=1, NextToken=response['NextToken']) response = client.list_documents(MaxResults=1, NextToken=response["NextToken"])
len(response['DocumentIdentifiers']).should.equal(1) len(response["DocumentIdentifiers"]).should.equal(1)
response['DocumentIdentifiers'][0]["Name"].should.equal("TestDocument3") response["DocumentIdentifiers"][0]["Name"].should.equal("TestDocument3")
response['DocumentIdentifiers'][0]["DocumentVersion"].should.equal("1") response["DocumentIdentifiers"][0]["DocumentVersion"].should.equal("1")
response['NextToken'].should.equal("") response["NextToken"].should.equal("")
# making sure no bad interactions with update # making sure no bad interactions with update
json_doc['description'] = "a new description" json_doc["description"] = "a new description"
client.update_document( client.update_document(
Content=json.dumps(json_doc), Name="TestDocument", DocumentVersion="$LATEST", Content=json.dumps(json_doc),
DocumentFormat="JSON" Name="TestDocument",
DocumentVersion="$LATEST",
DocumentFormat="JSON",
) )
client.update_document( client.update_document(
Content=json.dumps(json_doc), Name="TestDocument2", DocumentVersion="$LATEST", Content=json.dumps(json_doc),
DocumentFormat="JSON" Name="TestDocument2",
DocumentVersion="$LATEST",
DocumentFormat="JSON",
) )
response = client.update_document_default_version( response = client.update_document_default_version(
Name="TestDocument", Name="TestDocument", DocumentVersion="2"
DocumentVersion="2"
) )
response = client.list_documents() response = client.list_documents()
len(response['DocumentIdentifiers']).should.equal(3) len(response["DocumentIdentifiers"]).should.equal(3)
response['DocumentIdentifiers'][0]["Name"].should.equal("TestDocument") response["DocumentIdentifiers"][0]["Name"].should.equal("TestDocument")
response['DocumentIdentifiers'][0]["DocumentVersion"].should.equal("2") response["DocumentIdentifiers"][0]["DocumentVersion"].should.equal("2")
response['DocumentIdentifiers'][1]["Name"].should.equal("TestDocument2")
response['DocumentIdentifiers'][1]["DocumentVersion"].should.equal("1")
response['DocumentIdentifiers'][2]["Name"].should.equal("TestDocument3")
response['DocumentIdentifiers'][2]["DocumentVersion"].should.equal("1")
response['NextToken'].should.equal("")
response["DocumentIdentifiers"][1]["Name"].should.equal("TestDocument2")
response["DocumentIdentifiers"][1]["DocumentVersion"].should.equal("1")
response["DocumentIdentifiers"][2]["Name"].should.equal("TestDocument3")
response["DocumentIdentifiers"][2]["DocumentVersion"].should.equal("1")
response["NextToken"].should.equal("")