Backwards compat for plan overrides, 100% manager coverage, singular processor module name

octodns/manager.py

@@ -263,7 +263,7 @@ class Manager(object):
                 except TypeError as e:
                     if "keyword argument 'lenient'" not in text_type(e):
                         raise
-                    self.log.warn(': provider %s does not accept lenient '
+                    self.log.warn('provider %s does not accept lenient '
                                   'param', source.__class__.__name__)
                     source.populate(zone)
 
@@ -281,9 +281,15 @@ class Manager(object):
                     'value': 'provider={}'.format(target.id)
                 })
                 zone.add_record(meta, replace=True)
-            # TODO: if someone has overrriden plan already this will be a
-            # breaking change so we probably need to try both ways
-            plan = target.plan(zone, processors=processors)
+            try:
+                plan = target.plan(zone, processors=processors)
+            except TypeError as e:
+                if "keyword argument 'processors'" not in text_type(e):
+                    raise
+                self.log.warn('provider.plan %s does not accept processors '
+                              'param', target.__class__.__name__)
+                plan = target.plan(zone)
+
             for processor in processors:
                 plan = processor.process_plan(plan, sources=sources,
                                               target=target)

octodns/processor/__init__.py

@@ -0,0 +1,6 @@
+#
+#
+#
+
+from __future__ import absolute_import, division, print_function, \
+    unicode_literals

tests/config/processors-missing-class.yaml

@@ -0,0 +1,23 @@
+providers:
+  config:
+    class: octodns.provider.yaml.YamlProvider
+    directory: tests/config
+  dump:
+    class: octodns.provider.yaml.YamlProvider
+    directory: env/YAML_TMP_DIR
+  geo:
+    class: helpers.GeoProvider
+  nosshfp:
+    class: helpers.NoSshFpProvider
+
+processors:
+  no-class: {}
+
+zones:
+  unit.tests.:
+    processors:
+    - noop
+    sources:
+    - in
+    targets:
+    - dump

tests/config/processors-wants-config.yaml

@@ -0,0 +1,25 @@
+providers:
+  config:
+    class: octodns.provider.yaml.YamlProvider
+    directory: tests/config
+  dump:
+    class: octodns.provider.yaml.YamlProvider
+    directory: env/YAML_TMP_DIR
+  geo:
+    class: helpers.GeoProvider
+  nosshfp:
+    class: helpers.NoSshFpProvider
+
+processors:
+  # valid class, but it wants a param and we're not passing it
+  wants-config:
+    class: helpers.WantsConfigProcessor
+
+zones:
+  unit.tests.:
+    processors:
+    - noop
+    sources:
+    - in
+    targets:
+    - dump

tests/config/processors.yaml

@@ -0,0 +1,33 @@
+providers:
+  config:
+    class: octodns.provider.yaml.YamlProvider
+    directory: tests/config
+  dump:
+    class: octodns.provider.yaml.YamlProvider
+    directory: env/YAML_TMP_DIR
+  geo:
+    class: helpers.GeoProvider
+  nosshfp:
+    class: helpers.NoSshFpProvider
+
+processors:
+  # Just testing config so any processor will do
+  noop:
+    class: octodns.processor.base.BaseProcessor
+
+zones:
+  unit.tests.:
+    processors:
+    - noop
+    sources:
+    - config
+    targets:
+    - dump
+
+  bad.unit.tests.:
+    processors:
+    - doesnt-exist
+    sources:
+    - in
+    targets:
+    - dump

tests/helpers.py

@@ -7,6 +7,10 @@ from __future__ import absolute_import, division, print_function, \
 
 from shutil import rmtree
 from tempfile import mkdtemp
+from logging import getLogger
+
+from octodns.processor.base import BaseProcessor
+from octodns.provider.base import BaseProvider
 
 
 class SimpleSource(object):
@@ -90,3 +94,29 @@ class TemporaryDirectory(object):
             rmtree(self.dirname)
         else:
             raise Exception(self.dirname)
+
+
+class WantsConfigProcessor(BaseProcessor):
+
+    def __init__(self, name, some_config):
+        super(WantsConfigProcessor, self).__init__(name)
+
+
+class PlannableProvider(BaseProvider):
+    log = getLogger('PlannableProvider')
+
+    SUPPORTS_GEO = False
+    SUPPORTS_DYNAMIC = False
+    SUPPORTS = set(('A',))
+
+    def __init__(self, *args, **kwargs):
+        super(PlannableProvider, self).__init__(*args, **kwargs)
+
+    def populate(self, zone, source=False, target=False, lenient=False):
+        pass
+
+    def supports(self, record):
+        return True
+
+    def __repr__(self):
+        return self.__class__.__name__

tests/test_octodns_manager.py

@@ -9,9 +9,10 @@ from os import environ
 from os.path import dirname, join
 from six import text_type
 
-from octodns.record import Record
 from octodns.manager import _AggregateTarget, MainThreadExecutor, Manager, \
     ManagerException
+from octodns.processor.base import BaseProcessor
+from octodns.record import Create, Delete, Record
 from octodns.yaml import safe_load
 from octodns.zone import Zone
 
@@ -19,7 +20,7 @@ from mock import MagicMock, patch
 from unittest import TestCase
 
 from helpers import DynamicProvider, GeoProvider, NoSshFpProvider, \
-    SimpleProvider, TemporaryDirectory
+    PlannableProvider, SimpleProvider, TemporaryDirectory
 
 config_dir = join(dirname(__file__), 'config')
 
@@ -358,20 +359,48 @@ class TestManager(TestCase):
 
             class NoLenient(SimpleProvider):
 
-                def populate(self, zone, source=False):
+                def populate(self, zone):
                     pass
 
             # This should be ok, we'll fall back to not passing it
             manager._populate_and_plan('unit.tests.', [], [NoLenient()], [])
 
-            class NoZone(SimpleProvider):
+            class OtherType(SimpleProvider):
 
-                def populate(self, lenient=False):
+                def populate(self, zone, lenient=False):
+                    raise TypeError('something else')
+
+            # This will blow up, we don't fallback for source
+            with self.assertRaises(TypeError) as ctx:
+                manager._populate_and_plan('unit.tests.', [], [OtherType()],
+                                           [])
+            self.assertEquals('something else', text_type(ctx.exception))
+
+    def test_plan_processors_fallback(self):
+        with TemporaryDirectory() as tmpdir:
+            environ['YAML_TMP_DIR'] = tmpdir.dirname
+            # Only allow a target that doesn't exist
+            manager = Manager(get_config_filename('simple.yaml'))
+
+            class NoProcessors(SimpleProvider):
+
+                def plan(self, zone):
                     pass
 
+            # This should be ok, we'll fall back to not passing it
+            manager._populate_and_plan('unit.tests.', [], [],
+                                       [NoProcessors()])
+
+            class OtherType(SimpleProvider):
+
+                def plan(self, zone, processors):
+                    raise TypeError('something else')
+
             # This will blow up, we don't fallback for source
-            with self.assertRaises(TypeError):
-                manager._populate_and_plan('unit.tests.', [NoZone()], [])
+            with self.assertRaises(TypeError) as ctx:
+                manager._populate_and_plan('unit.tests.', [], [],
+                                           [OtherType()])
+            self.assertEquals('something else', text_type(ctx.exception))
 
     @patch('octodns.manager.Manager._get_named_class')
     def test_sync_passes_file_handle(self, mock):
@@ -391,6 +420,92 @@ class TestManager(TestCase):
         _, kwargs = plan_output_mock.run.call_args
         self.assertEqual(fh_mock, kwargs.get('fh'))
 
+    def test_processor_config(self):
+        # Smoke test loading a valid config
+        manager = Manager(get_config_filename('processors.yaml'))
+        self.assertEquals(['noop'], list(manager.processors.keys()))
+        # This zone specifies a valid processor
+        manager.sync(['unit.tests.'])
+
+        with self.assertRaises(ManagerException) as ctx:
+            # This zone specifies a non-existant processor
+            manager.sync(['bad.unit.tests.'])
+        self.assertTrue('Zone bad.unit.tests., unknown processor: '
+                        'doesnt-exist' in text_type(ctx.exception))
+
+        with self.assertRaises(ManagerException) as ctx:
+            Manager(get_config_filename('processors-missing-class.yaml'))
+        self.assertTrue('Processor no-class is missing class' in
+                        text_type(ctx.exception))
+
+        with self.assertRaises(ManagerException) as ctx:
+            Manager(get_config_filename('processors-wants-config.yaml'))
+        self.assertTrue('Incorrect processor config for wants-config' in
+                        text_type(ctx.exception))
+
+    def test_processors(self):
+        manager = Manager(get_config_filename('simple.yaml'))
+
+        targets = [PlannableProvider('prov')]
+
+        zone = Zone('unit.tests.', [])
+        record = Record.new(zone, 'a', {
+            'ttl': 30,
+            'type': 'A',
+            'value': '1.2.3.4',
+        })
+
+        # muck with sources
+        class MockProcessor(BaseProcessor):
+
+            def process_source_zone(self, zone, sources):
+                zone = self._clone_zone(zone)
+                zone.add_record(record)
+                return zone
+
+        mock = MockProcessor('mock')
+        plans, zone = manager._populate_and_plan('unit.tests.', [mock], [],
+                                                 targets)
+        # Our mock was called and added the record
+        self.assertEquals(record, list(zone.records)[0])
+        # We got a create for the thing added to the expected state (source)
+        self.assertIsInstance(plans[0][1].changes[0], Create)
+
+        # muck with targets
+        class MockProcessor(BaseProcessor):
+
+            def process_target_zone(self, zone, target):
+                zone = self._clone_zone(zone)
+                zone.add_record(record)
+                return zone
+
+        mock = MockProcessor('mock')
+        plans, zone = manager._populate_and_plan('unit.tests.', [mock], [],
+                                                 targets)
+        # No record added since it's target this time
+        self.assertFalse(zone.records)
+        # We got a delete for the thing added to the existing state (target)
+        self.assertIsInstance(plans[0][1].changes[0], Delete)
+
+        # muck with plans
+        class MockProcessor(BaseProcessor):
+
+            def process_target_zone(self, zone, target):
+                zone = self._clone_zone(zone)
+                zone.add_record(record)
+                return zone
+
+            def process_plan(self, plans, sources, target):
+                # get rid of the change
+                plans.changes.pop(0)
+
+        mock = MockProcessor('mock')
+        plans, zone = manager._populate_and_plan('unit.tests.', [mock], [],
+                                                 targets)
+        # We planned a delete again, but this time removed it from the plan, so
+        # no plans
+        self.assertFalse(plans)
+
 
 class TestMainThreadExecutor(TestCase):
 

tests/test_octodns_provider_base.py

@@ -9,9 +9,10 @@ from logging import getLogger
 from six import text_type
 from unittest import TestCase
 
-from octodns.record import Create, Delete, Record, Update
+from octodns.processor.base import BaseProcessor
 from octodns.provider.base import BaseProvider
 from octodns.provider.plan import Plan, UnsafePlan
+from octodns.record import Create, Delete, Record, Update
 from octodns.zone import Zone
 
 
@@ -21,7 +22,7 @@ class HelperProvider(BaseProvider):
     SUPPORTS = set(('A',))
     id = 'test'
 
-    def __init__(self, extra_changes, apply_disabled=False,
+    def __init__(self, extra_changes=[], apply_disabled=False,
                  include_change_callback=None):
         self.__extra_changes = extra_changes
         self.apply_disabled = apply_disabled
@@ -43,6 +44,29 @@ class HelperProvider(BaseProvider):
         pass
 
 
+class TrickyProcessor(BaseProcessor):
+
+    def __init__(self, name, add_during_process_target_zone):
+        super(TrickyProcessor, self).__init__(name)
+        self.add_during_process_target_zone = add_during_process_target_zone
+        self.reset()
+
+    def reset(self):
+        self.existing = None
+        self.target = None
+
+    def process_target_zone(self, existing, target):
+        self.existing = existing
+        self.target = target
+
+        new = self._clone_zone(existing)
+        for record in existing.records:
+            new.add_record(record)
+        for record in self.add_during_process_target_zone:
+            new.add_record(record)
+        return new
+
+
 class TestBaseProvider(TestCase):
 
     def test_base_provider(self):
@@ -138,6 +162,45 @@ class TestBaseProvider(TestCase):
         self.assertTrue(plan)
         self.assertEquals(1, len(plan.changes))
 
+    def test_plan_with_processors(self):
+        zone = Zone('unit.tests.', [])
+
+        record = Record.new(zone, 'a', {
+            'ttl': 30,
+            'type': 'A',
+            'value': '1.2.3.4',
+        })
+        provider = HelperProvider()
+        # Processor that adds a record to the zone, which planning will then
+        # delete since it won't know anything about it
+        tricky = TrickyProcessor('tricky', [record])
+        plan = provider.plan(zone, processors=[tricky])
+        self.assertTrue(plan)
+        self.assertEquals(1, len(plan.changes))
+        self.assertIsInstance(plan.changes[0], Delete)
+        # Called processor stored its params
+        self.assertTrue(tricky.existing)
+        self.assertEquals(zone.name, tricky.existing.name)
+
+        # Chain of processors happen one after the other
+        other = Record.new(zone, 'b', {
+            'ttl': 30,
+            'type': 'A',
+            'value': '5.6.7.8',
+        })
+        # Another processor will add its record, thus 2 deletes
+        another = TrickyProcessor('tricky', [other])
+        plan = provider.plan(zone, processors=[tricky, another])
+        self.assertTrue(plan)
+        self.assertEquals(2, len(plan.changes))
+        self.assertIsInstance(plan.changes[0], Delete)
+        self.assertIsInstance(plan.changes[1], Delete)
+        # 2nd processor stored its params, and we'll see the record the
+        # first one added
+        self.assertTrue(another.existing)
+        self.assertEquals(zone.name, another.existing.name)
+        self.assertEquals(1, len(another.existing.records))
+
     def test_apply(self):
         ignored = Zone('unit.tests.', [])