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

Cloudformation - EventBus support (#3052)

Browse source 
* add EventBus to model's map

* add support for creation of EventBus through cloudformation's api

* add cloudformation's delete

* add cloudformation's update

* add cloudformation's attribute
This commit is contained in:
Guilherme Martins Crocetti 2020-06-06 07:31:14 -03:00 • committed by GitHub
commit e32a60185f
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 223 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

175
tests/test_cloudformation/test_cloudformation_stack_integration.py
View file

@ -2585,3 +2585,178 @@ def test_autoscaling_propagate_tags():
assert propagation_dict["test-key-propagate"]
assert not propagation_dict["test-key-no-propagate"]
@mock_cloudformation
@mock_events
def test_stack_eventbus_create_from_cfn_integration():
eventbus_template = """{
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
"EventBus": {
"Type": "AWS::Events::EventBus",
"Properties": {
"Name": "MyCustomEventBus"
},
}
},
}"""
cf_conn = boto3.client("cloudformation", "us-west-2")
cf_conn.create_stack(StackName="test_stack", TemplateBody=eventbus_template)
event_buses = boto3.client("events", "us-west-2").list_event_buses(
NamePrefix="MyCustom"
)
event_buses["EventBuses"].should.have.length_of(1)
event_buses["EventBuses"][0]["Name"].should.equal("MyCustomEventBus")
@mock_cloudformation
@mock_events
def test_stack_events_delete_eventbus_integration():
eventbus_template = """{
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
"EventBus": {
"Type": "AWS::Events::EventBus",
"Properties": {
"Name": "MyCustomEventBus"
},
}
},
}"""
cf_conn = boto3.client("cloudformation", "us-west-2")
cf_conn.create_stack(StackName="test_stack", TemplateBody=eventbus_template)
event_buses = boto3.client("events", "us-west-2").list_event_buses(
NamePrefix="MyCustom"
)
event_buses["EventBuses"].should.have.length_of(1)
cf_conn.delete_stack(StackName="test_stack")
event_buses = boto3.client("events", "us-west-2").list_event_buses(
NamePrefix="MyCustom"
)
event_buses["EventBuses"].should.have.length_of(0)
@mock_cloudformation
@mock_events
def test_stack_events_delete_from_cfn_integration():
eventbus_template = Template(
"""{
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
"$resource_name": {
"Type": "AWS::Events::EventBus",
"Properties": {
"Name": "$name"
},
}
},
}"""
)
cf_conn = boto3.client("cloudformation", "us-west-2")
original_template = eventbus_template.substitute(
{"resource_name": "original", "name": "MyCustomEventBus"}
)
cf_conn.create_stack(StackName="test_stack", TemplateBody=original_template)
original_event_buses = boto3.client("events", "us-west-2").list_event_buses(
NamePrefix="MyCustom"
)
original_event_buses["EventBuses"].should.have.length_of(1)
original_eventbus = original_event_buses["EventBuses"][0]
updated_template = eventbus_template.substitute(
{"resource_name": "updated", "name": "AnotherEventBus"}
)
cf_conn.update_stack(StackName="test_stack", TemplateBody=updated_template)
update_event_buses = boto3.client("events", "us-west-2").list_event_buses(
NamePrefix="AnotherEventBus"
)
update_event_buses["EventBuses"].should.have.length_of(1)
update_event_buses["EventBuses"][0]["Arn"].shouldnt.equal(original_eventbus["Arn"])
@mock_cloudformation
@mock_events
def test_stack_events_update_from_cfn_integration():
eventbus_template = Template(
"""{
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
"EventBus": {
"Type": "AWS::Events::EventBus",
"Properties": {
"Name": "$name"
},
}
},
}"""
)
cf_conn = boto3.client("cloudformation", "us-west-2")
original_template = eventbus_template.substitute({"name": "MyCustomEventBus"})
cf_conn.create_stack(StackName="test_stack", TemplateBody=original_template)
original_event_buses = boto3.client("events", "us-west-2").list_event_buses(
NamePrefix="MyCustom"
)
original_event_buses["EventBuses"].should.have.length_of(1)
original_eventbus = original_event_buses["EventBuses"][0]
updated_template = eventbus_template.substitute({"name": "NewEventBus"})
cf_conn.update_stack(StackName="test_stack", TemplateBody=updated_template)
update_event_buses = boto3.client("events", "us-west-2").list_event_buses(
NamePrefix="NewEventBus"
)
update_event_buses["EventBuses"].should.have.length_of(1)
update_event_buses["EventBuses"][0]["Name"].should.equal("NewEventBus")
update_event_buses["EventBuses"][0]["Arn"].shouldnt.equal(original_eventbus["Arn"])
@mock_cloudformation
@mock_events
def test_stack_events_get_attribute_integration():
eventbus_template = """{
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
"EventBus": {
"Type": "AWS::Events::EventBus",
"Properties": {
"Name": "MyEventBus"
},
}
},
"Outputs": {
"bus_arn": {"Value": {"Fn::GetAtt": ["EventBus", "Arn"]}},
"bus_name": {"Value": {"Fn::GetAtt": ["EventBus", "Name"]}},
}
}"""
cf = boto3.client("cloudformation", "us-west-2")
events = boto3.client("events", "us-west-2")
cf.create_stack(StackName="test_stack", TemplateBody=eventbus_template)
stack = cf.describe_stacks(StackName="test_stack")["Stacks"][0]
outputs = stack["Outputs"]
output_arn = list(filter(lambda item: item["OutputKey"] == "bus_arn", outputs))[0]
output_name = list(filter(lambda item: item["OutputKey"] == "bus_name", outputs))[0]
event_bus = events.list_event_buses(NamePrefix="MyEventBus")["EventBuses"][0]
output_arn["OutputValue"].should.equal(event_bus["Arn"])
output_name["OutputValue"].should.equal(event_bus["Name"])