199da2220b
* Add events.create_archive * Add events.describe_archive * Add events.list_archives * Add events.update_archive * Add events.delete_archive * Add actual archive functionality * Fix test * Fix PR issues
108 lines
3.3 KiB
Python
108 lines
3.3 KiB
Python
import copy
|
|
from string import Template
|
|
|
|
import boto3
|
|
import json
|
|
from moto import mock_cloudformation, mock_events
|
|
import sure # noqa
|
|
|
|
from moto.core import ACCOUNT_ID
|
|
|
|
archive_template = Template(
|
|
json.dumps(
|
|
{
|
|
"AWSTemplateFormatVersion": "2010-09-09",
|
|
"Description": "EventBridge Archive Test",
|
|
"Resources": {
|
|
"Archive": {
|
|
"Type": "AWS::Events::Archive",
|
|
"Properties": {
|
|
"ArchiveName": "${archive_name}",
|
|
"SourceArn": {
|
|
"Fn::Sub": "arn:aws:events:$${AWS::Region}:$${AWS::AccountId}:event-bus/default"
|
|
},
|
|
},
|
|
}
|
|
},
|
|
"Outputs": {
|
|
"Arn": {
|
|
"Description": "Archive Arn",
|
|
"Value": {"Fn::GetAtt": ["Archive", "Arn"]},
|
|
}
|
|
},
|
|
}
|
|
)
|
|
)
|
|
|
|
|
|
@mock_events
|
|
@mock_cloudformation
|
|
def test_create_archive():
|
|
# given
|
|
cfn_client = boto3.client("cloudformation", region_name="eu-central-1")
|
|
name = "test-archive"
|
|
stack_name = "test-stack"
|
|
template = archive_template.substitute({"archive_name": name})
|
|
|
|
# when
|
|
cfn_client.create_stack(StackName=stack_name, TemplateBody=template)
|
|
|
|
# then
|
|
archive_arn = "arn:aws:events:eu-central-1:{0}:archive/{1}".format(ACCOUNT_ID, name)
|
|
stack = cfn_client.describe_stacks(StackName=stack_name)["Stacks"][0]
|
|
stack["Outputs"][0]["OutputValue"].should.equal(archive_arn)
|
|
|
|
events_client = boto3.client("events", region_name="eu-central-1")
|
|
response = events_client.describe_archive(ArchiveName=name)
|
|
|
|
response["ArchiveArn"].should.equal(archive_arn)
|
|
|
|
|
|
@mock_events
|
|
@mock_cloudformation
|
|
def test_update_archive():
|
|
# given
|
|
cfn_client = boto3.client("cloudformation", region_name="eu-central-1")
|
|
name = "test-archive"
|
|
stack_name = "test-stack"
|
|
template = archive_template.substitute({"archive_name": name})
|
|
cfn_client.create_stack(StackName=stack_name, TemplateBody=template)
|
|
|
|
template_update = copy.deepcopy(json.loads(template))
|
|
template_update["Resources"]["Archive"]["Properties"][
|
|
"Description"
|
|
] = "test archive"
|
|
|
|
# when
|
|
cfn_client.update_stack(
|
|
StackName=stack_name, TemplateBody=json.dumps(template_update)
|
|
)
|
|
|
|
# then
|
|
events_client = boto3.client("events", region_name="eu-central-1")
|
|
response = events_client.describe_archive(ArchiveName=name)
|
|
|
|
response["ArchiveArn"].should.equal(
|
|
"arn:aws:events:eu-central-1:{0}:archive/{1}".format(ACCOUNT_ID, name)
|
|
)
|
|
response["Description"].should.equal("test archive")
|
|
|
|
|
|
@mock_events
|
|
@mock_cloudformation
|
|
def test_delete_archive():
|
|
# given
|
|
cfn_client = boto3.client("cloudformation", region_name="eu-central-1")
|
|
name = "test-archive"
|
|
stack_name = "test-stack"
|
|
template = archive_template.substitute({"archive_name": name})
|
|
cfn_client.create_stack(StackName=stack_name, TemplateBody=template)
|
|
|
|
# when
|
|
cfn_client.delete_stack(StackName=stack_name)
|
|
|
|
# then
|
|
events_client = boto3.client("events", region_name="eu-central-1")
|
|
response = events_client.list_archives(NamePrefix="test")["Archives"]
|
|
response.should.have.length_of(0)
|