140 lines
4.4 KiB
Python
140 lines
4.4 KiB
Python
from __future__ import unicode_literals
|
|
|
|
from jinja2 import Template
|
|
|
|
from moto.core import BaseBackend
|
|
from moto.core.utils import get_random_hex
|
|
|
|
|
|
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')
|
|
|
|
@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>
|
|
<HealthCheckId></HealthCheckId>
|
|
</ResourceRecordSet>""")
|
|
return template.render(record_set=self)
|
|
|
|
|
|
class FakeZone(object):
|
|
|
|
def __init__(self, name, id_):
|
|
self.name = name
|
|
self.id = id_
|
|
self.rrsets = []
|
|
|
|
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 = [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 = {}
|
|
|
|
def create_hosted_zone(self, name):
|
|
new_id = get_random_hex()
|
|
new_zone = FakeZone(name, new_id)
|
|
self.zones[new_id] = new_zone
|
|
return new_zone
|
|
|
|
def get_all_hosted_zones(self):
|
|
return self.zones.values()
|
|
|
|
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
|
|
|
|
|
|
route53_backend = Route53Backend()
|