From c9b38e25b80b9127b6dfa006e9e2845491db0b47 Mon Sep 17 00:00:00 2001 From: Alex Bainbridge Date: Tue, 30 Jun 2020 12:43:42 -0400 Subject: [PATCH] black linting --- moto/ssm/exceptions.py | 13 +- moto/ssm/models.py | 359 +++++++++++++++++---------- moto/ssm/responses.py | 80 +++--- tests/test_ssm/test_ssm_docs.py | 420 +++++++++++++++++++++----------- 4 files changed, 576 insertions(+), 296 deletions(-) diff --git a/moto/ssm/exceptions.py b/moto/ssm/exceptions.py index a1e129002..2e715f16a 100644 --- a/moto/ssm/exceptions.py +++ b/moto/ssm/exceptions.py @@ -73,7 +73,9 @@ class InvalidDocumentOperation(JsonRESTError): code = 400 def __init__(self, message): - super(InvalidDocumentOperation, self).__init__("InvalidDocumentOperation", message) + super(InvalidDocumentOperation, self).__init__( + "InvalidDocumentOperation", message + ) class InvalidDocumentContent(JsonRESTError): @@ -94,12 +96,15 @@ class DuplicateDocumentVersionName(JsonRESTError): code = 400 def __init__(self, message): - super(DuplicateDocumentVersionName, self).__init__("DuplicateDocumentVersionName", message) + super(DuplicateDocumentVersionName, self).__init__( + "DuplicateDocumentVersionName", message + ) class DuplicateDocumentContent(JsonRESTError): code = 400 def __init__(self, message): - super(DuplicateDocumentContent, self).__init__("DuplicateDocumentContent", message) - + super(DuplicateDocumentContent, self).__init__( + "DuplicateDocumentContent", message + ) diff --git a/moto/ssm/models.py b/moto/ssm/models.py index 3fa71b2aa..ad9806e9f 100644 --- a/moto/ssm/models.py +++ b/moto/ssm/models.py @@ -31,21 +31,21 @@ from .exceptions import ( InvalidDocumentContent, InvalidDocumentVersion, DuplicateDocumentVersionName, - DuplicateDocumentContent + DuplicateDocumentContent, ) class Parameter(BaseModel): def __init__( - self, - name, - value, - type, - description, - allowed_pattern, - keyid, - last_modified_date, - version, + self, + name, + value, + type, + description, + allowed_pattern, + keyid, + last_modified_date, + version, ): self.name = name self.type = type @@ -73,7 +73,7 @@ class Parameter(BaseModel): prefix = "kms:{}:".format(self.keyid or "default") if value.startswith(prefix): - return value[len(prefix):] + return value[len(prefix) :] def response_object(self, decrypt=False, region=None): r = { @@ -123,7 +123,11 @@ def generate_ssm_doc_param_list(parameters): final_dict["Type"] = param_info["type"] 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"]) else: final_dict["DefaultValue"] = str(param_info["default"]) @@ -134,8 +138,19 @@ def generate_ssm_doc_param_list(parameters): class Document(BaseModel): - def __init__(self, name, version_name, content, document_type, document_format, requires, attachments, - target_type, tags, document_version="1"): + def __init__( + self, + name, + version_name, + content, + document_type, + document_format, + requires, + attachments, + target_type, + tags, + document_version="1", + ): self.name = name self.version_name = version_name self.content = content @@ -155,14 +170,18 @@ class Document(BaseModel): try: content_json = json.loads(content) 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": try: content_json = yaml.safe_load(content) except yaml.YAMLError: - raise InvalidDocumentContent("The content for the document is not valid.") + raise InvalidDocumentContent( + "The content for the document is not valid." + ) else: - raise ValidationException(f'Invalid document format {document_format}') + raise ValidationException(f"Invalid document format {document_format}") self.content_json = content_json @@ -171,11 +190,17 @@ class Document(BaseModel): self.description = content_json.get("description") self.outputs = content_json.get("outputs") 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.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"] elif self.schema_version == "1.2": self.runtimeConfig = content_json.get("runtimeConfig") @@ -184,25 +209,23 @@ class Document(BaseModel): raise InvalidDocumentContent("The content for the document is not valid.") - - class Command(BaseModel): def __init__( - self, - comment="", - document_name="", - timeout_seconds=MAX_TIMEOUT_SECONDS, - instance_ids=None, - max_concurrency="", - max_errors="", - notification_config=None, - output_s3_bucket_name="", - output_s3_key_prefix="", - output_s3_region="", - parameters=None, - service_role_arn="", - targets=None, - backend_region="us-east-1", + self, + comment="", + document_name="", + timeout_seconds=MAX_TIMEOUT_SECONDS, + instance_ids=None, + max_concurrency="", + max_errors="", + notification_config=None, + output_s3_bucket_name="", + output_s3_key_prefix="", + output_s3_region="", + parameters=None, + service_role_arn="", + targets=None, + backend_region="us-east-1", ): if instance_ids is None: @@ -356,14 +379,23 @@ class Command(BaseModel): def _validate_document_format(document_format): aws_doc_formats = ["JSON", "YAML"] 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): - 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_doc_types = ["Command", "Policy", "Automation", "Session", "Package", "ApplicationConfiguration", - "ApplicationConfigurationSchema", "DeploymentStrategy", "ChangeCalendar"] + aws_doc_types = [ + "Command", + "Policy", + "Automation", + "Session", + "Package", + "ApplicationConfiguration", + "ApplicationConfigurationSchema", + "DeploymentStrategy", + "ChangeCalendar", + ] _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") 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) 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: # 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): @@ -397,7 +429,9 @@ def _document_filter_list_includes_comparator(keyed_value_list, filter): def _document_filter_match(filters, ssm_doc): 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 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): return False - elif filter["Key"] == "PlatformTypes" and not \ - _document_filter_list_includes_comparator(ssm_doc.platform_types, filter): + elif filter[ + "Key" + ] == "PlatformTypes" and not _document_filter_list_includes_comparator( + ssm_doc.platform_types, filter + ): 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 - 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 True @@ -440,10 +481,10 @@ class SimpleSystemManagerBackend(BaseBackend): 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"] base = { - "Hash": hashlib.sha256(document.content.encode('utf-8')).hexdigest(), + "Hash": hashlib.sha256(document.content.encode("utf-8")).hexdigest(), "HashType": "Sha256", "Name": document.name, "Owner": document.owner, @@ -457,7 +498,7 @@ class SimpleSystemManagerBackend(BaseBackend): "SchemaVersion": document.schema_version, "LatestVersion": latest, "DefaultVersion": default_version, - "DocumentFormat": document.document_format + "DocumentFormat": document.document_format, } if document.version_name: base["VersionName"] = document.version_name @@ -475,7 +516,7 @@ class SimpleSystemManagerBackend(BaseBackend): "Status": ssm_document.status, "Content": ssm_document.content, "DocumentType": ssm_document.document_type, - "DocumentFormat": document_format + "DocumentFormat": document_format, } if document_format == "JSON": @@ -483,7 +524,7 @@ class SimpleSystemManagerBackend(BaseBackend): elif document_format == "YAML": base["Content"] = yaml.dump(ssm_document.content_json) else: - raise ValidationException(f'Invalid document format {document_format}') + raise ValidationException(f"Invalid document format {document_format}") if ssm_document.version_name: base["VersionName"] = ssm_document.version_name @@ -501,7 +542,7 @@ class SimpleSystemManagerBackend(BaseBackend): "DocumentVersion": ssm_document.document_version, "DocumentType": ssm_document.document_type, "SchemaVersion": ssm_document.schema_version, - "DocumentFormat": ssm_document.document_format + "DocumentFormat": ssm_document.document_format, } if ssm_document.version_name: base["VersionName"] = ssm_document.version_name @@ -516,24 +557,44 @@ class SimpleSystemManagerBackend(BaseBackend): return base - def create_document(self, content, requires, 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) + def create_document( + self, + content, + requires, + 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, - document_format=document_format) + _validate_document_info( + content=content, + name=name, + document_type=document_type, + document_format=document_format, + ) if self._documents.get(ssm_document.name): raise DocumentAlreadyExists(f"The specified document already exists.") self._documents[ssm_document.name] = { - "documents": { - ssm_document.document_version: ssm_document - }, + "documents": {ssm_document.document_version: ssm_document}, "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) @@ -545,20 +606,34 @@ class SimpleSystemManagerBackend(BaseBackend): if documents: default_version = self._documents[name]["default_version"] - if documents[default_version].document_type == "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 ( + documents[default_version].document_type + == "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: - 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: # We delete only a specific version delete_doc = self._find_document(name, document_version, version_name) # we can't delete only the default version - if 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 + 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: keys_to_delete.add(delete_doc.document_version) @@ -571,8 +646,6 @@ class SimpleSystemManagerBackend(BaseBackend): for key in keys_to_delete: del self._documents[name]["documents"][key] - keys = self._documents[name]["documents"].keys() - if len(self._documents[name]["documents"].keys()) == 0: del self._documents[name] else: @@ -586,7 +659,9 @@ class SimpleSystemManagerBackend(BaseBackend): else: 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): raise InvalidDocument(f"The specified document does not exist.") @@ -595,18 +670,21 @@ class SimpleSystemManagerBackend(BaseBackend): if not version_name and not document_version: # Retrieve default version - default_version = self._documents[name]['default_version'] + default_version = self._documents[name]["default_version"] ssm_document = documents.get(default_version) elif version_name and document_version: 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 break else: for doc_version, document in documents.items(): - if document_version and doc_version == document_version : + if document_version and doc_version == document_version: ssm_document = document break if version_name and document.version_name == version_name: @@ -642,32 +720,68 @@ class SimpleSystemManagerBackend(BaseBackend): return base - def update_document(self, 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) + def update_document( + self, + 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): raise InvalidDocument("The specified document does not exist.") - if self._documents[name]['latest_version'] != document_version and document_version != "$LATEST": - 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.") + if ( + self._documents[name]["latest_version"] != document_version + and document_version != "$LATEST" + ): + 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) - new_ssm_document = Document(name=name, version_name=version_name, content=content, - 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)) + new_ssm_document = Document( + name=name, + version_name=version_name, + content=content, + 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: - raise DuplicateDocumentContent("The content of the association document matches another document. " - "Change the content of the document and try again.") + raise DuplicateDocumentContent( + "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]["documents"][new_ssm_document.document_version] = new_ssm_document + self._documents[name]["latest_version"] = str( + 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) @@ -675,7 +789,9 @@ class SimpleSystemManagerBackend(BaseBackend): ssm_document = self._find_document(name, document_version, version_name) 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: raise ValidationException( "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 return results, str(next_token + len(results)) - if dummy_token_tracker < next_token: dummy_token_tracker = dummy_token_tracker + 1 continue - default_version = document_bundle['default_version'] - ssm_doc = self._documents[document_name]['documents'][default_version] + default_version = document_bundle["default_version"] + ssm_doc = self._documents[document_name]["documents"][default_version] if filters and not _document_filter_match(filters, ssm_doc): # If we have filters enabled, and we don't match them, continue @@ -871,9 +986,9 @@ class SimpleSystemManagerBackend(BaseBackend): "When using global parameters, please specify within a global namespace." ) if ( - "//" in value - or not value.startswith("/") - or not re.match("^[a-zA-Z0-9_.-/]*$", value) + "//" in value + or not value.startswith("/") + or not re.match("^[a-zA-Z0-9_.-/]*$", value) ): raise ValidationException( 'The parameter doesn\'t meet the parameter name requirements. The parameter name must begin with a forward slash "/". ' @@ -952,13 +1067,13 @@ class SimpleSystemManagerBackend(BaseBackend): return result def get_parameters_by_path( - self, - path, - with_decryption, - recursive, - filters=None, - next_token=None, - max_results=10, + self, + path, + with_decryption, + recursive, + filters=None, + next_token=None, + max_results=10, ): """Implement the get-parameters-by-path-API in the backend.""" result = [] @@ -968,10 +1083,10 @@ class SimpleSystemManagerBackend(BaseBackend): for param_name in self._parameters: if path != "/" and not param_name.startswith(path): continue - if "/" in param_name[len(path) + 1:] and not recursive: + if "/" in param_name[len(path) + 1 :] and not recursive: continue if not self._match_filters( - self.get_parameter(param_name, with_decryption), filters + self.get_parameter(param_name, with_decryption), filters ): continue result.append(self.get_parameter(param_name, with_decryption)) @@ -983,7 +1098,7 @@ class SimpleSystemManagerBackend(BaseBackend): next_token = 0 next_token = int(next_token) max_results = int(max_results) - values = values_list[next_token: next_token + max_results] + values = values_list[next_token : next_token + max_results] if len(values) == max_results: next_token = str(next_token + max_results) else: @@ -1021,7 +1136,7 @@ class SimpleSystemManagerBackend(BaseBackend): if what is None: return False elif option == "BeginsWith" and not any( - what.startswith(value) for value in values + what.startswith(value) for value in values ): return False elif option == "Equals" and not any(what == value for value in values): @@ -1030,10 +1145,10 @@ class SimpleSystemManagerBackend(BaseBackend): if any(value == "/" and len(what.split("/")) == 2 for value in values): continue elif any( - value != "/" - and what.startswith(value + "/") - and len(what.split("/")) - 1 == len(value.split("/")) - for value in values + value != "/" + and what.startswith(value + "/") + and len(what.split("/")) - 1 == len(value.split("/")) + for value in values ): continue else: @@ -1080,10 +1195,10 @@ class SimpleSystemManagerBackend(BaseBackend): invalid_labels = [] for label in labels: if ( - label.startswith("aws") - or label.startswith("ssm") - or label[:1].isdigit() - or not re.match(r"^[a-zA-z0-9_\.\-]*$", label) + label.startswith("aws") + or label.startswith("ssm") + or label[:1].isdigit() + or not re.match(r"^[a-zA-z0-9_\.\-]*$", label) ): invalid_labels.append(label) continue @@ -1113,7 +1228,7 @@ class SimpleSystemManagerBackend(BaseBackend): return [invalid_labels, version] def put_parameter( - self, name, description, value, type, allowed_pattern, keyid, overwrite + self, name, description, value, type, allowed_pattern, keyid, overwrite ): previous_parameter_versions = self._parameters[name] if len(previous_parameter_versions) == 0: diff --git a/moto/ssm/responses.py b/moto/ssm/responses.py index 6d818b065..66606c283 100644 --- a/moto/ssm/responses.py +++ b/moto/ssm/responses.py @@ -28,21 +28,31 @@ class SimpleSystemManagerResponse(BaseResponse): target_type = self._get_param("TargetType") tags = self._get_param("Tags") - result = self.ssm_backend.create_document(content=content, requires=requires, attachments=attachments, - name=name, version_name=version_name, document_type=document_type, - document_format=document_format, target_type=target_type, tags=tags) + result = self.ssm_backend.create_document( + content=content, + 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({ - 'DocumentDescription': result - }) + return json.dumps({"DocumentDescription": result}) def delete_document(self): name = self._get_param("Name") document_version = self._get_param("DocumentVersion") version_name = self._get_param("VersionName") force = self._get_param("Force", False) - self.ssm_backend.delete_document(name=name, document_version=document_version, - version_name=version_name, force=force) + self.ssm_backend.delete_document( + name=name, + document_version=document_version, + version_name=version_name, + force=force, + ) return json.dumps({}) @@ -52,8 +62,12 @@ class SimpleSystemManagerResponse(BaseResponse): document_version = self._get_param("DocumentVersion") document_format = self._get_param("DocumentFormat", "JSON") - document = self.ssm_backend.get_document(name=name, document_version=document_version, - document_format=document_format, version_name=version_name) + document = self.ssm_backend.get_document( + name=name, + document_version=document_version, + document_format=document_format, + version_name=version_name, + ) return json.dumps(document) @@ -62,12 +76,11 @@ class SimpleSystemManagerResponse(BaseResponse): document_version = self._get_param("DocumentVersion") version_name = self._get_param("VersionName") - result = self.ssm_backend.describe_document(name=name, document_version=document_version, - version_name=version_name) + result = self.ssm_backend.describe_document( + name=name, document_version=document_version, version_name=version_name + ) - return json.dumps({ - 'Document': result - }) + return json.dumps({"Document": result}) def update_document(self): content = self._get_param("Content") @@ -78,22 +91,26 @@ class SimpleSystemManagerResponse(BaseResponse): document_format = self._get_param("DocumentFormat", "JSON") target_type = self._get_param("TargetType") - result = self.ssm_backend.update_document(content=content, attachments=attachments, name=name, - version_name=version_name, document_version=document_version, - document_format=document_format, target_type=target_type) + result = self.ssm_backend.update_document( + content=content, + attachments=attachments, + name=name, + version_name=version_name, + document_version=document_version, + document_format=document_format, + target_type=target_type, + ) - return json.dumps({ - 'DocumentDescription': result - }) + return json.dumps({"DocumentDescription": result}) def update_document_default_version(self): name = self._get_param("Name") document_version = self._get_param("DocumentVersion") - result = self.ssm_backend.update_document_default_version(name=name, document_version=document_version) - return json.dumps({ - 'Description': result - }) + result = self.ssm_backend.update_document_default_version( + name=name, document_version=document_version + ) + return json.dumps({"Description": result}) def list_documents(self): document_filter_list = self._get_param("DocumentFilterList") @@ -101,13 +118,14 @@ class SimpleSystemManagerResponse(BaseResponse): max_results = self._get_param("MaxResults", 10) next_token = self._get_param("NextToken", "0") - documents, token = self.ssm_backend.list_documents(document_filter_list=document_filter_list, filters=filters, - max_results=max_results, next_token=next_token) + documents, token = self.ssm_backend.list_documents( + document_filter_list=document_filter_list, + filters=filters, + max_results=max_results, + next_token=next_token, + ) - return json.dumps({ - "DocumentIdentifiers": documents, - "NextToken": token - }) + return json.dumps({"DocumentIdentifiers": documents, "NextToken": token}) def _get_param(self, param, default=None): return self.request_params.get(param, default) diff --git a/tests/test_ssm/test_ssm_docs.py b/tests/test_ssm/test_ssm_docs.py index ac5460f9d..409a3bf95 100644 --- a/tests/test_ssm/test_ssm_docs.py +++ b/tests/test_ssm/test_ssm_docs.py @@ -21,18 +21,29 @@ from moto import mock_ssm, mock_cloudformation def _get_yaml_template(): - template_path = '/'.join(['test_ssm', 'test_templates', 'good.yaml']) - resource_path = pkg_resources.resource_string('tests', template_path) + template_path = "/".join(["test_ssm", "test_templates", "good.yaml"]) + resource_path = pkg_resources.resource_string("tests", template_path) return resource_path -def _validate_document_description(doc_name, doc_description, json_doc, expected_document_version, - expected_latest_version, expected_default_version, expected_format): +def _validate_document_description( + doc_name, + doc_description, + json_doc, + expected_document_version, + expected_latest_version, + expected_default_version, + expected_format, +): 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: - 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["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]["Type"].should.equal("StringList") 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]["Type"].should.equal("StringMap") @@ -74,22 +85,32 @@ def _validate_document_description(doc_name, doc_description, json_doc, expected if expected_format == "JSON": # We have to replace single quotes from the response to package it back up json.loads(doc_description["Parameters"][4]["DefaultValue"]).should.equal( - {'NotificationArn': '$dependency.topicArn', - 'NotificationEvents': ['Failed'], - 'NotificationType': 'Command'}) + { + "NotificationArn": "$dependency.topicArn", + "NotificationEvents": ["Failed"], + "NotificationType": "Command", + } + ) 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: yaml.safe_load(doc_description["Parameters"][4]["DefaultValue"]).should.equal( - {'NotificationArn': '$dependency.topicArn', - 'NotificationEvents': ['Failed'], - 'NotificationType': 'Command'}) + { + "NotificationArn": "$dependency.topicArn", + "NotificationEvents": ["Failed"], + "NotificationType": "Command", + } + ) 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") @@ -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["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") if 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["DocumentFormat"].should.equal(document_format) + # Done @mock_ssm def test_create_document(): @@ -120,27 +145,45 @@ def test_create_document(): client = boto3.client("ssm", region_name="us-east-1") 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"] - _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( - 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"] - _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( - Content=json.dumps(json_doc), Name="TestDocument3", DocumentType="Command", DocumentFormat="JSON", - VersionName="Base", TargetType="/AWS::EC2::Instance", Tags=[{'Key': 'testing', 'Value': 'testingValue'}] + Content=json.dumps(json_doc), + Name="TestDocument3", + DocumentType="Command", + DocumentFormat="JSON", + VersionName="Base", + TargetType="/AWS::EC2::Instance", + Tags=[{"Key": "testing", "Value": "testingValue"}], ) doc_description = response["DocumentDescription"] doc_description["VersionName"].should.equal("Base") 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 @mock_ssm @@ -155,18 +198,26 @@ def test_get_document(): raise RuntimeError("Should fail") except botocore.exceptions.ClientError as err: 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( - Content=yaml.dump(json_doc), Name="TestDocument3", DocumentType="Command", DocumentFormat="YAML", - VersionName="Base" + Content=yaml.dump(json_doc), + Name="TestDocument3", + DocumentType="Command", + DocumentFormat="YAML", + VersionName="Base", ) new_json_doc = copy.copy(json_doc) - new_json_doc['description'] = "a new description" + new_json_doc["description"] = "a new description" 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") @@ -190,32 +241,38 @@ def test_get_document(): response = client.get_document(Name="TestDocument3", VersionName="NewBase") _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") 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") except botocore.exceptions.ClientError as err: 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: response = client.get_document(Name="TestDocument3", DocumentVersion="3") raise RuntimeError("Should fail") except botocore.exceptions.ClientError as err: 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 - client.update_document_default_version( - Name="TestDocument3", - DocumentVersion="2" - ) + client.update_document_default_version(Name="TestDocument3", DocumentVersion="2") response = client.get_document(Name="TestDocument3", DocumentFormat="JSON") _get_doc_validator(response, "NewBase", "2", new_json_doc, "JSON") + @mock_ssm def test_delete_document(): template_file = _get_yaml_template() @@ -227,12 +284,18 @@ def test_delete_document(): raise RuntimeError("Should fail") except botocore.exceptions.ClientError as err: 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 client.create_document( - Content=yaml.dump(json_doc), Name="TestDocument3", DocumentType="Command", DocumentFormat="YAML", - VersionName="Base", TargetType="/AWS::EC2::Instance" + Content=yaml.dump(json_doc), + Name="TestDocument3", + DocumentType="Command", + DocumentFormat="YAML", + VersionName="Base", + TargetType="/AWS::EC2::Instance", ) client.delete_document(Name="TestDocument3") @@ -241,51 +304,68 @@ def test_delete_document(): raise RuntimeError("Should fail") except botocore.exceptions.ClientError as err: 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 client.create_document( - Content=yaml.dump(json_doc), Name="TestDocument3", DocumentType="Command", DocumentFormat="YAML", - VersionName="Base", TargetType="/AWS::EC2::Instance" + Content=yaml.dump(json_doc), + Name="TestDocument3", + DocumentType="Command", + DocumentFormat="YAML", + VersionName="Base", + TargetType="/AWS::EC2::Instance", ) new_json_doc = copy.copy(json_doc) - new_json_doc['description'] = "a new description" + new_json_doc["description"] = "a new description" 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( - 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( - 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( - Content=json.dumps(new_json_doc), Name="TestDocument3", DocumentVersion="$LATEST" + Content=json.dumps(new_json_doc), + Name="TestDocument3", + DocumentVersion="$LATEST", ) - try: client.delete_document(Name="TestDocument3", DocumentVersion="1") raise RuntimeError("Should fail") except botocore.exceptions.ClientError as err: 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: client.delete_document(Name="TestDocument3", VersionName="Base") raise RuntimeError("Should fail") except botocore.exceptions.ClientError as err: 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 response = client.get_document(Name="TestDocument3") @@ -311,24 +391,31 @@ def test_delete_document(): raise RuntimeError("Should fail") except botocore.exceptions.ClientError as err: 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: client.get_document(Name="TestDocument3", DocumentVersion="3") raise RuntimeError("Should fail") except botocore.exceptions.ClientError as err: 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: client.get_document(Name="TestDocument3", DocumentVersion="4") raise RuntimeError("Should fail") except botocore.exceptions.ClientError as err: 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() - len(response['DocumentIdentifiers']).should.equal(0) + len(response["DocumentIdentifiers"]).should.equal(0) + # Done @mock_ssm @@ -342,46 +429,55 @@ def test_update_document_default_version(): raise RuntimeError("Should fail") except botocore.exceptions.ClientError as err: 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( - 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( - Content=json.dumps(json_doc), Name="TestDocument", DocumentVersion="$LATEST", - DocumentFormat="JSON" + Content=json.dumps(json_doc), + Name="TestDocument", + DocumentVersion="$LATEST", + DocumentFormat="JSON", ) - json_doc['description'] = "a new description2" + json_doc["description"] = "a new description2" client.update_document( Content=json.dumps(json_doc), Name="TestDocument", DocumentVersion="$LATEST" ) response = client.update_document_default_version( - Name="TestDocument", - DocumentVersion="2" + Name="TestDocument", DocumentVersion="2" ) response["Description"]["Name"].should.equal("TestDocument") response["Description"]["DefaultVersion"].should.equal("2") - json_doc['description'] = "a new description3" + json_doc["description"] = "a new description3" 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( - Name="TestDocument", - DocumentVersion="4" + Name="TestDocument", DocumentVersion="4" ) response["Description"]["Name"].should.equal("TestDocument") response["Description"]["DefaultVersion"].should.equal("4") response["Description"]["DefaultVersionName"].should.equal("NewBase") + # Done @mock_ssm def test_update_document(): @@ -391,54 +487,80 @@ def test_update_document(): client = boto3.client("ssm", region_name="us-east-1") 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") except botocore.exceptions.ClientError as err: 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( - Content=json.dumps(json_doc), Name="TestDocument", DocumentType="Command", DocumentFormat="JSON", - VersionName="Base" + Content=json.dumps(json_doc), + Name="TestDocument", + DocumentType="Command", + DocumentFormat="JSON", + VersionName="Base", ) # Duplicate content throws an error try: 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") except botocore.exceptions.ClientError as err: err.operation_name.should.equal("UpdateDocument") - err.response["Error"]["Message"].should.equal("The content of the association document matches another " - "document. Change the content of the document and try again.") + err.response["Error"]["Message"].should.equal( + "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 try: client.update_document( - Content=json.dumps(json_doc), Name="TestDocument", DocumentVersion="1", DocumentFormat="JSON", - VersionName="Base" + Content=json.dumps(json_doc), + Name="TestDocument", + DocumentVersion="1", + DocumentFormat="JSON", + VersionName="Base", ) raise RuntimeError("Should fail") except botocore.exceptions.ClientError as err: 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( - Content=json.dumps(json_doc), Name="TestDocument", VersionName="Base2", DocumentVersion="1", - DocumentFormat="JSON" + Content=json.dumps(json_doc), + Name="TestDocument", + VersionName="Base2", + DocumentVersion="1", + DocumentFormat="JSON", ) response["DocumentDescription"]["Description"].should.equal("a new description") response["DocumentDescription"]["DocumentVersion"].should.equal("2") response["DocumentDescription"]["LatestVersion"].should.equal("2") response["DocumentDescription"]["DefaultVersion"].should.equal("1") - json_doc['description'] = "a new description2" + json_doc["description"] = "a new description2" response = client.update_document( - Content=json.dumps(json_doc), Name="TestDocument", DocumentVersion="$LATEST", - DocumentFormat="JSON", VersionName="NewBase" + Content=json.dumps(json_doc), + Name="TestDocument", + DocumentVersion="$LATEST", + DocumentFormat="JSON", + VersionName="NewBase", ) response["DocumentDescription"]["Description"].should.equal("a new description2") response["DocumentDescription"]["DocumentVersion"].should.equal("3") @@ -446,6 +568,7 @@ def test_update_document(): response["DocumentDescription"]["DefaultVersion"].should.equal("1") response["DocumentDescription"]["VersionName"].should.equal("NewBase") + # Done @mock_ssm def test_describe_document(): @@ -458,26 +581,38 @@ def test_describe_document(): raise RuntimeError("Should fail") except botocore.exceptions.ClientError as err: 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( - Content=yaml.dump(json_doc), Name="TestDocument", DocumentType="Command", DocumentFormat="YAML", - VersionName="Base", TargetType="/AWS::EC2::Instance", Tags=[{'Key': 'testing', 'Value': 'testingValue'}] + Content=yaml.dump(json_doc), + Name="TestDocument", + DocumentType="Command", + DocumentFormat="YAML", + VersionName="Base", + TargetType="/AWS::EC2::Instance", + Tags=[{"Key": "testing", "Value": "testingValue"}], ) response = client.describe_document(Name="TestDocument") - doc_description=response['Document'] - _validate_document_description("TestDocument", doc_description, json_doc, "1", "1", "1", "YAML") + doc_description = response["Document"] + _validate_document_description( + "TestDocument", doc_description, json_doc, "1", "1", "1", "YAML" + ) # Adding update to check for issues new_json_doc = copy.copy(json_doc) - new_json_doc['description'] = "a new description2" + new_json_doc["description"] = "a new description2" client.update_document( Content=json.dumps(new_json_doc), Name="TestDocument", DocumentVersion="$LATEST" ) response = client.describe_document(Name="TestDocument") - doc_description = response['Document'] - _validate_document_description("TestDocument", doc_description, json_doc, "1", "2", "1", "YAML") + doc_description = response["Document"] + _validate_document_description( + "TestDocument", doc_description, json_doc, "1", "2", "1", "YAML" + ) + # Done @mock_ssm @@ -488,70 +623,77 @@ def test_list_documents(): client = boto3.client("ssm", region_name="us-east-1") 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( - 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( - 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() - len(response['DocumentIdentifiers']).should.equal(3) - response['DocumentIdentifiers'][0]["Name"].should.equal("TestDocument") - response['DocumentIdentifiers'][1]["Name"].should.equal("TestDocument2") - response['DocumentIdentifiers'][2]["Name"].should.equal("TestDocument3") - response['NextToken'].should.equal("") + len(response["DocumentIdentifiers"]).should.equal(3) + response["DocumentIdentifiers"][0]["Name"].should.equal("TestDocument") + response["DocumentIdentifiers"][1]["Name"].should.equal("TestDocument2") + response["DocumentIdentifiers"][2]["Name"].should.equal("TestDocument3") + response["NextToken"].should.equal("") response = client.list_documents(MaxResults=1) - len(response['DocumentIdentifiers']).should.equal(1) - response['DocumentIdentifiers'][0]["Name"].should.equal("TestDocument") - response['DocumentIdentifiers'][0]["DocumentVersion"].should.equal("1") - response['NextToken'].should.equal("1") + len(response["DocumentIdentifiers"]).should.equal(1) + response["DocumentIdentifiers"][0]["Name"].should.equal("TestDocument") + response["DocumentIdentifiers"][0]["DocumentVersion"].should.equal("1") + response["NextToken"].should.equal("1") - response = client.list_documents(MaxResults=1, NextToken=response['NextToken']) - len(response['DocumentIdentifiers']).should.equal(1) - response['DocumentIdentifiers'][0]["Name"].should.equal("TestDocument2") - response['DocumentIdentifiers'][0]["DocumentVersion"].should.equal("1") - response['NextToken'].should.equal("2") + response = client.list_documents(MaxResults=1, NextToken=response["NextToken"]) + len(response["DocumentIdentifiers"]).should.equal(1) + response["DocumentIdentifiers"][0]["Name"].should.equal("TestDocument2") + response["DocumentIdentifiers"][0]["DocumentVersion"].should.equal("1") + response["NextToken"].should.equal("2") - response = client.list_documents(MaxResults=1, NextToken=response['NextToken']) - len(response['DocumentIdentifiers']).should.equal(1) - response['DocumentIdentifiers'][0]["Name"].should.equal("TestDocument3") - response['DocumentIdentifiers'][0]["DocumentVersion"].should.equal("1") - response['NextToken'].should.equal("") + response = client.list_documents(MaxResults=1, NextToken=response["NextToken"]) + len(response["DocumentIdentifiers"]).should.equal(1) + response["DocumentIdentifiers"][0]["Name"].should.equal("TestDocument3") + response["DocumentIdentifiers"][0]["DocumentVersion"].should.equal("1") + response["NextToken"].should.equal("") # making sure no bad interactions with update - json_doc['description'] = "a new description" + json_doc["description"] = "a new description" client.update_document( - Content=json.dumps(json_doc), Name="TestDocument", DocumentVersion="$LATEST", - DocumentFormat="JSON" + Content=json.dumps(json_doc), + Name="TestDocument", + DocumentVersion="$LATEST", + DocumentFormat="JSON", ) client.update_document( - Content=json.dumps(json_doc), Name="TestDocument2", DocumentVersion="$LATEST", - DocumentFormat="JSON" + Content=json.dumps(json_doc), + Name="TestDocument2", + DocumentVersion="$LATEST", + DocumentFormat="JSON", ) response = client.update_document_default_version( - Name="TestDocument", - DocumentVersion="2" + Name="TestDocument", DocumentVersion="2" ) response = client.list_documents() - len(response['DocumentIdentifiers']).should.equal(3) - response['DocumentIdentifiers'][0]["Name"].should.equal("TestDocument") - 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("") - - - + len(response["DocumentIdentifiers"]).should.equal(3) + response["DocumentIdentifiers"][0]["Name"].should.equal("TestDocument") + 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("")