Add functionality to get r53 zone id by name

octodns/provider/route53.py

@@ -610,9 +610,11 @@ class Route53Provider(BaseProvider):
 
     def __init__(self, id, access_key_id=None, secret_access_key=None,
                  max_changes=1000, client_max_attempts=None,
-                 session_token=None, delegation_set_id=None, *args, **kwargs):
+                 session_token=None, delegation_set_id=None,
+                 get_zones_by_name=False, *args, **kwargs):
         self.max_changes = max_changes
         self.delegation_set_id = delegation_set_id
+        self.get_zones_by_name = get_zones_by_name
         _msg = f'access_key_id={access_key_id}, secret_access_key=***, ' \
                'session_token=***'
         use_fallback_auth = access_key_id is None and \
@@ -661,7 +663,18 @@ class Route53Provider(BaseProvider):
 
     def _get_zone_id(self, name, create=False):
         self.log.debug('_get_zone_id: name=%s', name)
-        if name in self.r53_zones:
+        if self.get_zones_by_name:
+            # attempt to get zone by name
+            # limited to one as this should be unique
+            response = self._conn.list_hosted_zones_by_name(
+                DNSName=name, MaxItems="1"
+            )
+            if len(response['HostedZones']) == 1:
+                # if there is a single response
+                id = response['HostedZones'][0]['Id']
+                self.log.debug(id)
+                return id
+        elif name in self.r53_zones:
             id = self.r53_zones[name]
             self.log.debug('_get_zone_id:   id=%s', id)
             return id

tests/test_octodns_provider_route53.py

@@ -1745,6 +1745,82 @@ class TestRoute53Provider(TestCase):
         self.assertEquals([], extra)
         stubber.assert_no_pending_responses()
 
+    def test_no_changes_with_get_zones_by_name(self):
+        provider = Route53Provider(
+            'test', 'abc', '123', get_zones_by_name=True)
+
+        # Use the stubber
+        stubber = Stubber(provider._conn)
+        stubber.activate()
+
+        list_hosted_zones_by_name_resp = {
+            'HostedZones': [{
+                'Id': 'z42',
+                'Name': 'unit.tests.',
+                'CallerReference': 'abc',
+                'Config': {
+                    'Comment': 'string',
+                    'PrivateZone': False
+                },
+                'ResourceRecordSetCount': 123,
+            }, ],
+            'DNSName': 'unit.tests.',
+            'HostedZoneId': 'z42',
+            'IsTruncated': False,
+            'MaxItems': 'string'
+        }
+
+        stubber.add_response(
+            'list_hosted_zones_by_name',
+            list_hosted_zones_by_name_resp,
+            {'DNSName': 'unit.tests.', 'MaxItems': '1'}
+        )
+
+        # empty is empty
+        desired = Zone('unit.tests.', [])
+        extra = provider._extra_changes(desired=desired, changes=[])
+        self.assertEquals([], extra)
+        stubber.assert_no_pending_responses()
+
+    def test_plan_with_get_zones_by_name(self):
+        provider = Route53Provider(
+            'test', 'abc', '123', get_zones_by_name=True)
+
+        # Use the stubber
+        stubber = Stubber(provider._conn)
+        stubber.activate()
+
+        # this is an empty response
+        # zone name not found
+        list_hosted_zones_by_name_resp = {
+            'HostedZones': [],
+            'DNSName': 'unit.tests.',
+            'HostedZoneId': 'z42',
+            'IsTruncated': False,
+            'MaxItems': 'string'
+        }
+
+        # list_hosted_zones_by_name gets called 3 times in this process
+        # so adding 3 responses
+        stubber.add_response(
+            'list_hosted_zones_by_name',
+            list_hosted_zones_by_name_resp,
+            {'DNSName': 'unit.tests.', 'MaxItems': '1'}
+        )
+        stubber.add_response(
+            'list_hosted_zones_by_name',
+            list_hosted_zones_by_name_resp,
+            {'DNSName': 'unit.tests.', 'MaxItems': '1'}
+        )
+        stubber.add_response(
+            'list_hosted_zones_by_name',
+            list_hosted_zones_by_name_resp,
+            {'DNSName': 'unit.tests.', 'MaxItems': '1'}
+        )
+
+        plan = provider.plan(self.expected)
+        self.assertEquals(9, len(plan.changes))
+
     def test_extra_change_no_health_check(self):
         provider, stubber = self._get_stubbed_provider()