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

Techdebt: Replace sure with regular assertions in SSM (#6645)

Browse Source
This commit is contained in:
kbalk 2023-08-13 16:42:50 -04:00 committed by GitHub
parent 82363e49e9
commit 0520df02c1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 925 additions and 894 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
tests/test_ses/test_ses_sns_boto3.py
View File

@ -2,7 +2,6 @@ import json
import boto3
import sure # noqa # pylint: disable=unused-import
from moto import mock_ses, mock_sns, mock_sqs
from moto.ses.models import SESFeedback
from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID

1078
tests/test_ssm/test_ssm_boto3.py
View File

File diff suppressed because it is too large Load Diff

2
tests/test_ssm/test_ssm_cloudformation.py
View File

@ -1,6 +1,6 @@
import boto3
import json
import boto3
from moto import mock_ssm, mock_cloudformation
from tests import EXAMPLE_AMI_ID

13
tests/test_ssm/test_ssm_default_amis.py
View File

@ -1,5 +1,4 @@
import boto3
import sure # noqa # pylint: disable=unused-import
from moto import mock_ssm
@ -12,14 +11,14 @@ def test_ssm_get_latest_ami_by_path():
client = boto3.client("ssm", region_name="us-west-1")
path = "/aws/service/ami-amazon-linux-latest"
params = client.get_parameters_by_path(Path=path)["Parameters"]
params.should.have.length_of(10)
assert len(params) == 10
assert all(
[p["Name"].startswith("/aws/service/ami-amazon-linux-latest") for p in params]
{p["Name"].startswith("/aws/service/ami-amazon-linux-latest") for p in params}
)
assert all([p["Type"] == "String" for p in params])
assert all([p["DataType"] == "text" for p in params])
assert all([p["ARN"].startswith("arn:aws:ssm:us-west-1") for p in params])
assert all({p["Type"] == "String" for p in params})
assert all({p["DataType"] == "text" for p in params})
assert all({p["ARN"].startswith("arn:aws:ssm:us-west-1") for p in params})
@mock_ssm
@ -30,4 +29,4 @@ def test_ssm_latest_amis_are_different_in_regions():
client = boto3.client("ssm", region_name="eu-north-1")
ami_eunorth = client.get_parameter(Name=test_ami)["Parameter"]["Value"]
ami_uswest.shouldnt.equal(ami_eunorth)
assert ami_uswest != ami_eunorth

21
tests/test_ssm/test_ssm_defaults.py
View File

@ -1,5 +1,4 @@
import boto3
import sure # noqa # pylint: disable=unused-import
from moto import mock_ssm
from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
@ -12,15 +11,14 @@ def test_ssm_get_by_path():
params = client.get_parameters_by_path(Path=path)["Parameters"]
pacific = [p for p in params if p["Value"] == "af-south-1"][0]
pacific["Name"].should.equal(
"/aws/service/global-infrastructure/regions/af-south-1"
assert pacific["Name"] == "/aws/service/global-infrastructure/regions/af-south-1"
assert pacific["Type"] == "String"
assert pacific["Version"] == 1
assert pacific["ARN"] == (
f"arn:aws:ssm:us-west-1:{ACCOUNT_ID}:parameter/aws/service"
"/global-infrastructure/regions/af-south-1"
)
pacific["Type"].should.equal("String")
pacific["Version"].should.equal(1)
pacific["ARN"].should.equal(
f"arn:aws:ssm:us-west-1:{ACCOUNT_ID}:parameter/aws/service/global-infrastructure/regions/af-south-1"
)
pacific.should.have.key("LastModifiedDate")
assert "LastModifiedDate" in pacific
@mock_ssm
@ -28,7 +26,7 @@ def test_global_infrastructure_services():
client = boto3.client("ssm", region_name="us-west-1")
path = "/aws/service/global-infrastructure/services"
params = client.get_parameters_by_path(Path=path)["Parameters"]
params[0]["Name"].should.equal(
assert params[0]["Name"] == (
"/aws/service/global-infrastructure/services/accessanalyzer"
)
@ -41,5 +39,4 @@ def test_ssm_region_query():
)
value = param["Parameter"]["Value"]
value.should.equal("US West (N. California)")
assert value == "US West (N. California)"

54
tests/test_ssm/test_ssm_doc_permissions.py
View File

@ -1,8 +1,10 @@
import boto3
import pytest
import re
import yaml
import boto3
from botocore.exceptions import ClientError
import pytest
from moto import mock_ssm
from .test_ssm_docs import _get_yaml_template
@ -16,8 +18,8 @@ def test_describe_document_permissions_unknown_document():
Name="UnknownDocument", PermissionType="Share"
)
err = ex.value.response["Error"]
err["Code"].should.equal("InvalidDocument")
err["Message"].should.equal("The specified document does not exist.")
assert err["Code"] == "InvalidDocument"
assert err["Message"] == "The specified document does not exist."
def get_client():
@ -41,8 +43,8 @@ def test_describe_document_permissions_initial():
res = client.describe_document_permission(
Name="TestDocument", PermissionType="Share"
)
res.should.have.key("AccountIds").equal([])
res.should.have.key("AccountSharingInfoList").equal([])
assert res["AccountIds"] == []
assert res["AccountSharingInfoList"] == []
@pytest.mark.parametrize(
@ -60,14 +62,14 @@ def test_modify_document_permission_add_account_id(ids):
res = client.describe_document_permission(
Name="TestDocument", PermissionType="Share"
)
res.should.have.key("AccountIds")
set(res["AccountIds"]).should.equal(set(ids))
res.should.have.key("AccountSharingInfoList").length_of(len(ids))
assert "AccountIds" in res
assert set(res["AccountIds"]) == set(ids)
assert len(res["AccountSharingInfoList"]) == len(ids)
expected_account_sharing = [
{"AccountId": _id, "SharedDocumentVersion": "$DEFAULT"} for _id in ids
]
res.should.have.key("AccountSharingInfoList").equal(expected_account_sharing)
assert res["AccountSharingInfoList"] == expected_account_sharing
@pytest.mark.parametrize(
@ -96,15 +98,15 @@ def test_modify_document_permission_remove_account_id(initial, to_remove):
res = client.describe_document_permission(
Name="TestDocument", PermissionType="Share"
)
res.should.have.key("AccountIds")
expected_new_list = set([x for x in initial if x not in to_remove])
set(res["AccountIds"]).should.equal(expected_new_list)
assert "AccountIds" in res
expected_new_list = {x for x in initial if x not in to_remove}
assert set(res["AccountIds"]) == expected_new_list
expected_account_sharing = [
{"AccountId": _id, "SharedDocumentVersion": "$DEFAULT"}
for _id in expected_new_list
]
res.should.have.key("AccountSharingInfoList").equal(expected_account_sharing)
assert res["AccountSharingInfoList"] == expected_account_sharing
@mock_ssm
@ -115,8 +117,8 @@ def test_fail_modify_document_permission_wrong_permission_type():
Name="TestDocument", PermissionType="WrongValue", AccountIdsToAdd=[]
)
err = ex.value.response["Error"]
err["Code"].should.equal("InvalidPermissionType")
err["Message"].should.match(r"Member must satisfy enum value set: \[Share\]")
assert err["Code"] == "InvalidPermissionType"
assert re.search(r"Member must satisfy enum value set: \[Share\]", err["Message"])
@mock_ssm
@ -130,8 +132,8 @@ def test_fail_modify_document_permission_wrong_document_version():
AccountIdsToAdd=[],
)
err = ex.value.response["Error"]
err["Code"].should.equal("ValidationException")
err["Message"].should.match(r"Member must satisfy regular expression pattern")
assert err["Code"] == "ValidationException"
assert re.search(r"Member must satisfy regular expression pattern", err["Message"])
@pytest.mark.parametrize(
@ -147,8 +149,8 @@ def test_fail_modify_document_permission_add_invalid_account_ids(value):
Name="TestDocument", PermissionType="Share", AccountIdsToAdd=value
)
err = ex.value.response["Error"]
err["Code"].should.equal("ValidationException")
err["Message"].should.match(r"Member must satisfy regular expression pattern:")
assert err["Code"] == "ValidationException"
assert re.search(r"Member must satisfy regular expression pattern:", err["Message"])
@pytest.mark.parametrize(
@ -164,8 +166,8 @@ def test_fail_modify_document_permission_remove_invalid_account_ids(value):
Name="TestDocument", PermissionType="Share", AccountIdsToRemove=value
)
err = ex.value.response["Error"]
err["Code"].should.equal("ValidationException")
err["Message"].should.match(r"Member must satisfy regular expression pattern:")
assert err["Code"] == "ValidationException"
assert re.search(r"Member must satisfy regular expression pattern:", err["Message"])
@mock_ssm
@ -178,8 +180,8 @@ def test_fail_modify_document_permission_add_all_and_specific():
AccountIdsToAdd=["all", "123412341234"],
)
err = ex.value.response["Error"]
err["Code"].should.equal("DocumentPermissionLimit")
err["Message"].should.equal("Accounts can either be all or a group of AWS accounts")
assert err["Code"] == "DocumentPermissionLimit"
assert err["Message"] == "Accounts can either be all or a group of AWS accounts"
@mock_ssm
@ -192,5 +194,5 @@ def test_fail_modify_document_permission_remove_all_and_specific():
AccountIdsToRemove=["all", "123412341234"],
)
err = ex.value.response["Error"]
err["Code"].should.equal("DocumentPermissionLimit")
err["Message"].should.equal("Accounts can either be all or a group of AWS accounts")
assert err["Code"] == "DocumentPermissionLimit"
assert err["Message"] == "Accounts can either be all or a group of AWS accounts"

355
tests/test_ssm/test_ssm_docs.py
View File

@ -1,19 +1,17 @@
import boto3
import botocore.exceptions
import sure # noqa # pylint: disable=unused-import
import copy
import datetime
from datetime import timezone
import json
import yaml
import hashlib
import copy
import json
import pkgutil
import yaml
import boto3
import botocore.exceptions
from botocore.exceptions import ClientError
import pytest
from botocore.exceptions import ClientError
from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
from moto import mock_ssm
@ -32,107 +30,99 @@ def _validate_document_description(
):
if expected_format == "JSON":
doc_description["Hash"].should.equal(
assert doc_description["Hash"] == (
hashlib.sha256(json.dumps(json_doc).encode("utf-8")).hexdigest()
)
else:
doc_description["Hash"].should.equal(
assert doc_description["Hash"] == (
hashlib.sha256(yaml.dump(json_doc).encode("utf-8")).hexdigest()
)
doc_description["HashType"].should.equal("Sha256")
doc_description["Name"].should.equal(doc_name)
doc_description["Owner"].should.equal(ACCOUNT_ID)
assert doc_description["HashType"] == "Sha256"
assert doc_description["Name"] == doc_name
assert doc_description["Owner"] == ACCOUNT_ID
difference = datetime.datetime.now(tz=timezone.utc) - doc_description["CreatedDate"]
if difference.min > datetime.timedelta(minutes=1):
assert False
doc_description["Status"].should.equal("Active")
doc_description["DocumentVersion"].should.equal(expected_document_version)
doc_description["Description"].should.equal(json_doc["description"])
assert doc_description["Status"] == "Active"
assert doc_description["DocumentVersion"] == expected_document_version
assert doc_description["Description"] == json_doc["description"]
doc_description["Parameters"] = sorted(
doc_description["Parameters"], key=lambda doc: doc["Name"]
)
doc_description["Parameters"][0]["Name"].should.equal("Parameter1")
doc_description["Parameters"][0]["Type"].should.equal("Integer")
doc_description["Parameters"][0]["Description"].should.equal("Command Duration.")
doc_description["Parameters"][0]["DefaultValue"].should.equal("3")
assert doc_description["Parameters"][0]["Name"] == "Parameter1"
assert doc_description["Parameters"][0]["Type"] == "Integer"
assert doc_description["Parameters"][0]["Description"] == "Command Duration."
assert doc_description["Parameters"][0]["DefaultValue"] == "3"
doc_description["Parameters"][1]["Name"].should.equal("Parameter2")
doc_description["Parameters"][1]["Type"].should.equal("String")
doc_description["Parameters"][1]["DefaultValue"].should.equal("def")
assert doc_description["Parameters"][1]["Name"] == "Parameter2"
assert doc_description["Parameters"][1]["Type"] == "String"
assert doc_description["Parameters"][1]["DefaultValue"] == "def"
doc_description["Parameters"][2]["Name"].should.equal("Parameter3")
doc_description["Parameters"][2]["Type"].should.equal("Boolean")
doc_description["Parameters"][2]["Description"].should.equal("A boolean")
doc_description["Parameters"][2]["DefaultValue"].should.equal("False")
assert doc_description["Parameters"][2]["Name"] == "Parameter3"
assert doc_description["Parameters"][2]["Type"] == "Boolean"
assert doc_description["Parameters"][2]["Description"] == "A boolean"
assert doc_description["Parameters"][2]["DefaultValue"] == "False"
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"]')
assert doc_description["Parameters"][3]["Name"] == "Parameter4"
assert doc_description["Parameters"][3]["Type"] == "StringList"
assert doc_description["Parameters"][3]["Description"] == "A string list"
assert doc_description["Parameters"][3]["DefaultValue"] == '["abc", "def"]'
doc_description["Parameters"][4]["Name"].should.equal("Parameter5")
doc_description["Parameters"][4]["Type"].should.equal("StringMap")
assert doc_description["Parameters"][4]["Name"] == "Parameter5"
assert doc_description["Parameters"][4]["Type"] == "StringMap"
doc_description["Parameters"][5]["Name"].should.equal("Parameter6")
doc_description["Parameters"][5]["Type"].should.equal("MapList")
assert doc_description["Parameters"][5]["Name"] == "Parameter6"
assert doc_description["Parameters"][5]["Type"] == "MapList"
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",
}
)
assert json.loads(doc_description["Parameters"][4]["DefaultValue"]) == {
"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"}},
]
)
assert json.loads(doc_description["Parameters"][5]["DefaultValue"]) == [
{"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",
}
)
yaml.safe_load(doc_description["Parameters"][5]["DefaultValue"]).should.equal(
[
{"DeviceName": "/dev/sda1", "Ebs": {"VolumeSize": "50"}},
{"DeviceName": "/dev/sdm", "Ebs": {"VolumeSize": "100"}},
]
)
assert yaml.safe_load(doc_description["Parameters"][4]["DefaultValue"]) == {
"NotificationArn": "$dependency.topicArn",
"NotificationEvents": ["Failed"],
"NotificationType": "Command",
}
assert yaml.safe_load(doc_description["Parameters"][5]["DefaultValue"]) == [
{"DeviceName": "/dev/sda1", "Ebs": {"VolumeSize": "50"}},
{"DeviceName": "/dev/sdm", "Ebs": {"VolumeSize": "100"}},
]
doc_description["DocumentType"].should.equal("Command")
doc_description["SchemaVersion"].should.equal("2.2")
doc_description["LatestVersion"].should.equal(expected_latest_version)
doc_description["DefaultVersion"].should.equal(expected_default_version)
doc_description["DocumentFormat"].should.equal(expected_format)
assert doc_description["DocumentType"] == "Command"
assert doc_description["SchemaVersion"] == "2.2"
assert doc_description["LatestVersion"] == expected_latest_version
assert doc_description["DefaultVersion"] == expected_default_version
assert doc_description["DocumentFormat"] == expected_format
def _get_doc_validator(
response, version_name, doc_version, json_doc_content, document_format
):
response["Name"].should.equal("TestDocument3")
assert response["Name"] == "TestDocument3"
if version_name:
response["VersionName"].should.equal(version_name)
response["DocumentVersion"].should.equal(doc_version)
response["Status"].should.equal("Active")
assert response["VersionName"] == version_name
assert response["DocumentVersion"] == doc_version
assert response["Status"] == "Active"
if document_format == "JSON":
json.loads(response["Content"]).should.equal(json_doc_content)
assert json.loads(response["Content"]) == json_doc_content
else:
yaml.safe_load(response["Content"]).should.equal(json_doc_content)
response["DocumentType"].should.equal("Command")
response["DocumentFormat"].should.equal(document_format)
assert yaml.safe_load(response["Content"]) == json_doc_content
assert response["DocumentType"] == "Command"
assert response["DocumentFormat"] == document_format
@mock_ssm
@ -174,9 +164,9 @@ def test_create_document():
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"}])
assert doc_description["VersionName"] == "Base"
assert doc_description["TargetType"] == "/AWS::EC2::Instance"
assert doc_description["Tags"] == [{"Key": "testing", "Value": "testingValue"}]
_validate_document_description(
"TestDocument3", doc_description, json_doc, "1", "1", "1", "JSON"
@ -191,9 +181,9 @@ def test_create_document():
)
raise RuntimeError("Should fail")
except botocore.exceptions.ClientError as err:
err.operation_name.should.equal("CreateDocument")
err.response["Error"]["Message"].should.equal(
"The specified document already exists."
assert err.operation_name == "CreateDocument"
assert (
err.response["Error"]["Message"] == "The specified document already exists."
)
try:
@ -205,9 +195,10 @@ def test_create_document():
)
raise RuntimeError("Should fail")
except botocore.exceptions.ClientError as err:
err.operation_name.should.equal("CreateDocument")
err.response["Error"]["Message"].should.equal(
"The content for the document is not valid."
assert err.operation_name == "CreateDocument"
assert (
err.response["Error"]["Message"]
== "The content for the document is not valid."
)
del json_doc["parameters"]
@ -219,25 +210,25 @@ def test_create_document():
)
doc_description = response["DocumentDescription"]
doc_description["Hash"].should.equal(
assert doc_description["Hash"] == (
hashlib.sha256(yaml.dump(json_doc).encode("utf-8")).hexdigest()
)
doc_description["HashType"].should.equal("Sha256")
doc_description["Name"].should.equal("EmptyParamDoc")
doc_description["Owner"].should.equal(ACCOUNT_ID)
assert doc_description["HashType"] == "Sha256"
assert doc_description["Name"] == "EmptyParamDoc"
assert doc_description["Owner"] == ACCOUNT_ID
difference = datetime.datetime.now(tz=timezone.utc) - doc_description["CreatedDate"]
if difference.min > datetime.timedelta(minutes=1):
assert False
doc_description["Status"].should.equal("Active")
doc_description["DocumentVersion"].should.equal("1")
doc_description["Description"].should.equal(json_doc["description"])
doc_description["DocumentType"].should.equal("Command")
doc_description["SchemaVersion"].should.equal("2.2")
doc_description["LatestVersion"].should.equal("1")
doc_description["DefaultVersion"].should.equal("1")
doc_description["DocumentFormat"].should.equal("YAML")
assert doc_description["Status"] == "Active"
assert doc_description["DocumentVersion"] == "1"
assert doc_description["Description"] == json_doc["description"]
assert doc_description["DocumentType"] == "Command"
assert doc_description["SchemaVersion"] == "2.2"
assert doc_description["LatestVersion"] == "1"
assert doc_description["DefaultVersion"] == "1"
assert doc_description["DocumentFormat"] == "YAML"
@mock_ssm
@ -251,9 +242,9 @@ def test_get_document():
client.get_document(Name="DNE")
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."
assert err.operation_name == "GetDocument"
assert (
err.response["Error"]["Message"] == "The specified document does not exist."
)
client.create_document(
@ -306,18 +297,18 @@ 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."
assert err.operation_name == "GetDocument"
assert (
err.response["Error"]["Message"] == "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."
assert err.operation_name == "GetDocument"
assert (
err.response["Error"]["Message"] == "The specified document does not exist."
)
# Updating default should update normal get
@ -337,9 +328,9 @@ def test_delete_document():
client.delete_document(Name="DNE")
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."
assert err.operation_name == "DeleteDocument"
assert (
err.response["Error"]["Message"] == "The specified document does not exist."
)
# Test simple
@ -357,9 +348,9 @@ def test_delete_document():
client.get_document(Name="TestDocument3")
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."
assert err.operation_name == "GetDocument"
assert (
err.response["Error"]["Message"] == "The specified document does not exist."
)
# Delete default version with other version is bad
@ -407,8 +398,8 @@ def test_delete_document():
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(
assert err.operation_name == "DeleteDocument"
assert err.response["Error"]["Message"] == (
"Default version of the document can't be deleted."
)
@ -416,8 +407,8 @@ def test_delete_document():
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(
assert err.operation_name == "DeleteDocument"
assert err.response["Error"]["Message"] == (
"Default version of the document can't be deleted."
)
@ -429,7 +420,7 @@ def test_delete_document():
# Check that latest version is changed
response = client.describe_document(Name="TestDocument3")
response["Document"]["LatestVersion"].should.equal("4")
assert response["Document"]["LatestVersion"] == "4"
client.delete_document(Name="TestDocument3", VersionName="NewBase")
@ -444,31 +435,31 @@ def test_delete_document():
client.get_document(Name="TestDocument3", DocumentVersion="1")
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."
assert err.operation_name == "GetDocument"
assert (
err.response["Error"]["Message"] == "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."
assert err.operation_name == "GetDocument"
assert (
err.response["Error"]["Message"] == "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."
assert err.operation_name == "GetDocument"
assert (
err.response["Error"]["Message"] == "The specified document does not exist."
)
response = client.list_documents()
len(response["DocumentIdentifiers"]).should.equal(0)
assert len(response["DocumentIdentifiers"]) == 0
@mock_ssm
@ -481,9 +472,9 @@ def test_update_document_default_version():
client.update_document_default_version(Name="DNE", DocumentVersion="1")
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."
assert err.operation_name == "UpdateDocumentDefaultVersion"
assert (
err.response["Error"]["Message"] == "The specified document does not exist."
)
client.create_document(
@ -511,8 +502,8 @@ def test_update_document_default_version():
response = client.update_document_default_version(
Name="TestDocument", DocumentVersion="2"
)
response["Description"]["Name"].should.equal("TestDocument")
response["Description"]["DefaultVersion"].should.equal("2")
assert response["Description"]["Name"] == "TestDocument"
assert response["Description"]["DefaultVersion"] == "2"
json_doc["description"] = "a new description3"
@ -526,9 +517,9 @@ def test_update_document_default_version():
response = client.update_document_default_version(
Name="TestDocument", DocumentVersion="4"
)
response["Description"]["Name"].should.equal("TestDocument")
response["Description"]["DefaultVersion"].should.equal("4")
response["Description"]["DefaultVersionName"].should.equal("NewBase")
assert response["Description"]["Name"] == "TestDocument"
assert response["Description"]["DefaultVersion"] == "4"
assert response["Description"]["DefaultVersionName"] == "NewBase"
@mock_ssm
@ -547,9 +538,9 @@ def test_update_document():
)
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."
assert err.operation_name == "UpdateDocument"
assert (
err.response["Error"]["Message"] == "The specified document does not exist."
)
client.create_document(
@ -569,8 +560,8 @@ def test_update_document():
)
raise RuntimeError("Should fail")
except botocore.exceptions.ClientError as err:
err.operation_name.should.equal("UpdateDocument")
err.response["Error"]["Message"].should.equal(
assert err.operation_name == "UpdateDocument"
assert err.response["Error"]["Message"] == (
"The document version is not valid or does not exist."
)
@ -584,8 +575,8 @@ def test_update_document():
)
raise RuntimeError("Should fail")
except botocore.exceptions.ClientError as err:
err.operation_name.should.equal("UpdateDocument")
err.response["Error"]["Message"].should.equal(
assert err.operation_name == "UpdateDocument"
assert err.response["Error"]["Message"] == (
"The content of the association document matches another "
"document. Change the content of the document and try again."
)
@ -602,8 +593,8 @@ def test_update_document():
)
raise RuntimeError("Should fail")
except botocore.exceptions.ClientError as err:
err.operation_name.should.equal("UpdateDocument")
err.response["Error"]["Message"].should.equal(
assert err.operation_name == "UpdateDocument"
assert err.response["Error"]["Message"] == (
"The specified version name is a duplicate."
)
@ -614,10 +605,10 @@ def test_update_document():
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")
assert response["DocumentDescription"]["Description"] == "a new description"
assert response["DocumentDescription"]["DocumentVersion"] == "2"
assert response["DocumentDescription"]["LatestVersion"] == "2"
assert response["DocumentDescription"]["DefaultVersion"] == "1"
json_doc["description"] = "a new description2"
@ -628,11 +619,11 @@ def test_update_document():
DocumentFormat="JSON",
VersionName="NewBase",
)
response["DocumentDescription"]["Description"].should.equal("a new description2")
response["DocumentDescription"]["DocumentVersion"].should.equal("3")
response["DocumentDescription"]["LatestVersion"].should.equal("3")
response["DocumentDescription"]["DefaultVersion"].should.equal("1")
response["DocumentDescription"]["VersionName"].should.equal("NewBase")
assert response["DocumentDescription"]["Description"] == "a new description2"
assert response["DocumentDescription"]["DocumentVersion"] == "3"
assert response["DocumentDescription"]["LatestVersion"] == "3"
assert response["DocumentDescription"]["DefaultVersion"] == "1"
assert response["DocumentDescription"]["VersionName"] == "NewBase"
@mock_ssm
@ -645,9 +636,9 @@ def test_describe_document():
client.describe_document(Name="DNE")
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."
assert err.operation_name == "DescribeDocument"
assert (
err.response["Error"]["Message"] == "The specified document does not exist."
)
client.create_document(
@ -707,29 +698,29 @@ def test_list_documents():
)
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("")
assert len(response["DocumentIdentifiers"]) == 3
assert response["DocumentIdentifiers"][0]["Name"] == "TestDocument"
assert response["DocumentIdentifiers"][1]["Name"] == "TestDocument2"
assert response["DocumentIdentifiers"][2]["Name"] == "TestDocument3"
assert response["NextToken"] == ""
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")
assert len(response["DocumentIdentifiers"]) == 1
assert response["DocumentIdentifiers"][0]["Name"] == "TestDocument"
assert response["DocumentIdentifiers"][0]["DocumentVersion"] == "1"
assert response["NextToken"] == "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")
assert len(response["DocumentIdentifiers"]) == 1
assert response["DocumentIdentifiers"][0]["Name"] == "TestDocument2"
assert response["DocumentIdentifiers"][0]["DocumentVersion"] == "1"
assert response["NextToken"] == "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("")
assert len(response["DocumentIdentifiers"]) == 1
assert response["DocumentIdentifiers"][0]["Name"] == "TestDocument3"
assert response["DocumentIdentifiers"][0]["DocumentVersion"] == "1"
assert response["NextToken"] == ""
# making sure no bad interactions with update
json_doc["description"] = "a new description"
@ -750,24 +741,24 @@ def test_list_documents():
client.update_document_default_version(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")
assert len(response["DocumentIdentifiers"]) == 3
assert response["DocumentIdentifiers"][0]["Name"] == "TestDocument"
assert response["DocumentIdentifiers"][0]["DocumentVersion"] == "2"
response["DocumentIdentifiers"][1]["Name"].should.equal("TestDocument2")
response["DocumentIdentifiers"][1]["DocumentVersion"].should.equal("1")
assert response["DocumentIdentifiers"][1]["Name"] == "TestDocument2"
assert response["DocumentIdentifiers"][1]["DocumentVersion"] == "1"
response["DocumentIdentifiers"][2]["Name"].should.equal("TestDocument3")
response["DocumentIdentifiers"][2]["DocumentVersion"].should.equal("1")
response["NextToken"].should.equal("")
assert response["DocumentIdentifiers"][2]["Name"] == "TestDocument3"
assert response["DocumentIdentifiers"][2]["DocumentVersion"] == "1"
assert response["NextToken"] == ""
response = client.list_documents(Filters=[{"Key": "Owner", "Values": ["Self"]}])
len(response["DocumentIdentifiers"]).should.equal(3)
assert len(response["DocumentIdentifiers"]) == 3
response = client.list_documents(
Filters=[{"Key": "TargetType", "Values": ["/AWS::EC2::Instance"]}]
)
len(response["DocumentIdentifiers"]).should.equal(1)
assert len(response["DocumentIdentifiers"]) == 1
@mock_ssm

3
tests/test_ssm/test_ssm_ec2_integration.py
View File

@ -1,5 +1,4 @@
import boto3
import sure # noqa # pylint: disable=unused-import
from moto import mock_ec2, mock_ssm
@ -14,4 +13,4 @@ def test_ssm_get_latest_ami_by_path():
ami = ssm.get_parameter(Name=test_ami)["Parameter"]["Value"]
ec2 = boto3.client("ec2", region_name="us-east-1")
ec2.describe_images(ImageIds=[ami])["Images"].should.have.length_of(1)
assert len(ec2.describe_images(ImageIds=[ami])["Images"]) == 1

5
tests/test_ssm/test_ssm_ecs_images.py
View File

@ -1,5 +1,4 @@
import boto3
import sure # noqa # pylint: disable=unused-import
from moto import mock_ec2, mock_ssm
@ -10,10 +9,10 @@ def test_ssm_get_latest_ami_by_path():
ssm = boto3.client("ssm", region_name="us-east-1")
path = "/aws/service/ecs/optimized-ami"
params = ssm.get_parameters_by_path(Path=path, Recursive=True)["Parameters"]
params.should.have.length_of(10)
assert len(params) == 10
ec2 = boto3.client("ec2", region_name="us-east-1")
for param in params:
if "Value" in param and isinstance(param["Value"], dict):
ami = param["Value"]["image_id"]
ec2.describe_images(ImageIds=[ami])["Images"].should.have.length_of(1)
assert len(ec2.describe_images(ImageIds=[ami])["Images"]) == 1

129
tests/test_ssm/test_ssm_maintenance_windows.py
View File

@ -1,5 +1,4 @@
import boto3
import sure # noqa # pylint: disable=unused-import
from moto import mock_ssm
@ -9,12 +8,12 @@ def test_describe_maintenance_window():
ssm = boto3.client("ssm", region_name="us-east-1")
resp = ssm.describe_maintenance_windows()
resp.should.have.key("WindowIdentities").equals([])
assert resp["WindowIdentities"] == []
resp = ssm.describe_maintenance_windows(
Filters=[{"Key": "Name", "Values": ["fake-maintenance-window-name"]}]
)
resp.should.have.key("WindowIdentities").equals([])
assert resp["WindowIdentities"] == []
@mock_ssm
@ -28,25 +27,25 @@ def test_create_maintenance_windows_simple():
Cutoff=1,
AllowUnassociatedTargets=False,
)
resp.should.have.key("WindowId")
assert "WindowId" in resp
_id = resp["WindowId"] # mw-01d6bbfdf6af2c39a
resp = ssm.describe_maintenance_windows()
resp.should.have.key("WindowIdentities").have.length_of(1)
assert len(resp["WindowIdentities"]) == 1
my_window = resp["WindowIdentities"][0]
my_window.should.have.key("WindowId").equal(_id)
my_window.should.have.key("Name").equal("simple-window")
my_window.should.have.key("Enabled").equal(True)
my_window.should.have.key("Duration").equal(2)
my_window.should.have.key("Cutoff").equal(1)
my_window.should.have.key("Schedule").equal("cron(15 12 * * ? *)")
# my_window.should.have.key("NextExecutionTime")
my_window.shouldnt.have.key("Description")
my_window.shouldnt.have.key("ScheduleTimezone")
my_window.shouldnt.have.key("ScheduleOffset")
my_window.shouldnt.have.key("EndDate")
my_window.shouldnt.have.key("StartDate")
assert my_window["WindowId"] == _id
assert my_window["Name"] == "simple-window"
assert my_window["Enabled"] is True
assert my_window["Duration"] == 2
assert my_window["Cutoff"] == 1
assert my_window["Schedule"] == "cron(15 12 * * ? *)"
# assert "NextExecutionTime" in my_window
assert "Description" not in my_window
assert "ScheduleTimezone" not in my_window
assert "ScheduleOffset" not in my_window
assert "EndDate" not in my_window
assert "StartDate" not in my_window
@mock_ssm
@ -65,25 +64,25 @@ def test_create_maintenance_windows_advanced():
StartDate="2021-11-01",
EndDate="2021-12-31",
)
resp.should.have.key("WindowId")
assert "WindowId" in resp
_id = resp["WindowId"] # mw-01d6bbfdf6af2c39a
resp = ssm.describe_maintenance_windows()
resp.should.have.key("WindowIdentities").have.length_of(1)
assert len(resp["WindowIdentities"]) == 1
my_window = resp["WindowIdentities"][0]
my_window.should.have.key("WindowId").equal(_id)
my_window.should.have.key("Name").equal("simple-window")
my_window.should.have.key("Enabled").equal(True)
my_window.should.have.key("Duration").equal(5)
my_window.should.have.key("Cutoff").equal(4)
my_window.should.have.key("Schedule").equal("cron(15 12 * * ? *)")
# my_window.should.have.key("NextExecutionTime")
my_window.should.have.key("Description").equals("French windows are just too fancy")
my_window.should.have.key("ScheduleTimezone").equals("Europe/London")
my_window.should.have.key("ScheduleOffset").equals(1)
my_window.should.have.key("StartDate").equals("2021-11-01")
my_window.should.have.key("EndDate").equals("2021-12-31")
assert my_window["WindowId"] == _id
assert my_window["Name"] == "simple-window"
assert my_window["Enabled"] is True
assert my_window["Duration"] == 5
assert my_window["Cutoff"] == 4
assert my_window["Schedule"] == "cron(15 12 * * ? *)"
# assert "NextExecutionTime" in my_window
assert my_window["Description"] == "French windows are just too fancy"
assert my_window["ScheduleTimezone"] == "Europe/London"
assert my_window["ScheduleOffset"] == 1
assert my_window["StartDate"] == "2021-11-01"
assert my_window["EndDate"] == "2021-12-31"
@mock_ssm
@ -97,22 +96,22 @@ def test_get_maintenance_windows():
Cutoff=1,
AllowUnassociatedTargets=False,
)
resp.should.have.key("WindowId")
assert "WindowId" in resp
_id = resp["WindowId"] # mw-01d6bbfdf6af2c39a
my_window = ssm.get_maintenance_window(WindowId=_id)
my_window.should.have.key("WindowId").equal(_id)
my_window.should.have.key("Name").equal("my-window")
my_window.should.have.key("Enabled").equal(True)
my_window.should.have.key("Duration").equal(2)
my_window.should.have.key("Cutoff").equal(1)
my_window.should.have.key("Schedule").equal("cron(15 12 * * ? *)")
# my_window.should.have.key("NextExecutionTime")
my_window.shouldnt.have.key("Description")
my_window.shouldnt.have.key("ScheduleTimezone")
my_window.shouldnt.have.key("ScheduleOffset")
my_window.shouldnt.have.key("EndDate")
my_window.shouldnt.have.key("StartDate")
assert my_window["WindowId"] == _id
assert my_window["Name"] == "my-window"
assert my_window["Enabled"] is True
assert my_window["Duration"] == 2
assert my_window["Cutoff"] == 1
assert my_window["Schedule"] == "cron(15 12 * * ? *)"
# assert "NextExecutionTime" in my_window
assert "Description" not in my_window
assert "ScheduleTimezone" not in my_window
assert "ScheduleOffset" not in my_window
assert "EndDate" not in my_window
assert "StartDate" not in my_window
@mock_ssm
@ -129,12 +128,12 @@ def test_describe_maintenance_windows():
)
resp = ssm.describe_maintenance_windows()
resp.should.have.key("WindowIdentities").have.length_of(4)
assert len(resp["WindowIdentities"]) == 4
resp = ssm.describe_maintenance_windows(
Filters=[{"Key": "Name", "Values": ["window_0", "window_2"]}]
)
resp.should.have.key("WindowIdentities").have.length_of(2)
assert len(resp["WindowIdentities"]) == 2
@mock_ssm
@ -149,10 +148,10 @@ def test_delete_maintenance_windows():
AllowUnassociatedTargets=False,
)
ssm.delete_maintenance_window(WindowId=(resp["WindowId"]))
ssm.delete_maintenance_window(WindowId=resp["WindowId"])
resp = ssm.describe_maintenance_windows()
resp.should.have.key("WindowIdentities").equals([])
assert resp["WindowIdentities"] == []
@mock_ssm
@ -232,17 +231,17 @@ def test_register_maintenance_window_target():
ResourceType="INSTANCE",
Targets=[{"Key": "tag:Name", "Values": ["my-instance"]}],
)
resp.should.have.key("WindowTargetId")
assert "WindowTargetId" in resp
_id = resp["WindowTargetId"]
resp = ssm.describe_maintenance_window_targets(
WindowId=window_id,
)
resp.should.have.key("Targets").should.have.length_of(1)
resp["Targets"][0].should.have.key("ResourceType").equal("INSTANCE")
resp["Targets"][0].should.have.key("WindowTargetId").equal(_id)
resp["Targets"][0]["Targets"][0].should.have.key("Key").equal("tag:Name")
resp["Targets"][0]["Targets"][0].should.have.key("Values").equal(["my-instance"])
assert len(resp["Targets"]) == 1
assert resp["Targets"][0]["ResourceType"] == "INSTANCE"
assert resp["Targets"][0]["WindowTargetId"] == _id
assert resp["Targets"][0]["Targets"][0]["Key"] == "tag:Name"
assert resp["Targets"][0]["Targets"][0]["Values"] == ["my-instance"]
@mock_ssm
@ -273,7 +272,7 @@ def test_deregister_target_from_maintenance_window():
resp = ssm.describe_maintenance_window_targets(
WindowId=window_id,
)
resp.should.have.key("Targets").should.have.length_of(0)
assert len(resp["Targets"]) == 0
@mock_ssm
@ -292,12 +291,12 @@ def test_describe_maintenance_window_with_no_task_or_targets():
resp = ssm.describe_maintenance_window_tasks(
WindowId=window_id,
)
resp.should.have.key("Tasks").should.have.length_of(0)
assert len(resp["Tasks"]) == 0
resp = ssm.describe_maintenance_window_targets(
WindowId=window_id,
)
resp.should.have.key("Targets").should.have.length_of(0)
assert len(resp["Targets"]) == 0
@mock_ssm
@ -329,18 +328,18 @@ def test_register_maintenance_window_task():
MaxErrors="1",
)
resp.should.have.key("WindowTaskId")
assert "WindowTaskId" in resp
_id = resp["WindowTaskId"]
resp = ssm.describe_maintenance_window_tasks(
WindowId=window_id,
)
resp.should.have.key("Tasks").should.have.length_of(1)
resp["Tasks"][0].should.have.key("WindowTaskId").equal(_id)
resp["Tasks"][0].should.have.key("WindowId").equal(window_id)
resp["Tasks"][0].should.have.key("TaskArn").equal("AWS-RunShellScript")
resp["Tasks"][0].should.have.key("MaxConcurrency").equal("1")
resp["Tasks"][0].should.have.key("MaxErrors").equal("1")
assert len(resp["Tasks"]) == 1
assert resp["Tasks"][0]["WindowTaskId"] == _id
assert resp["Tasks"][0]["WindowId"] == window_id
assert resp["Tasks"][0]["TaskArn"] == "AWS-RunShellScript"
assert resp["Tasks"][0]["MaxConcurrency"] == "1"
assert resp["Tasks"][0]["MaxErrors"] == "1"
@mock_ssm
@ -381,4 +380,4 @@ def test_deregister_maintenance_window_task():
resp = ssm.describe_maintenance_window_tasks(
WindowId=window_id,
)
resp.should.have.key("Tasks").should.have.length_of(0)
assert len(resp["Tasks"]) == 0

53
tests/test_ssm/test_ssm_parameterstore.py
View File

@ -1,5 +1,3 @@
import sure # noqa # pylint: disable=unused-import
from moto.ssm.models import ParameterDict
@ -7,42 +5,40 @@ def test_simple_setget():
store = ParameterDict("accnt", "region")
store["/a/b/c"] = "some object"
store.get("/a/b/c").should.equal("some object")
assert store.get("/a/b/c") == "some object"
def test_get_none():
store = ParameterDict("accnt", "region")
store.get(None).should.equal(None)
assert store.get(None) is None
def test_get_aws_param():
store = ParameterDict("accnt", "region")
p = store["/aws/service/global-infrastructure/regions/us-west-1/longName"]
p.should.have.length_of(1)
p[0].value.should.equal("US West (N. California)")
assert len(p) == 1
assert p[0].value == "US West (N. California)"
def test_iter():
store = ParameterDict("accnt", "region")
store["/a/b/c"] = "some object"
"/a/b/c".should.be.within(store)
"/a/b/d".shouldnt.be.within(store)
assert "/a/b/c" in store
assert "/a/b/d" not in store
def test_iter_none():
store = ParameterDict("accnt", "region")
None.shouldnt.be.within(store)
assert None not in store
def test_iter_aws():
store = ParameterDict("accnt", "region")
"/aws/service/global-infrastructure/regions/us-west-1/longName".should.be.within(
store
)
assert "/aws/service/global-infrastructure/regions/us-west-1/longName" in store
def test_get_key_beginning_with():
@ -52,20 +48,17 @@ def test_get_key_beginning_with():
store["/a/c/d"] = "some third object"
begins_with_ab = list(store.get_keys_beginning_with("/a/b", recursive=False))
begins_with_ab.should.equal(["/a/b/c"])
assert begins_with_ab == ["/a/b/c"]
begins_with_a = list(store.get_keys_beginning_with("/a", recursive=False))
begins_with_a.should.equal([])
assert not begins_with_a
begins_with_a_recursive = list(store.get_keys_beginning_with("/a", recursive=True))
set(begins_with_a_recursive).should.equal({"/a/b/c", "/a/c/d"})
assert set(begins_with_a_recursive) == {"/a/b/c", "/a/c/d"}
def test_get_key_beginning_with_aws():
"""
ParameterDict should load the default parameters if we request a key starting with '/aws'
:return:
"""
"""Test ParameterDict loads default params for key starting with '/aws'."""
store = ParameterDict("accnt", "region")
uswest_params = set(
@ -73,22 +66,20 @@ def test_get_key_beginning_with_aws():
"/aws/service/global-infrastructure/regions/us-west-1", recursive=False
)
)
uswest_params.should.equal(
{
"/aws/service/global-infrastructure/regions/us-west-1",
"/aws/service/global-infrastructure/regions/us-west-1/domain",
"/aws/service/global-infrastructure/regions/us-west-1/geolocationCountry",
"/aws/service/global-infrastructure/regions/us-west-1/geolocationRegion",
"/aws/service/global-infrastructure/regions/us-west-1/longName",
"/aws/service/global-infrastructure/regions/us-west-1/partition",
}
)
assert uswest_params == {
"/aws/service/global-infrastructure/regions/us-west-1",
"/aws/service/global-infrastructure/regions/us-west-1/domain",
"/aws/service/global-infrastructure/regions/us-west-1/geolocationCountry",
"/aws/service/global-infrastructure/regions/us-west-1/geolocationRegion",
"/aws/service/global-infrastructure/regions/us-west-1/longName",
"/aws/service/global-infrastructure/regions/us-west-1/partition",
}
def test_ssm_parameter_from_unknown_region():
store = ParameterDict("accnt", "region")
list(
assert not list(
store.get_keys_beginning_with(
"/aws/service/ami-amazon-linux-latest", recursive=False
)
).should.equal([])
)

15
tests/test_ssm/test_ssm_patch_baseline.py
View File

@ -1,5 +1,4 @@
import boto3
import sure # noqa # pylint: disable=unused-import
from moto import mock_ssm
@ -48,13 +47,13 @@ def test_create_patch_baseLine():
],
MaxResults=50,
)
response.should.have.key("BaselineIdentities").have.length_of(1)
assert len(response["BaselineIdentities"]) == 1
baseline = response["BaselineIdentities"][0]
baseline.should.have.key("BaselineId").equal(_id)
baseline.should.have.key("BaselineName").equal(baseline_name)
baseline.should.have.key("DefaultBaseline").equal(False)
baseline.should.have.key("OperatingSystem").equal("AMAZON_LINUX")
baseline.should.have.key("BaselineDescription").equal(baseline_description)
assert baseline["BaselineId"] == _id
assert baseline["BaselineName"] == baseline_name
assert baseline["DefaultBaseline"] is False
assert baseline["OperatingSystem"] == "AMAZON_LINUX"
assert baseline["BaselineDescription"] == baseline_description
@mock_ssm
@ -97,4 +96,4 @@ def test_delete_patch_baseline():
],
MaxResults=50,
)
response.should.have.key("BaselineIdentities").have.length_of(0)
assert len(response["BaselineIdentities"]) == 0

44
tests/test_ssm/test_ssm_secretsmanager.py
View File

@ -1,8 +1,9 @@
import boto3
import json
import boto3
from botocore.exceptions import ClientError
import pytest
from botocore.exceptions import ClientError
from moto import mock_ssm, mock_secretsmanager
@ -22,18 +23,18 @@ def test_get_value_from_secrets_manager__by_name():
param = ssm.get_parameter(
Name=f"/aws/reference/secretsmanager/{secret_name}", WithDecryption=True
)["Parameter"]
param.should.have.key("Name").equals("mysecret")
param.should.have.key("Type").equals("SecureString")
param.should.have.key("Value").equals("some secret")
param.should.have.key("Version").equals(0)
param.should.have.key("SourceResult")
assert param["Name"] == "mysecret"
assert param["Type"] == "SecureString"
assert param["Value"] == "some secret"
assert param["Version"] == 0
assert "SourceResult" in param
secret = secrets_manager.describe_secret(SecretId=secret_name)
source_result = json.loads(param["SourceResult"])
source_result["ARN"].should.equal(secret["ARN"])
source_result["Name"].should.equal(secret["Name"])
source_result["VersionIdsToStages"].should.equal(secret["VersionIdsToStages"])
assert source_result["ARN"] == secret["ARN"]
assert source_result["Name"] == secret["Name"]
assert source_result["VersionIdsToStages"] == secret["VersionIdsToStages"]
@mock_secretsmanager
@ -44,8 +45,8 @@ def test_get_value_from_secrets_manager__without_decryption():
with pytest.raises(ClientError) as exc:
ssm.get_parameter(Name="/aws/reference/secretsmanager/sth")
err = exc.value.response["Error"]
err["Code"].should.equal("ValidationException")
err["Message"].should.equal(
assert err["Code"] == "ValidationException"
assert err["Message"] == (
"WithDecryption flag must be True for retrieving a Secret Manager secret."
)
@ -60,8 +61,8 @@ def test_get_value_from_secrets_manager__with_decryption_false():
Name="/aws/reference/secretsmanager/sth", WithDecryption=False
)
err = exc.value.response["Error"]
err["Code"].should.equal("ValidationException")
err["Message"].should.equal(
assert err["Code"] == "ValidationException"
assert err["Message"] == (
"WithDecryption flag must be True for retrieving a Secret Manager secret."
)
@ -86,15 +87,15 @@ def test_get_value_from_secrets_manager__by_id():
# then
full_name = f"/aws/reference/secretsmanager/{name}:{version_id1}"
param = ssm.get_parameter(Name=full_name, WithDecryption=True)["Parameter"]
param.should.have.key("Value").equals("1st")
assert param["Value"] == "1st"
full_name = f"/aws/reference/secretsmanager/{name}"
param = ssm.get_parameter(Name=full_name, WithDecryption=True)["Parameter"]
param.should.have.key("Value").equals("2nd")
assert param["Value"] == "2nd"
full_name = f"/aws/reference/secretsmanager/{name}:{version_id3}"
param = ssm.get_parameter(Name=full_name, WithDecryption=True)["Parameter"]
param.should.have.key("Value").equals("3rd")
assert param["Value"] == "3rd"
@mock_secretsmanager
@ -112,7 +113,7 @@ def test_get_value_from_secrets_manager__by_version():
# then
full_name = f"/aws/reference/secretsmanager/{name}:AWSPREVIOUS"
param = ssm.get_parameter(Name=full_name, WithDecryption=True)["Parameter"]
param.should.have.key("Value").equals("1st")
assert param["Value"] == "1st"
@mock_secretsmanager
@ -124,7 +125,8 @@ def test_get_value_from_secrets_manager__param_does_not_exist():
Name="/aws/reference/secretsmanager/test", WithDecryption=True
)
err = exc.value.response["Error"]
err["Code"].should.equal("ParameterNotFound")
err["Message"].should.equal(
"An error occurred (ParameterNotFound) when referencing Secrets Manager: Secret /aws/reference/secretsmanager/test not found."
assert err["Code"] == "ParameterNotFound"
assert err["Message"] == (
"An error occurred (ParameterNotFound) when referencing Secrets "
"Manager: Secret /aws/reference/secretsmanager/test not found."
)

46
tests/test_ssm/test_ssm_utils.py
View File

@ -1,11 +1,11 @@
import sure # noqa # pylint: disable=unused-import
from moto.ssm.utils import convert_to_tree, convert_to_params
SOURCE_PARAMS = [
{
"ARN": "arn:aws:ssm:us-west-1::parameter/aws/service/global-infrastructure/regions/af-south-1",
"ARN": (
"arn:aws:ssm:us-west-1::parameter/aws/service"
"/global-infrastructure/regions/af-south-1"
),
"DataType": "text",
"Name": "/aws/service/global-infrastructure/regions/af-south-1",
"Type": "String",
@ -13,7 +13,10 @@ SOURCE_PARAMS = [
"Version": 1,
},
{
"ARN": "arn:aws:ssm:us-west-1::parameter/aws/service/global-infrastructure/regions/ap-northeast-2",
"ARN": (
"arn:aws:ssm:us-west-1::parameter/aws/service"
"/global-infrastructure/regions/ap-northeast-2"
),
"DataType": "text",
"Name": "/aws/service/global-infrastructure/regions/ap-northeast-2",
"Type": "String",
@ -21,7 +24,10 @@ SOURCE_PARAMS = [
"Version": 1,
},
{
"ARN": "arn:aws:ssm:us-west-1::parameter/aws/service/global-infrastructure/regions/cn-north-1",
"ARN": (
"arn:aws:ssm:us-west-1::parameter/aws/service"
"/global-infrastructure/regions/cn-north-1"
),
"DataType": "text",
"Name": "/aws/service/global-infrastructure/regions/cn-north-1",
"Type": "String",
@ -29,9 +35,16 @@ SOURCE_PARAMS = [
"Version": 1,
},
{
"ARN": "arn:aws:ssm:us-west-1::parameter/aws/service/global-infrastructure/regions/ap-northeast-2/services/codestar-notifications",
"ARN": (
"arn:aws:ssm:us-west-1::parameter/aws/service"
"/global-infrastructure/regions/ap-northeast-2/services"
"/codestar-notifications"
),
"DataType": "text",
"Name": "/aws/service/global-infrastructure/regions/ap-northeast-2/services/codestar-notifications",
"Name": (
"/aws/service/global-infrastructure/regions"
"/ap-northeast-2/services/codestar-notifications"
),
"Type": "String",
"Value": "codestar-notifications",
"Version": 1,
@ -73,7 +86,10 @@ CONVERTED_PARAMS = [
"Value": "ap-northeast-2",
},
{
"Name": "/aws/service/global-infrastructure/regions/ap-northeast-2/services/codestar-notifications",
"Name": (
"/aws/service/global-infrastructure/regions/ap-northeast-2"
"/services/codestar-notifications"
),
"Value": "codestar-notifications",
},
]
@ -82,20 +98,18 @@ CONVERTED_PARAMS = [
def test_convert_to_tree():
tree = convert_to_tree(SOURCE_PARAMS)
tree.should.equal(EXPECTED_TREE)
assert tree == EXPECTED_TREE
def test_convert_to_params():
actual = convert_to_params(EXPECTED_TREE)
actual.should.have.length_of(len(CONVERTED_PARAMS))
assert len(actual) == len(CONVERTED_PARAMS)
for param in CONVERTED_PARAMS:
actual.should.contain(param)
assert param in actual
def test_input_is_correct():
"""
Test input should match
"""
"""Test input should match."""
for src in SOURCE_PARAMS:
minimized_src = {"Name": src["Name"], "Value": src["Value"]}
CONVERTED_PARAMS.should.contain(minimized_src)
assert minimized_src in CONVERTED_PARAMS