Put delegated zone logic in own provider

README.md

@@ -44,7 +44,24 @@ providers:
     #   soa_default_ttl: 3600
     #   view: default
     #   use_grid_zone_timer: true
-    # zone_type: zone_delegated
+  delegated:
+    class: octoblox.DelegatedProvider
+    endpoint: infoblox.example.com
+    username: admin
+    password: env/INFOBLOX_PASSWORD
+    # verify: ./infoblox.pem
+    # apiver: 1.0
+    # dns_view: default
+    # log_change: true
+    # create_zones: true
+    # new_zone_fields:
+    #   delegate_to:
+    #     - name: ns1.delegated.example.com
+    #       address: 1.1.1.1
+    #     - name: ns2.delegated.example.com
+    #       address: 8.8.8.8
+    #   ns_group: default
+    #   view: default
 ```
 
 ## Alias Record Update Behaviour

octoblox/__init__.py

@@ -80,7 +80,6 @@ def __init__(
         log_change=False,
         new_zone_fields=None,
         log=None,
-        zone_type='zone_auth',
     ):
         super(InfoBlox, self).__init__()
         self.fqdn = fqdn
@@ -92,7 +91,6 @@ def __init__(
         self.log_change = log_change
         self.new_zone_fields = new_zone_fields or {}
         self.log = log
-        self.zone_type = zone_type or 'zone_auth'
         if not apiver:  # pragma: no branch
             self.apiver = self.get_api_version()
 
@@ -141,25 +139,25 @@ def get_zone_fqdn(self, zone):
         else:
             return zone[:-1]
 
-    def get_zone(self, zone):
+    def get_zone(self, zone, zone_type='zone_auth', return_fields='soa_default_ttl'):
         return self.get(
-            self.zone_type,
+            zone_type,
             params={
                 'fqdn': self.get_zone_fqdn(zone),
-                '_return_fields+': 'soa_default_ttl',
+                '_return_fields+': return_fields,
                 **({'view': self.dns_view} if self.dns_view else {}),
             },
         ).json()
 
-    def add_zone(self, zone):
+    def add_zone(self, zone, zone_type='zone_auth', return_fields='soa_default_ttl'):
         fqdn = self.get_zone_fqdn(zone)
-        zone_format = 'IPV6' if ':' in fqdn else 'IPV4' if '/' in fqdn else 'FORWARDING'
+        zone_format = 'IPV6' if ':' in fqdn else 'IPV4' if '/' in fqdn else 'FORWARD'
         return self.post(
-            self.zone_type,
+            zone_type,
             json={
                 'fqdn': fqdn,
                 'zone_format': zone_format,
-                '_return_fields+': 'soa_default_ttl',
+                '_return_fields+': return_fields,
                 **self.new_zone_fields,
             },
         ).json()
@@ -267,7 +265,6 @@ def __init__(
         log_change=False,
         create_zones=False,
         new_zone_fields=None,
-        zone_type='zone_auth',
         *args,
         **kwargs,
     ):
@@ -283,7 +280,6 @@ def __init__(
             log_change,
             new_zone_fields,
             self.log,
-            zone_type,
         )
         self.create_zones = create_zones
         self.log.debug(
@@ -449,3 +445,32 @@ def _apply(self, plan):
                 continue
             class_name = change.__class__.__name__
             getattr(self, f'_apply_{class_name}')(zone[:-1], change, default_ttl)
+
+
+class DelegatedProvider(InfoBloxProvider):
+    def populate(self, zone, target=False, lenient=False):
+        self.log.debug(
+            'populate: name=%s, target=%s, lenient=%s', zone.name, target, lenient
+        )
+
+        zone_data = self.conn.get_zone(zone.name, 'zone_delegated', 'delegated_ttl')
+
+        zone.exists = bool(zone_data)
+
+        if not zone_data:
+            if target and not self.create_zones:  # pragma: no cover
+                raise ValueError(f'Zone does not exist in InfoBlox: {zone.name}')
+            return False
+
+        return True
+
+    def _apply(self, plan):
+
+        zone = plan.desired.name
+
+        zone_data = self.conn.get_zone(zone, 'zone_delegated', 'delegated_ttl')
+
+        if not zone_data:
+            if not self.create_zones:  # pragma: no cover
+                raise ValueError(f'Zone does not exist in InfoBlox: {zone}')
+            self.conn.add_zone(zone, 'zone_delegated', 'delegated_ttl')

tests/conftest.py

@@ -2,7 +2,7 @@
 import uuid
 import pytest
 from urllib.parse import urlparse, quote_plus
-from octoblox import InfoBloxProvider
+from octoblox import InfoBloxProvider, DelegatedProvider
 
 
 @pytest.fixture
@@ -76,9 +76,20 @@ def zones(zone_name, new_zone_name, empty_zone_name, new_ipv4_cidr, new_ipv6_cid
         ],
         f'/wapi/v1.0/zone_auth?fqdn={new_zone_name[:-1]}': [],
         f'/wapi/v1.0/zone_auth?fqdn={empty_zone_name[:-1]}': [],
-        f'/wapi/v1.0/zone_delegated?fqdn={empty_zone_name[:-1]}': [],
         f'/wapi/v1.0/zone_auth?fqdn={quote_plus(new_ipv4_cidr)}': [],
         f'/wapi/v1.0/zone_auth?fqdn={quote_plus(new_ipv6_cidr)}': [],
+        f'/wapi/v1.0/zone_delegated?fqdn={zone_name[:-1]}': [
+            {
+                '_ref': f'zone_delegated/{uuid.uuid4()}:{zone_name[:-1]}/default',
+                'fqdn': zone_name[:-1],
+                'view': 'default',
+                'delegated_ttl': 28800,
+            }
+        ],
+        f'/wapi/v1.0/zone_delegated?fqdn={new_zone_name[:-1]}': [],
+        f'/wapi/v1.0/zone_delegated?fqdn={empty_zone_name[:-1]}': [],
+        f'/wapi/v1.0/zone_delegated?fqdn={quote_plus(new_ipv4_cidr)}': [],
+        f'/wapi/v1.0/zone_delegated?fqdn={quote_plus(new_ipv6_cidr)}': [],
     }
 
 
@@ -144,8 +155,7 @@ def records(zone_name):
     return (re.compile('/wapi/v1.0/record:\\w+([?/]|$)'), get_records(zone_name[:-1]))
 
 
-@pytest.fixture
-def provider(requests_mock, zones, records, schema):
+def make_provider(cls, requests_mock, zones, records, schema):
     requests_mock.get(schema[0], json=schema[1])
     for url, data in zones.items():
         requests_mock.get(url, json=data)
@@ -164,11 +174,21 @@ def provider(requests_mock, zones, records, schema):
     requests_mock.delete(records[0], status_code=200)
     requests_mock.post(records[0], status_code=201)
     requests_mock.put(records[0], status_code=200)
-    return InfoBloxProvider(
+    return cls(
         'test',
         'non.existent',
         'username',
         'password',
         log_change=True,
         create_zones=True,
     )
+
+
+@pytest.fixture
+def provider(requests_mock, zones, records, schema):
+    return make_provider(InfoBloxProvider, requests_mock, zones, records, schema)
+
+
+@pytest.fixture
+def delegated_provider(requests_mock, zones, records, schema):
+    return make_provider(DelegatedProvider, requests_mock, zones, records, schema)

tests/delegated/12.11.10.in-addr.arpa.yaml

@@ -0,0 +1 @@
+--- {}

tests/delegated/create.tests.yaml

@@ -0,0 +1 @@
+--- {}

tests/delegated/empty.tests.yaml

@@ -0,0 +1 @@
+--- {}

tests/delegated/f.e.d.c.b.a.9.8.7.6.5.4.3.2.1.0.ip6.arpa.yaml

@@ -0,0 +1 @@
+--- {}

tests/delegated/unit.tests.yaml

@@ -0,0 +1,23 @@
+---
+a:
+  ttl: 28800
+  type: A
+  value: 1.2.3.4
+alias:
+  octodns:
+    lenient: true
+  type: ALIAS
+  value: www.unit.tests.
+alias-update:
+  octodns:
+    lenient: true
+  ttl: 28800
+  type: ALIAS
+  value: www.unit.tests.
+cname:
+  ttl: 28800
+  type: CNAME
+  value: example2.unit.tests.
+www:
+  type: A
+  value: 192.168.0.2

tests/test_delegated_provider.py

@@ -0,0 +1,61 @@
+import os
+import logging
+from octodns.provider.yaml import YamlProvider
+from octodns.zone import Zone
+
+logging.basicConfig(level='INFO')
+
+
+def test_zone_data(delegated_provider, zone_name):
+    expected = Zone(zone_name, [])
+    source = YamlProvider('test', os.path.join(os.path.dirname(__file__), 'delegated'))
+    source.populate(expected)
+    zone = Zone(zone_name, [])
+    delegated_provider.populate(zone, lenient=True)
+    assert len(zone.records) == 0
+    plan = delegated_provider.plan(expected)
+    delegated_provider.apply(plan)
+
+
+def test_zone_creation(delegated_provider, new_zone_name):
+    expected = Zone(new_zone_name, [])
+    source = YamlProvider('test', os.path.join(os.path.dirname(__file__), 'delegated'))
+    source.populate(expected)
+    assert len(expected.records) == 0
+    zone = Zone(new_zone_name, [])
+    delegated_provider.populate(zone)
+    plan = delegated_provider.plan(expected)
+    delegated_provider.apply(plan)
+
+
+def test_empty_zone_creation(delegated_provider, empty_zone_name):
+    expected = Zone(empty_zone_name, [])
+    source = YamlProvider('test', os.path.join(os.path.dirname(__file__), 'delegated'))
+    source.populate(expected)
+    assert len(expected.records) == 0
+    zone = Zone(empty_zone_name, [])
+    delegated_provider.populate(zone)
+    plan = delegated_provider.plan(expected)
+    delegated_provider.apply(plan)
+
+
+def test_ipv4_zone(delegated_provider, new_ipv4_zone):
+    expected = Zone(new_ipv4_zone, [])
+    source = YamlProvider('test', os.path.join(os.path.dirname(__file__), 'delegated'))
+    source.populate(expected)
+    assert len(expected.records) == 0
+    zone = Zone(new_ipv4_zone, [])
+    delegated_provider.populate(zone)
+    plan = delegated_provider.plan(expected)
+    delegated_provider.apply(plan)
+
+
+def test_ipv6_zone(delegated_provider, new_ipv6_zone):
+    expected = Zone(new_ipv6_zone, [])
+    source = YamlProvider('test', os.path.join(os.path.dirname(__file__), 'delegated'))
+    source.populate(expected)
+    assert len(expected.records) == 0
+    zone = Zone(new_ipv6_zone, [])
+    delegated_provider.populate(zone)
+    plan = delegated_provider.plan(expected)
+    delegated_provider.apply(plan)

tests/test_infoblox_provider.py

@@ -39,18 +39,6 @@ def test_empty_zone_creation(provider, empty_zone_name):
     provider.apply(plan)
 
 
-def test_delegated_zone_creation(provider, empty_zone_name):
-    provider.conn.zone_type = 'zone_delegated'
-    expected = Zone(empty_zone_name, [])
-    source = YamlProvider('test', os.path.join(os.path.dirname(__file__), 'config'))
-    source.populate(expected)
-    assert len(expected.records) == 0
-    zone = Zone(empty_zone_name, [])
-    provider.populate(zone)
-    plan = provider.plan(expected)
-    provider.apply(plan)
-
-
 def test_ipv4_zone(provider, new_ipv4_zone):
     expected = Zone(new_ipv4_zone, [])
     source = YamlProvider('test', os.path.join(os.path.dirname(__file__), 'config'))