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

Merge remote-tracking branch 'upstream/master' into adding_rds2_support

Browse Source
This commit is contained in:
Mike Fuller 2015-01-19 08:22:27 +11:00
commit e2ed5ed733
11 changed files with 626 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
moto/cloudformation/parsing.py
View File

@ -8,6 +8,8 @@ from moto.ec2 import models as ec2_models
from moto.elb import models as elb_models
from moto.iam import models as iam_models
from moto.rds import models as rds_models
from moto.route53 import models as route53_models
from moto.sns import models as sns_models
from moto.sqs import models as sqs_models
from .utils import random_suffix
from .exceptions import MissingParameterError, UnformattedGetAttTemplateException
@ -36,6 +38,11 @@ MODEL_MAP = {
"AWS::RDS::DBInstance": rds_models.Database,
"AWS::RDS::DBSecurityGroup": rds_models.SecurityGroup,
"AWS::RDS::DBSubnetGroup": rds_models.SubnetGroup,
"AWS::Route53::HealthCheck": route53_models.HealthCheck,
"AWS::Route53::HostedZone": route53_models.FakeZone,
"AWS::Route53::RecordSet": route53_models.RecordSet,
"AWS::Route53::RecordSetGroup": route53_models.RecordSetGroup,
"AWS::SNS::Topic": sns_models.Topic,
"AWS::SQS::Queue": sqs_models.Queue,
}
@ -144,8 +151,6 @@ def resource_class_from_type(resource_type):
def resource_name_property_from_type(resource_type):
if resource_type not in NAME_TYPE_MAP:
return None
return NAME_TYPE_MAP.get(resource_type)

2
moto/ec2/models.py
View File

@ -299,6 +299,7 @@ class Instance(BotoInstance, TaggedEC2Resource):
self.subnet_id = kwargs.get("subnet_id")
self.key_name = kwargs.get("key_name")
self.source_dest_check = "true"
self.private_ip_address = kwargs.get('private_ip_address')
self.block_device_mapping = BlockDeviceMapping()
self.block_device_mapping['/dev/sda1'] = BlockDeviceType(volume_id=random_volume_id())
@ -344,6 +345,7 @@ class Instance(BotoInstance, TaggedEC2Resource):
instance_type=properties.get("InstanceType", "m1.small"),
subnet_id=properties.get("SubnetId"),
key_name=properties.get("KeyName"),
private_ip_address=properties.get('PrivateIpAddress'),
)
return reservation.instances[0]

173
moto/route53/models.py
View File

@ -1,26 +1,174 @@
from __future__ import unicode_literals
import uuid
from jinja2 import Template
from moto.core import BaseBackend
from moto.core.utils import get_random_hex
class HealthCheck(object):
def __init__(self, health_check_id, health_check_args):
self.id = health_check_id
self.ip_address = health_check_args.get("ip_address")
self.port = health_check_args.get("port", 80)
self._type = health_check_args.get("type")
self.resource_path = health_check_args.get("resource_path")
self.fqdn = health_check_args.get("fqdn")
self.search_string = health_check_args.get("search_string")
self.request_interval = health_check_args.get("request_interval", 30)
self.failure_threshold = health_check_args.get("failure_threshold", 3)
@property
def physical_resource_id(self):
return self.id
@classmethod
def create_from_cloudformation_json(cls, resource_name, cloudformation_json, region_name):
properties = cloudformation_json['Properties']['HealthCheckConfig']
health_check_args = {
"ip_address": properties.get('IPAddress'),
"port": properties.get('Port'),
"type": properties['Type'],
"resource_path": properties.get('ResourcePath'),
"fqdn": properties.get('FullyQualifiedDomainName'),
"search_string": properties.get('SearchString'),
"request_interval": properties.get('RequestInterval'),
"failure_threshold": properties.get('FailureThreshold'),
}
health_check = route53_backend.create_health_check(health_check_args)
return health_check
def to_xml(self):
template = Template("""<HealthCheck>
<Id>{{ health_check.id }}</Id>
<CallerReference>example.com 192.0.2.17</CallerReference>
<HealthCheckConfig>
<IPAddress>{{ health_check.ip_address }}</IPAddress>
<Port>{{ health_check.port }}</Port>
<Type>{{ health_check._type }}</Type>
<ResourcePath>{{ health_check.resource_path }}</ResourcePath>
<FullyQualifiedDomainName>{{ health_check.fqdn }}</FullyQualifiedDomainName>
<RequestInterval>{{ health_check.request_interval }}</RequestInterval>
<FailureThreshold>{{ health_check.failure_threshold }}</FailureThreshold>
{% if health_check.search_string %}
<SearchString>{{ health_check.search_string }}</SearchString>
{% endif %}
</HealthCheckConfig>
<HealthCheckVersion>1</HealthCheckVersion>
</HealthCheck>""")
return template.render(health_check=self)
class RecordSet(object):
def __init__(self, kwargs):
self.name = kwargs.get('Name')
self._type = kwargs.get('Type')
self.ttl = kwargs.get('TTL')
self.records = kwargs.get('ResourceRecords', [])
self.set_identifier = kwargs.get('SetIdentifier')
self.weight = kwargs.get('Weight')
self.health_check = kwargs.get('HealthCheckId')
@classmethod
def create_from_cloudformation_json(cls, resource_name, cloudformation_json, region_name):
properties = cloudformation_json['Properties']
zone_name = properties["HostedZoneName"]
hosted_zone = route53_backend.get_hosted_zone_by_name(zone_name)
record_set = hosted_zone.add_rrset(properties)
return record_set
def to_xml(self):
template = Template("""<ResourceRecordSet>
<Name>{{ record_set.name }}</Name>
<Type>{{ record_set._type }}</Type>
{% if record_set.set_identifier %}
<SetIdentifier>{{ record_set.set_identifier }}</SetIdentifier>
{% endif %}
{% if record_set.weight %}
<Weight>{{ record_set.weight }}</Weight>
{% endif %}
<TTL>{{ record_set.ttl }}</TTL>
<ResourceRecords>
{% for record in record_set.records %}
<ResourceRecord>
<Value>{{ record }}</Value>
</ResourceRecord>
{% endfor %}
</ResourceRecords>
{% if record_set.health_check %}
<HealthCheckId>{{ record_set.health_check }}</HealthCheckId>
{% endif %}
</ResourceRecordSet>""")
return template.render(record_set=self)
class FakeZone(object):
def __init__(self, name, id_):
self.name = name
self.id = id_
self.rrsets = {}
self.rrsets = []
def add_rrset(self, name, rrset):
self.rrsets[name] = rrset
def add_rrset(self, record_set):
record_set = RecordSet(record_set)
self.rrsets.append(record_set)
return record_set
def delete_rrset(self, name):
self.rrsets.pop(name, None)
self.rrsets = [record_set for record_set in self.rrsets if record_set.name != name]
def get_record_sets(self, type_filter, name_filter):
record_sets = list(self.rrsets) # Copy the list
if type_filter:
record_sets = [record_set for record_set in record_sets if record_set._type == type_filter]
if name_filter:
record_sets = [record_set for record_set in record_sets if record_set.name == name_filter]
return record_sets
@property
def physical_resource_id(self):
return self.name
@classmethod
def create_from_cloudformation_json(cls, resource_name, cloudformation_json, region_name):
properties = cloudformation_json['Properties']
name = properties["Name"]
hosted_zone = route53_backend.create_hosted_zone(name)
return hosted_zone
class RecordSetGroup(object):
def __init__(self, hosted_zone_id, record_sets):
self.hosted_zone_id = hosted_zone_id
self.record_sets = record_sets
@property
def physical_resource_id(self):
return "arn:aws:route53:::hostedzone/{0}".format(self.hosted_zone_id)
@classmethod
def create_from_cloudformation_json(cls, resource_name, cloudformation_json, region_name):
properties = cloudformation_json['Properties']
zone_name = properties["HostedZoneName"]
hosted_zone = route53_backend.get_hosted_zone_by_name(zone_name)
record_sets = properties["RecordSets"]
for record_set in record_sets:
hosted_zone.add_rrset(record_set)
record_set_group = RecordSetGroup(hosted_zone.id, record_sets)
return record_set_group
class Route53Backend(BaseBackend):
def __init__(self):
self.zones = {}
self.health_checks = {}
def create_hosted_zone(self, name):
new_id = get_random_hex()
@ -34,12 +182,27 @@ class Route53Backend(BaseBackend):
def get_hosted_zone(self, id_):
return self.zones.get(id_)
def get_hosted_zone_by_name(self, name):
for zone in self.get_all_hosted_zones():
if zone.name == name:
return zone
def delete_hosted_zone(self, id_):
zone = self.zones.get(id_)
if zone:
del self.zones[id_]
return zone
return None
def create_health_check(self, health_check_args):
health_check_id = str(uuid.uuid4())
health_check = HealthCheck(health_check_id, health_check_args)
self.health_checks[health_check_id] = health_check
return health_check
def get_health_checks(self):
return self.health_checks.values()
def delete_health_check(self, health_check_id):
return self.health_checks.pop(health_check_id, None)
route53_backend = Route53Backend()

73
moto/route53/responses.py
View File

@ -3,7 +3,6 @@ from jinja2 import Template
from six.moves.urllib.parse import parse_qs, urlparse
from .models import route53_backend
import xmltodict
import dicttoxml
def list_or_create_hostzone_response(request, full_url, headers):
@ -53,33 +52,57 @@ def rrset_response(request, full_url, headers):
for value in change_list:
action = value['Action']
rrset = value['ResourceRecordSet']
record_set = value['ResourceRecordSet']
if action == 'CREATE':
the_zone.add_rrset(rrset["Name"], rrset)
record_set['ResourceRecords'] = [x['Value'] for x in record_set['ResourceRecords'].values()]
the_zone.add_rrset(record_set)
elif action == "DELETE":
the_zone.delete_rrset(rrset["Name"])
the_zone.delete_rrset(record_set["Name"])
return 200, headers, CHANGE_RRSET_RESPONSE
elif method == "GET":
querystring = parse_qs(parsed_url.query)
template = Template(LIST_RRSET_REPONSE)
rrset_list = []
for key, value in the_zone.rrsets.items():
if 'type' in querystring and querystring["type"][0] != value["Type"]:
continue
if 'name' in querystring and querystring["name"][0] != value["Name"]:
continue
rrset_list.append(dicttoxml.dicttoxml({"ResourceRecordSet": value}, root=False))
type_filter = querystring.get("type", [None])[0]
name_filter = querystring.get("name", [None])[0]
record_sets = the_zone.get_record_sets(type_filter, name_filter)
return 200, headers, template.render(record_sets=record_sets)
return 200, headers, template.render(rrsets=rrset_list)
def health_check_response(request, full_url, headers):
parsed_url = urlparse(full_url)
method = request.method
if method == "POST":
properties = xmltodict.parse(request.body)['CreateHealthCheckRequest']['HealthCheckConfig']
health_check_args = {
"ip_address": properties.get('IPAddress'),
"port": properties.get('Port'),
"type": properties['Type'],
"resource_path": properties.get('ResourcePath'),
"fqdn": properties.get('FullyQualifiedDomainName'),
"search_string": properties.get('SearchString'),
"request_interval": properties.get('RequestInterval'),
"failure_threshold": properties.get('FailureThreshold'),
}
health_check = route53_backend.create_health_check(health_check_args)
template = Template(CREATE_HEALTH_CHECK_RESPONSE)
return 201, headers, template.render(health_check=health_check)
elif method == "DELETE":
health_check_id = parsed_url.path.split("/")[-1]
route53_backend.delete_health_check(health_check_id)
return 200, headers, DELETE_HEALTH_CHECK_REPONSE
elif method == "GET":
template = Template(LIST_HEALTH_CHECKS_REPONSE)
health_checks = route53_backend.get_health_checks()
return 200, headers, template.render(health_checks=health_checks)
LIST_RRSET_REPONSE = """<ListResourceRecordSetsResponse xmlns="https://route53.amazonaws.com/doc/2012-12-12/">
<ResourceRecordSets>
{% for rrset in rrsets %}
{{ rrset }}
{% for record_set in record_sets %}
{{ record_set.to_xml() }}
{% endfor %}
</ResourceRecordSets>
</ListResourceRecordSetsResponse>"""
@ -132,3 +155,23 @@ LIST_HOSTED_ZONES_RESPONSE = """<ListHostedZonesResponse xmlns="https://route53.
{% endfor %}
</HostedZones>
</ListHostedZonesResponse>"""
CREATE_HEALTH_CHECK_RESPONSE = """<?xml version="1.0" encoding="UTF-8"?>
<CreateHealthCheckResponse xmlns="https://route53.amazonaws.com/doc/2013-04-01/">
{{ health_check.to_xml() }}
</CreateHealthCheckResponse>"""
LIST_HEALTH_CHECKS_REPONSE = """<?xml version="1.0" encoding="UTF-8"?>
<ListHealthChecksResponse xmlns="https://route53.amazonaws.com/doc/2013-04-01/">
<HealthChecks>
{% for health_check in health_checks %}
{{ health_check.to_xml() }}
{% endfor %}
</HealthChecks>
<IsTruncated>false</IsTruncated>
<MaxItems>{{ health_checks|length }}</MaxItems>
</ListHealthChecksResponse>"""
DELETE_HEALTH_CHECK_REPONSE = """<?xml version="1.0" encoding="UTF-8"?>
<DeleteHealthCheckResponse xmlns="https://route53.amazonaws.com/doc/2013-04-01/">
</DeleteHealthCheckResponse>"""

9
moto/route53/urls.py
View File

@ -2,11 +2,12 @@ from __future__ import unicode_literals
from . import responses
url_bases = [
"https://route53.amazonaws.com/201.-..-../hostedzone",
"https://route53.amazonaws.com/201.-..-../",
]
url_paths = {
'{0}$': responses.list_or_create_hostzone_response,
'{0}/[^/]+$': responses.get_or_delete_hostzone_response,
'{0}/[^/]+/rrset$': responses.rrset_response,
'{0}hostedzone$': responses.list_or_create_hostzone_response,
'{0}hostedzone/[^/]+$': responses.get_or_delete_hostzone_response,
'{0}hostedzone/[^/]+/rrset$': responses.rrset_response,
'{0}healthcheck': responses.health_check_response,
}

16
moto/sns/models.py
View File

@ -45,6 +45,22 @@ class Topic(object):
return self.name
raise UnformattedGetAttTemplateException()
@property
def physical_resource_id(self):
return self.arn
@classmethod
def create_from_cloudformation_json(cls, resource_name, cloudformation_json, region_name):
sns_backend = sns_backends[region_name]
properties = cloudformation_json['Properties']
topic = sns_backend.create_topic(
properties.get("TopicName")
)
for subscription in properties.get("Subscription", []):
sns_backend.subscribe(topic.arn, subscription['Endpoint'], subscription['Protocol'])
return topic
class Subscription(object):
def __init__(self, topic, endpoint, protocol):

40
tests/test_cloudformation/fixtures/route53_ec2_instance_with_public_ip.py Normal file
View File

@ -0,0 +1,40 @@
from __future__ import unicode_literals
template = {
"Resources" : {
"Ec2Instance" : {
"Type" : "AWS::EC2::Instance",
"Properties" : {
"ImageId" : "ami-1234abcd",
"PrivateIpAddress": "10.0.0.25",
}
},
"HostedZone": {
"Type" : "AWS::Route53::HostedZone",
"Properties" : {
"Name" : "my_zone"
}
},
"myDNSRecord" : {
"Type" : "AWS::Route53::RecordSet",
"Properties" : {
"HostedZoneName" : { "Ref" : "HostedZone" },
"Comment" : "DNS name for my instance.",
"Name" : {
"Fn::Join" : [ "", [
{"Ref" : "Ec2Instance"}, ".",
{"Ref" : "AWS::Region"}, ".",
{"Ref" : "HostedZone"} ,"."
] ]
},
"Type" : "A",
"TTL" : "900",
"ResourceRecords" : [
{ "Fn::GetAtt" : [ "Ec2Instance", "PrivateIp" ] }
]
}
}
},
}

39
tests/test_cloudformation/fixtures/route53_health_check.py Normal file
View File

@ -0,0 +1,39 @@
from __future__ import unicode_literals
template = {
"Resources" : {
"HostedZone": {
"Type" : "AWS::Route53::HostedZone",
"Properties" : {
"Name" : "my_zone"
}
},
"my_health_check": {
"Type": "AWS::Route53::HealthCheck",
"Properties" : {
"HealthCheckConfig" : {
"FailureThreshold" : 3,
"IPAddress" : "10.0.0.4",
"Port" : 80,
"RequestInterval" : 10,
"ResourcePath" : "/",
"Type" : "HTTP",
}
}
},
"myDNSRecord" : {
"Type" : "AWS::Route53::RecordSet",
"Properties" : {
"HostedZoneName" : { "Ref" : "HostedZone" },
"Comment" : "DNS name for my instance.",
"Name" : "my_record_set",
"Type" : "A",
"TTL" : "900",
"ResourceRecords" : ["my.example.com"],
"HealthCheckId": {"Ref": "my_health_check"},
}
}
},
}

47
tests/test_cloudformation/fixtures/route53_roundrobin.py Normal file
View File

@ -0,0 +1,47 @@
from __future__ import unicode_literals
template = {
"AWSTemplateFormatVersion" : "2010-09-09",
"Description" : "AWS CloudFormation Sample Template Route53_RoundRobin: Sample template showing how to use weighted round robin (WRR) DNS entried via Amazon Route 53. This contrived sample uses weighted CNAME records to illustrate that the weighting influences the return records. It assumes that you already have a Hosted Zone registered with Amazon Route 53. **WARNING** This template creates one or more AWS resources. You will be billed for the AWS resources used if you create a stack from this template.",
"Resources" : {
"MyZone": {
"Type" : "AWS::Route53::HostedZone",
"Properties" : {
"Name" : "my_zone"
}
},
"MyDNSRecord" : {
"Type" : "AWS::Route53::RecordSetGroup",
"Properties" : {
"HostedZoneName" : {"Ref": "MyZone"},
"Comment" : "Contrived example to redirect to aws.amazon.com 75% of the time and www.amazon.com 25% of the time.",
"RecordSets" : [{
"SetIdentifier" : { "Fn::Join" : [ " ", [{"Ref" : "AWS::StackName"}, "AWS" ]]},
"Name" : { "Fn::Join" : [ "", [{"Ref" : "AWS::StackName"}, ".", {"Ref" : "AWS::Region"}, ".", {"Ref" : "MyZone"}, "."]]},
"Type" : "CNAME",
"TTL" : "900",
"ResourceRecords" : ["aws.amazon.com"],
"Weight" : "3"
},{
"SetIdentifier" : { "Fn::Join" : [ " ", [{"Ref" : "AWS::StackName"}, "Amazon" ]]},
"Name" : { "Fn::Join" : [ "", [{"Ref" : "AWS::StackName"}, ".", {"Ref" : "AWS::Region"}, ".", {"Ref" : "MyZone"}, "."]]},
"Type" : "CNAME",
"TTL" : "900",
"ResourceRecords" : ["www.amazon.com"],
"Weight" : "1"
}]
}
}
},
"Outputs" : {
"DomainName" : {
"Description" : "Fully qualified domain name",
"Value" : { "Ref" : "MyDNSRecord" }
}
}
}

167
tests/test_cloudformation/test_cloudformation_stack_integration.py
View File

@ -8,6 +8,7 @@ import boto.ec2.autoscale
import boto.ec2.elb
from boto.exception import BotoServerError
import boto.iam
import boto.sns
import boto.sqs
import boto.vpc
import sure # noqa
@ -19,6 +20,8 @@ from moto import (
mock_elb,
mock_iam,
mock_rds,
mock_route53,
mock_sns,
mock_sqs,
)
@ -26,6 +29,9 @@ from .fixtures import (
ec2_classic_eip,
fn_join,
rds_mysql_with_read_replica,
route53_ec2_instance_with_public_ip,
route53_health_check,
route53_roundrobin,
single_instance_with_ebs_volume,
vpc_eip,
vpc_single_instance_in_subnet,
@ -769,3 +775,164 @@ def test_cloudformation_mapping():
reservation = ec2_conn.get_all_instances()[0]
ec2_instance = reservation.instances[0]
ec2_instance.image_id.should.equal("ami-c9c7978c")
@mock_cloudformation()
@mock_route53()
def test_route53_roundrobin():
route53_conn = boto.connect_route53()
template_json = json.dumps(route53_roundrobin.template)
conn = boto.cloudformation.connect_to_region("us-west-1")
stack = conn.create_stack(
"test_stack",
template_body=template_json,
)
zones = route53_conn.get_all_hosted_zones()['ListHostedZonesResponse']['HostedZones']
list(zones).should.have.length_of(1)
zone_id = zones[0]['Id']
rrsets = route53_conn.get_all_rrsets(zone_id)
rrsets.hosted_zone_id.should.equal(zone_id)
rrsets.should.have.length_of(2)
record_set1 = rrsets[0]
record_set1.name.should.equal('test_stack.us-west-1.my_zone.')
record_set1.identifier.should.equal("test_stack AWS")
record_set1.type.should.equal('CNAME')
record_set1.ttl.should.equal('900')
record_set1.weight.should.equal('3')
record_set1.resource_records[0].should.equal("aws.amazon.com")
record_set2 = rrsets[1]
record_set2.name.should.equal('test_stack.us-west-1.my_zone.')
record_set2.identifier.should.equal("test_stack Amazon")
record_set2.type.should.equal('CNAME')
record_set2.ttl.should.equal('900')
record_set2.weight.should.equal('1')
record_set2.resource_records[0].should.equal("www.amazon.com")
stack = conn.describe_stacks()[0]
output = stack.outputs[0]
output.key.should.equal('DomainName')
output.value.should.equal('arn:aws:route53:::hostedzone/{0}'.format(zone_id))
@mock_cloudformation()
@mock_ec2()
@mock_route53()
def test_route53_ec2_instance_with_public_ip():
route53_conn = boto.connect_route53()
ec2_conn = boto.ec2.connect_to_region("us-west-1")
template_json = json.dumps(route53_ec2_instance_with_public_ip.template)
conn = boto.cloudformation.connect_to_region("us-west-1")
conn.create_stack(
"test_stack",
template_body=template_json,
)
instance_id = ec2_conn.get_all_reservations()[0].instances[0].id
zones = route53_conn.get_all_hosted_zones()['ListHostedZonesResponse']['HostedZones']
list(zones).should.have.length_of(1)
zone_id = zones[0]['Id']
rrsets = route53_conn.get_all_rrsets(zone_id)
rrsets.should.have.length_of(1)
record_set1 = rrsets[0]
record_set1.name.should.equal('{0}.us-west-1.my_zone.'.format(instance_id))
record_set1.identifier.should.equal(None)
record_set1.type.should.equal('A')
record_set1.ttl.should.equal('900')
record_set1.weight.should.equal(None)
record_set1.resource_records[0].should.equal("10.0.0.25")
@mock_cloudformation()
@mock_route53()
def test_route53_associate_health_check():
route53_conn = boto.connect_route53()
template_json = json.dumps(route53_health_check.template)
conn = boto.cloudformation.connect_to_region("us-west-1")
conn.create_stack(
"test_stack",
template_body=template_json,
)
checks = route53_conn.get_list_health_checks()['ListHealthChecksResponse']['HealthChecks']
list(checks).should.have.length_of(1)
check = checks[0]
health_check_id = check['Id']
config = check['HealthCheckConfig']
config["FailureThreshold"].should.equal("3")
config["IPAddress"].should.equal("10.0.0.4")
config["Port"].should.equal("80")
config["RequestInterval"].should.equal("10")
config["ResourcePath"].should.equal("/")
config["Type"].should.equal("HTTP")
zones = route53_conn.get_all_hosted_zones()['ListHostedZonesResponse']['HostedZones']
list(zones).should.have.length_of(1)
zone_id = zones[0]['Id']
rrsets = route53_conn.get_all_rrsets(zone_id)
rrsets.should.have.length_of(1)
record_set = rrsets[0]
record_set.health_check.should.equal(health_check_id)
@mock_cloudformation()
@mock_sns()
def test_sns_topic():
dummy_template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
"MySNSTopic": {
"Type": "AWS::SNS::Topic",
"Properties": {
"Subscription": [
{"Endpoint": "https://example.com", "Protocol": "https"},
],
"TopicName": "my_topics",
}
}
},
"Outputs": {
"topic_name": {
"Value": {"Fn::GetAtt": ["MySNSTopic", "TopicName"]}
},
"topic_arn": {
"Value": {"Ref": "MySNSTopic"}
},
}
}
template_json = json.dumps(dummy_template)
conn = boto.cloudformation.connect_to_region("us-west-1")
stack = conn.create_stack(
"test_stack",
template_body=template_json,
)
sns_conn = boto.sns.connect_to_region("us-west-1")
topics = sns_conn.get_all_topics()["ListTopicsResponse"]["ListTopicsResult"]["Topics"]
topics.should.have.length_of(1)
topic_arn = topics[0]['TopicArn']
topic_arn.should.contain("my_topics")
subscriptions = sns_conn.get_all_subscriptions()["ListSubscriptionsResponse"]["ListSubscriptionsResult"]["Subscriptions"]
subscriptions.should.have.length_of(1)
subscription = subscriptions[0]
subscription["TopicArn"].should.equal(topic_arn)
subscription["Protocol"].should.equal("https")
subscription["SubscriptionArn"].should.contain(topic_arn)
subscription["Endpoint"].should.equal("https://example.com")
stack = conn.describe_stacks()[0]
topic_name_output = [x for x in stack.outputs if x.key == 'topic_name'][0]
topic_name_output.value.should.equal("my_topics")
topic_arn_output = [x for x in stack.outputs if x.key == 'topic_arn'][0]
topic_arn_output.value.should.equal(topic_arn)

77
tests/test_route53/test_route53.py
View File

@ -1,6 +1,7 @@
from __future__ import unicode_literals
import boto
from boto.route53.healthcheck import HealthCheck
from boto.route53.record import ResourceRecordSets
import sure # noqa
@ -89,3 +90,79 @@ def test_rrset():
rrsets = conn.get_all_rrsets(zoneid, name="foo.foo.testdns.aws.com", type="A")
rrsets.should.have.length_of(0)
@mock_route53
def test_create_health_check():
conn = boto.connect_route53('the_key', 'the_secret')
check = HealthCheck(
ip_addr="10.0.0.25",
port=80,
hc_type="HTTP",
resource_path="/",
fqdn="example.com",
string_match="a good response",
request_interval=10,
failure_threshold=2,
)
conn.create_health_check(check)
checks = conn.get_list_health_checks()['ListHealthChecksResponse']['HealthChecks']
list(checks).should.have.length_of(1)
check = checks[0]
config = check['HealthCheckConfig']
config['IPAddress'].should.equal("10.0.0.25")
config['Port'].should.equal("80")
config['Type'].should.equal("HTTP")
config['ResourcePath'].should.equal("/")
config['FullyQualifiedDomainName'].should.equal("example.com")
config['SearchString'].should.equal("a good response")
config['RequestInterval'].should.equal("10")
config['FailureThreshold'].should.equal("2")
@mock_route53
def test_delete_health_check():
conn = boto.connect_route53('the_key', 'the_secret')
check = HealthCheck(
ip_addr="10.0.0.25",
port=80,
hc_type="HTTP",
resource_path="/",
)
conn.create_health_check(check)
checks = conn.get_list_health_checks()['ListHealthChecksResponse']['HealthChecks']
list(checks).should.have.length_of(1)
health_check_id = checks[0]['Id']
conn.delete_health_check(health_check_id)
checks = conn.get_list_health_checks()['ListHealthChecksResponse']['HealthChecks']
list(checks).should.have.length_of(0)
@mock_route53
def test_use_health_check_in_resource_record_set():
conn = boto.connect_route53('the_key', 'the_secret')
check = HealthCheck(
ip_addr="10.0.0.25",
port=80,
hc_type="HTTP",
resource_path="/",
)
check = conn.create_health_check(check)['CreateHealthCheckResponse']['HealthCheck']
check_id = check['Id']
zone = conn.create_hosted_zone("testdns.aws.com")
zone_id = zone["CreateHostedZoneResponse"]["HostedZone"]["Id"].split("/")[-1]
changes = ResourceRecordSets(conn, zone_id)
change = changes.add_change("CREATE", "foo.bar.testdns.aws.com", "A", health_check=check_id)
change.add_value("1.2.3.4")
changes.commit()
record_sets = conn.get_all_rrsets(zone_id)
record_sets[0].health_check.should.equal(check_id)