catalin/moto
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Add EventBridge Archive CRUD endpoints (#3617)

Browse source 
* 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
This commit is contained in:
Anton Grübel 2021-01-28 11:47:53 +01:00 • committed by GitHub
commit 199da2220b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 985 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

107
tests/test_events/test_events_cloudformation.py Normal file
View file

@ -0,0 +1,107 @@
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)