Remove reliance on YamlProvider for delegation

octoblox/__init__.py

@@ -3,6 +3,7 @@
 from collections import defaultdict
 from functools import lru_cache
 from octodns.provider.base import BaseProvider
+from octodns.source.base import BaseSource
 from octodns.record import Change, Record
 
 # fmt: off
@@ -355,6 +356,9 @@ def populate(self, zone, target=False, lenient=False):
         return True
 
     def _extra_changes(self, existing, changes, **kwargs):
+        self.log.debug(
+            '_extra_changes: zone=%s, len(changes)=%d', existing.name, len(changes)
+        )
         if self.create_zones and not existing.exists and not changes:
             return [
                 Create(
@@ -474,3 +478,27 @@ def _apply(self, plan):
             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')
+
+
+class EmptySource(BaseSource):
+    """docstring for EmptySource"""
+
+    SUPPORTS = {'NS'}
+    SUPPORTS_GEO = False
+    SUPPORTS_DYNAMIC = False
+    SUPPORTS_ROOT_NS = True
+
+    def __init__(self, id, *args, **kwargs):
+        self.log = logging.getLogger(f'{self.__class__.__name__}[{id}]')
+        self.log.debug('__init__')
+        super().__init__(id, *args, **kwargs)
+
+    def populate(self, zone, target=False, lenient=False):
+        self.log.debug(
+            'populate: name=%s, target=%s, lenient=%s, len(records)=%d',
+            zone.name,
+            target,
+            lenient,
+            len(zone.records),
+        )
+        return True

tests/conftest.py

@@ -1,9 +1,13 @@
 import re
 import uuid
+import logging
+
 import pytest
 from urllib.parse import urlparse, quote_plus
 from octoblox import InfoBloxProvider, DelegatedProvider
 
+logging.basicConfig(level='DEBUG')
+
 
 @pytest.fixture
 def zone_name():

tests/test_delegated_provider.py

@@ -1,61 +1,79 @@
-import os
-import logging
+import pytest
+from octoblox import EmptySource
 from octodns.provider.yaml import YamlProvider
 from octodns.zone import Zone
+from pathlib import Path
 
-logging.basicConfig(level='INFO')
+keys = 'result,source'
+values = [
+    (0, lambda: EmptySource('E')),
+    (1, lambda: YamlProvider('Y', Path(__file__).parent / 'config')),
+]
 
 
-def test_zone_data(delegated_provider, zone_name):
+@pytest.mark.parametrize(keys, values)
+def test_zone_data(delegated_provider, zone_name, result, source):
     expected = Zone(zone_name, [])
-    source = YamlProvider('test', os.path.join(os.path.dirname(__file__), 'delegated'))
+    source = source()
     source.populate(expected)
+    assert len(expected.records) == 5 * result
     zone = Zone(zone_name, [])
     delegated_provider.populate(zone, lenient=True)
     assert len(zone.records) == 0
     plan = delegated_provider.plan(expected)
-    delegated_provider.apply(plan)
+    if result:
+        delegated_provider.apply(plan)
+    else:
+        assert plan is None
 
 
-def test_zone_creation(delegated_provider, new_zone_name):
+@pytest.mark.parametrize(keys, values)
+def test_zone_creation(delegated_provider, new_zone_name, result, source):
     expected = Zone(new_zone_name, [])
-    source = YamlProvider('test', os.path.join(os.path.dirname(__file__), 'delegated'))
+    source = source()
     source.populate(expected)
-    assert len(expected.records) == 0
+    assert len(expected.records) == 4 * result
     zone = Zone(new_zone_name, [])
     delegated_provider.populate(zone)
+    assert len(zone.records) == 0
     plan = delegated_provider.plan(expected)
     delegated_provider.apply(plan)
 
 
-def test_empty_zone_creation(delegated_provider, empty_zone_name):
+@pytest.mark.parametrize(keys, values)
+def test_empty_zone_creation(delegated_provider, empty_zone_name, result, source):
     expected = Zone(empty_zone_name, [])
-    source = YamlProvider('test', os.path.join(os.path.dirname(__file__), 'delegated'))
+    source = source()
     source.populate(expected)
     assert len(expected.records) == 0
     zone = Zone(empty_zone_name, [])
     delegated_provider.populate(zone)
+    assert len(zone.records) == 0
     plan = delegated_provider.plan(expected)
     delegated_provider.apply(plan)
 
 
-def test_ipv4_zone(delegated_provider, new_ipv4_zone):
+@pytest.mark.parametrize(keys, values)
+def test_ipv4_zone(delegated_provider, new_ipv4_zone, result, source):
     expected = Zone(new_ipv4_zone, [])
-    source = YamlProvider('test', os.path.join(os.path.dirname(__file__), 'delegated'))
+    source = source()
     source.populate(expected)
-    assert len(expected.records) == 0
+    assert len(expected.records) == 1 * result
     zone = Zone(new_ipv4_zone, [])
     delegated_provider.populate(zone)
+    assert len(zone.records) == 0
     plan = delegated_provider.plan(expected)
     delegated_provider.apply(plan)
 
 
-def test_ipv6_zone(delegated_provider, new_ipv6_zone):
+@pytest.mark.parametrize(keys, values)
+def test_ipv6_zone(delegated_provider, new_ipv6_zone, result, source):
     expected = Zone(new_ipv6_zone, [])
-    source = YamlProvider('test', os.path.join(os.path.dirname(__file__), 'delegated'))
+    source = source()
     source.populate(expected)
-    assert len(expected.records) == 0
+    assert len(expected.records) == 1 * result
     zone = Zone(new_ipv6_zone, [])
     delegated_provider.populate(zone)
+    assert len(zone.records) == 0
     plan = delegated_provider.plan(expected)
     delegated_provider.apply(plan)

tests/test_infoblox_provider.py

@@ -1,10 +1,7 @@
 import os
-import logging
 from octodns.provider.yaml import YamlProvider
 from octodns.zone import Zone
 
-logging.basicConfig(level='INFO')
-
 
 def test_zone_data(provider, zone_name):
     expected = Zone(zone_name, [])