CloudFormation: Add actions for ssm parameter (#3256)

This commit is contained in:
Ninh Khong 2022-11-05 18:06:12 +07:00 committed by GitHub
parent 2500affb1d
commit 37a127acd5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 226 additions and 2 deletions

View File

@ -4,7 +4,7 @@ from typing import Dict
from collections import defaultdict
from moto.core import BaseBackend, BaseModel
from moto.core import BaseBackend, BaseModel, CloudFormationModel
from moto.core.exceptions import RESTError
from moto.core.utils import BackendDict
from moto.ec2 import ec2_backends
@ -174,7 +174,7 @@ PARAMETER_VERSION_LIMIT = 100
PARAMETER_HISTORY_MAX_RESULTS = 50
class Parameter(BaseModel):
class Parameter(CloudFormationModel):
def __init__(
self,
account_id,
@ -259,6 +259,62 @@ class Parameter(BaseModel):
return r
@staticmethod
def cloudformation_name_type():
return "Name"
@staticmethod
def cloudformation_type():
# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-ssm-parameter.html
return "AWS::SSM::Parameter"
@classmethod
def create_from_cloudformation_json(
cls, resource_name, cloudformation_json, account_id, region_name, **kwargs
):
ssm_backend = ssm_backends[account_id][region_name]
properties = cloudformation_json["Properties"]
parameter_args = {
"name": properties.get("Name"),
"description": properties.get("Description", None),
"value": properties.get("Value", None),
"parameter_type": properties.get("Type", None),
"allowed_pattern": properties.get("AllowedPattern", None),
"keyid": properties.get("KeyId", None),
"overwrite": properties.get("Overwrite", False),
"tags": properties.get("Tags", None),
"data_type": properties.get("DataType", "text"),
}
ssm_backend.put_parameter(**parameter_args)
parameter = ssm_backend.get_parameter(properties.get("Name"))
return parameter
@classmethod
def update_from_cloudformation_json(
cls,
original_resource,
new_resource_name,
cloudformation_json,
account_id,
region_name,
):
cls.delete_from_cloudformation_json(
original_resource.name, cloudformation_json, account_id, region_name
)
return cls.create_from_cloudformation_json(
new_resource_name, cloudformation_json, account_id, region_name
)
@classmethod
def delete_from_cloudformation_json(
cls, resource_name, cloudformation_json, account_id, region_name
):
ssm_backend = ssm_backends[account_id][region_name]
properties = cloudformation_json["Properties"]
ssm_backend.delete_parameter(properties.get("Name"))
MAX_TIMEOUT_SECONDS = 3600

View File

@ -22,6 +22,7 @@ from moto import (
mock_s3,
mock_sqs,
mock_elbv2,
mock_ssm,
)
from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
@ -1676,3 +1677,170 @@ def test_dynamodb_table_creation():
ddb = boto3.client("dynamodb", "us-west-2")
table_names = ddb.list_tables()["TableNames"]
table_names.should.equal([outputs[0]["OutputValue"]])
@mock_cloudformation
@mock_ssm
def test_ssm_parameter():
parameter_template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
"BasicParameter": {
"Type": "AWS::SSM::Parameter",
"Properties": {
"Name": "test_ssm",
"Type": "String",
"Value": "Test SSM Parameter",
"Description": "Test SSM Description",
"AllowedPattern": "^[a-zA-Z]{1,10}$",
},
}
},
}
stack_name = "test_stack"
cfn = boto3.client("cloudformation", "us-west-2")
cfn.create_stack(StackName=stack_name, TemplateBody=json.dumps(parameter_template))
# Wait until moto creates the stack
waiter = cfn.get_waiter("stack_create_complete")
waiter.wait(StackName=stack_name)
ssm_client = boto3.client("ssm", region_name="us-west-2")
parameters = ssm_client.get_parameters(Names=["test_ssm"], WithDecryption=False)[
"Parameters"
]
parameters.should.have.length_of(1)
parameters[0]["Name"].should.equal("test_ssm")
parameters[0]["Value"].should.equal("Test SSM Parameter")
@mock_cloudformation
@mock_ssm
def test_ssm_parameter_update_stack():
parameter_template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
"BasicParameter": {
"Type": "AWS::SSM::Parameter",
"Properties": {
"Name": "test_ssm",
"Type": "String",
"Value": "Test SSM Parameter",
"Description": "Test SSM Description",
"AllowedPattern": "^[a-zA-Z]{1,10}$",
},
}
},
}
stack_name = "test_stack"
cfn = boto3.client("cloudformation", "us-west-2")
cfn.create_stack(StackName=stack_name, TemplateBody=json.dumps(parameter_template))
# Wait until moto creates the stack
waiter = cfn.get_waiter("stack_create_complete")
waiter.wait(StackName=stack_name)
ssm_client = boto3.client("ssm", region_name="us-west-2")
parameters = ssm_client.get_parameters(Names=["test_ssm"], WithDecryption=False)[
"Parameters"
]
parameters.should.have.length_of(1)
parameters[0]["Name"].should.equal("test_ssm")
parameters[0]["Value"].should.equal("Test SSM Parameter")
parameter_template["Resources"]["BasicParameter"]["Properties"][
"Value"
] = "Test SSM Parameter Updated"
cfn.update_stack(StackName=stack_name, TemplateBody=json.dumps(parameter_template))
ssm_client = boto3.client("ssm", region_name="us-west-2")
parameters = ssm_client.get_parameters(Names=["test_ssm"], WithDecryption=False)[
"Parameters"
]
parameters.should.have.length_of(1)
parameters[0]["Name"].should.equal("test_ssm")
parameters[0]["Value"].should.equal("Test SSM Parameter Updated")
@mock_cloudformation
@mock_ssm
def test_ssm_parameter_update_stack_and_remove_resource():
parameter_template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
"BasicParameter": {
"Type": "AWS::SSM::Parameter",
"Properties": {
"Name": "test_ssm",
"Type": "String",
"Value": "Test SSM Parameter",
"Description": "Test SSM Description",
"AllowedPattern": "^[a-zA-Z]{1,10}$",
},
}
},
}
stack_name = "test_stack"
cfn = boto3.client("cloudformation", "us-west-2")
cfn.create_stack(StackName=stack_name, TemplateBody=json.dumps(parameter_template))
# Wait until moto creates the stack
waiter = cfn.get_waiter("stack_create_complete")
waiter.wait(StackName=stack_name)
ssm_client = boto3.client("ssm", region_name="us-west-2")
parameters = ssm_client.get_parameters(Names=["test_ssm"], WithDecryption=False)[
"Parameters"
]
parameters.should.have.length_of(1)
parameters[0]["Name"].should.equal("test_ssm")
parameters[0]["Value"].should.equal("Test SSM Parameter")
parameter_template["Resources"].pop("BasicParameter")
cfn.update_stack(StackName=stack_name, TemplateBody=json.dumps(parameter_template))
ssm_client = boto3.client("ssm", region_name="us-west-1")
parameters = ssm_client.get_parameters(Names=["test_ssm"], WithDecryption=False)[
"Parameters"
]
parameters.should.have.length_of(0)
@mock_cloudformation
@mock_ssm
def test_ssm_parameter_update_stack_and_add_resource():
parameter_template = {"AWSTemplateFormatVersion": "2010-09-09", "Resources": {}}
stack_name = "test_stack"
cfn = boto3.client("cloudformation", "us-west-2")
cfn.create_stack(StackName=stack_name, TemplateBody=json.dumps(parameter_template))
# Wait until moto creates the stack
waiter = cfn.get_waiter("stack_create_complete")
waiter.wait(StackName=stack_name)
ssm_client = boto3.client("ssm", region_name="us-west-2")
parameters = ssm_client.get_parameters(Names=["test_ssm"], WithDecryption=False)[
"Parameters"
]
parameters.should.have.length_of(0)
parameter_template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
"BasicParameter": {
"Type": "AWS::SSM::Parameter",
"Properties": {
"Name": "test_ssm",
"Type": "String",
"Value": "Test SSM Parameter",
"Description": "Test SSM Description",
"AllowedPattern": "^[a-zA-Z]{1,10}$",
},
}
},
}
cfn.update_stack(StackName=stack_name, TemplateBody=json.dumps(parameter_template))
ssm_client = boto3.client("ssm", region_name="us-west-2")
parameters = ssm_client.get_parameters(Names=["test_ssm"], WithDecryption=False)[
"Parameters"
]
parameters.should.have.length_of(1)
parameters[0]["Name"].should.equal("test_ssm")
parameters[0]["Value"].should.equal("Test SSM Parameter")