Merge branch 'master' of https://github.com/meghashyamps/octodns into NA-limitation-fix

CHANGELOG.md

@@ -1,3 +1,13 @@
+## v0.9.13 - 2021-..-.. -
+
+#### Noteworthy changes
+
+* Alpha support for Processors has been added. Processors allow for hooking
+  into the source, target, and planing process to make nearly arbitrary changes
+  to data. See the [octodns/processor/](/octodns/processor) directory for
+  examples. The change has been designed to have no impact on the process
+  unless the `processors` key is present in zone configs.
+
 ## v0.9.12 - 2021-04-30 - Enough time has passed
 
 #### Noteworthy changes

README.md

@@ -192,7 +192,7 @@ The above command pulled the existing data out of Route53 and placed the results
 
 | Provider | Requirements | Record Support | Dynamic | Notes |
 |--|--|--|--|--|
-| [AzureProvider](/octodns/provider/azuredns.py) | azure-identity, azure-mgmt-dns, azure-mgmt-trafficmanager | A, AAAA, CAA, CNAME, MX, NS, PTR, SRV, TXT | Alpha (CNAMEs only) | |
+| [AzureProvider](/octodns/provider/azuredns.py) | azure-identity, azure-mgmt-dns, azure-mgmt-trafficmanager | A, AAAA, CAA, CNAME, MX, NS, PTR, SRV, TXT | Alpha (CNAMEs and partial A/AAAA) | |
 | [Akamai](/octodns/provider/edgedns.py) | edgegrid-python | A, AAAA, CNAME, MX, NAPTR, NS, PTR, SPF, SRV, SSHFP, TXT | No | |
 | [CloudflareProvider](/octodns/provider/cloudflare.py) | | A, AAAA, ALIAS, CAA, CNAME, LOC, MX, NS, PTR, SPF, SRV, TXT | No | CAA tags restricted |
 | [ConstellixProvider](/octodns/provider/constellix.py) | | A, AAAA, ALIAS (ANAME), CAA, CNAME, MX, NS, PTR, SPF, SRV, TXT | No | CAA tags restricted |

octodns/manager.py

@@ -122,6 +122,25 @@ class Manager(object):
                 raise ManagerException('Incorrect provider config for {}'
                                        .format(provider_name))
 
+        self.processors = {}
+        for processor_name, processor_config in \
+                self.config.get('processors', {}).items():
+            try:
+                _class = processor_config.pop('class')
+            except KeyError:
+                self.log.exception('Invalid processor class')
+                raise ManagerException('Processor {} is missing class'
+                                       .format(processor_name))
+            _class = self._get_named_class('processor', _class)
+            kwargs = self._build_kwargs(processor_config)
+            try:
+                self.processors[processor_name] = _class(processor_name,
+                                                         **kwargs)
+            except TypeError:
+                self.log.exception('Invalid processor config')
+                raise ManagerException('Incorrect processor config for {}'
+                                       .format(processor_name))
+
         zone_tree = {}
         # sort by reversed strings so that parent zones always come first
         for name in sorted(self.config['zones'].keys(), key=lambda s: s[::-1]):
@@ -223,8 +242,8 @@ class Manager(object):
         self.log.debug('configured_sub_zones: subs=%s', sub_zone_names)
         return set(sub_zone_names)
 
-    def _populate_and_plan(self, zone_name, sources, targets, desired=None,
-                           lenient=False):
+    def _populate_and_plan(self, zone_name, processors, sources, targets,
+                           desired=None, lenient=False):
 
         self.log.debug('sync:   populating, zone=%s, lenient=%s',
                        zone_name, lenient)
@@ -237,7 +256,6 @@ class Manager(object):
             for _, records in desired._records.items():
                 for record in records:
                     zone.add_record(record.copy(zone=zone), lenient=lenient)
-
         else:
             for source in sources:
                 try:
@@ -245,10 +263,13 @@ 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)
 
+        for processor in processors:
+            zone = processor.process_source_zone(zone, sources=sources)
+
         self.log.debug('sync:   planning, zone=%s', zone_name)
         plans = []
 
@@ -260,7 +281,18 @@ class Manager(object):
                     'value': 'provider={}'.format(target.id)
                 })
                 zone.add_record(meta, replace=True)
-            plan = target.plan(zone)
+            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)
             if plan:
                 plans.append((target, plan))
 
@@ -319,6 +351,8 @@ class Manager(object):
                 raise ManagerException('Zone {} is missing targets'
                                        .format(zone_name))
 
+            processors = config.get('processors', [])
+
             if (eligible_sources and not
                     [s for s in sources if s in eligible_sources]):
                 self.log.info('sync:   no eligible sources, skipping')
@@ -336,6 +370,15 @@ class Manager(object):
 
             self.log.info('sync:   sources=%s -> targets=%s', sources, targets)
 
+            try:
+                collected = []
+                for processor in processors:
+                    collected.append(self.processors[processor])
+                processors = collected
+            except KeyError:
+                raise ManagerException('Zone {}, unknown processor: {}'
+                                       .format(zone_name, processor))
+
             try:
                 # rather than using a list comprehension, we break this loop
                 # out so that the `except` block below can reference the
@@ -362,8 +405,9 @@ class Manager(object):
                                        .format(zone_name, target))
 
             futures.append(self._executor.submit(self._populate_and_plan,
-                                                 zone_name, sources,
-                                                 targets, lenient=lenient))
+                                                 zone_name, processors,
+                                                 sources, targets,
+                                                 lenient=lenient))
 
         # Wait on all results and unpack/flatten the plans and store the
         # desired states in case we need them below
@@ -388,6 +432,7 @@ class Manager(object):
             futures.append(self._executor.submit(
                 self._populate_and_plan,
                 zone_name,
+                processors,
                 [],
                 [self.providers[t] for t in source_config['targets']],
                 desired=desired_config,
@@ -528,6 +573,16 @@ class Manager(object):
                 if isinstance(source, YamlProvider):
                     source.populate(zone)
 
+            # check that processors are in order if any are specified
+            processors = config.get('processors', [])
+            try:
+                # same as above, but for processors this time
+                for processor in processors:
+                    collected.append(self.processors[processor])
+            except KeyError:
+                raise ManagerException('Zone {}, unknown processor: {}'
+                                       .format(zone_name, processor))
+
     def get_zone(self, zone_name):
         if not zone_name[-1] == '.':
             raise ManagerException('Invalid zone name {}, missing ending dot'

octodns/processor/__init__.py

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

octodns/processor/base.py

@@ -0,0 +1,30 @@
+#
+#
+#
+
+from __future__ import absolute_import, division, print_function, \
+    unicode_literals
+
+from ..zone import Zone
+
+
+class BaseProcessor(object):
+
+    def __init__(self, name):
+        self.name = name
+
+    def _clone_zone(self, zone):
+        return Zone(zone.name, sub_zones=zone.sub_zones)
+
+    def process_source_zone(self, zone, sources):
+        # sources may be empty, as will be the case for aliased zones
+        return zone
+
+    def process_target_zone(self, zone, target):
+        return zone
+
+    def process_plan(self, plan, sources, target):
+        # plan may be None if no changes were detected up until now, the
+        # process may still create a plan.
+        # sources may be empty, as will be the case for aliased zones
+        return plan

octodns/processor/filter.py

@@ -0,0 +1,44 @@
+#
+#
+#
+
+from __future__ import absolute_import, division, print_function, \
+    unicode_literals
+
+from .base import BaseProcessor
+
+
+class TypeAllowlistFilter(BaseProcessor):
+
+    def __init__(self, name, allowlist):
+        super(TypeAllowlistFilter, self).__init__(name)
+        self.allowlist = set(allowlist)
+
+    def _process(self, zone, *args, **kwargs):
+        ret = self._clone_zone(zone)
+        for record in zone.records:
+            if record._type in self.allowlist:
+                ret.add_record(record)
+
+        return ret
+
+    process_source_zone = _process
+    process_target_zone = _process
+
+
+class TypeRejectlistFilter(BaseProcessor):
+
+    def __init__(self, name, rejectlist):
+        super(TypeRejectlistFilter, self).__init__(name)
+        self.rejectlist = set(rejectlist)
+
+    def _process(self, zone, *args, **kwargs):
+        ret = self._clone_zone(zone)
+        for record in zone.records:
+            if record._type not in self.rejectlist:
+                ret.add_record(record)
+
+        return ret
+
+    process_source_zone = _process
+    process_target_zone = _process

octodns/processor/ownership.py

@@ -0,0 +1,103 @@
+#
+#
+#
+
+from __future__ import absolute_import, division, print_function, \
+    unicode_literals
+
+from collections import defaultdict
+
+from ..provider.plan import Plan
+from ..record import Record
+
+from .base import BaseProcessor
+
+
+# Mark anything octoDNS is managing that way it can know it's safe to modify or
+# delete. We'll take ownership of existing records that we're told to manage
+# and thus "own" them going forward.
+class OwnershipProcessor(BaseProcessor):
+
+    def __init__(self, name, txt_name='_owner', txt_value='*octodns*'):
+        super(OwnershipProcessor, self).__init__(name)
+        self.txt_name = txt_name
+        self.txt_value = txt_value
+        self._txt_values = [txt_value]
+
+    def process_source_zone(self, zone, *args, **kwargs):
+        ret = self._clone_zone(zone)
+        for record in zone.records:
+            # Always copy over the source records
+            ret.add_record(record)
+            # Then create and add an ownership TXT for each of them
+            record_name = record.name.replace('*', '_wildcard')
+            if record.name:
+                name = '{}.{}.{}'.format(self.txt_name, record._type,
+                                         record_name)
+            else:
+                name = '{}.{}'.format(self.txt_name, record._type)
+            txt = Record.new(zone, name, {
+                'type': 'TXT',
+                'ttl': 60,
+                'value': self.txt_value,
+            })
+            ret.add_record(txt)
+
+        return ret
+
+    def _is_ownership(self, record):
+        return record._type == 'TXT' and \
+            record.name.startswith(self.txt_name) \
+            and record.values == self._txt_values
+
+    def process_plan(self, plan, *args, **kwargs):
+        if not plan:
+            # If we don't have any change there's nothing to do
+            return plan
+
+        # First find all the ownership info
+        owned = defaultdict(dict)
+        # We need to look for ownership in both the desired and existing
+        # states, many things will show up in both, but that's fine.
+        for record in list(plan.existing.records) + list(plan.desired.records):
+            if self._is_ownership(record):
+                pieces = record.name.split('.', 2)
+                if len(pieces) > 2:
+                    _, _type, name = pieces
+                    name = name.replace('_wildcard', '*')
+                else:
+                    _type = pieces[1]
+                    name = ''
+                owned[name][_type.upper()] = True
+
+        # Cases:
+        # - Configured in source
+        #   - We'll fully CRU/manage it adding ownership TXT,
+        #     thanks to process_source_zone, if needed
+        # - Not in source
+        #   - Has an ownership TXT - delete it & the ownership TXT
+        #   - Does not have an ownership TXT - don't delete it
+        # - Special records like octodns-meta
+        #   - Should be left alone and should not have ownerthis TXTs
+
+        filtered_changes = []
+        for change in plan.changes:
+            record = change.record
+
+            if not self._is_ownership(record) and \
+               record._type not in owned[record.name] and \
+               record.name != 'octodns-meta':
+                # It's not an ownership TXT, it's not owned, and it's not
+                # special we're going to ignore it
+                continue
+
+            # We own this record or owned it up until now so whatever the
+            # change is we should do
+            filtered_changes.append(change)
+
+        if plan.changes != filtered_changes:
+            return Plan(plan.existing, plan.desired, filtered_changes,
+                        plan.exists, plan.update_pcent_threshold,
+                        plan.delete_pcent_threshold)
+
+        return plan

octodns/provider/azuredns.py

@@ -125,6 +125,9 @@ class _AzureRecord(object):
         self.params['ttl'] = record.ttl
 
     def _params_for_A(self, data, key_name, azure_class):
+        if self._record.dynamic and self.traffic_manager:
+            return {'target_resource': self.traffic_manager}
+
         try:
             values = data['values']
         except KeyError:
@@ -132,6 +135,9 @@ class _AzureRecord(object):
         return {key_name: [azure_class(ipv4_address=v) for v in values]}
 
     def _params_for_AAAA(self, data, key_name, azure_class):
+        if self._record.dynamic and self.traffic_manager:
+            return {'target_resource': self.traffic_manager}
+
         try:
             values = data['values']
         except KeyError:
@@ -263,7 +269,10 @@ def _root_traffic_manager_name(record):
     # ATM names can only have letters, numbers and hyphens
     # replace dots with double hyphens to ensure unique mapping,
     # hoping that real life FQDNs won't have double hyphens
-    return record.fqdn[:-1].replace('.', '--')
+    name = record.fqdn[:-1].replace('.', '--')
+    if record._type != 'CNAME':
+        name += '-{}'.format(record._type)
+    return name
 
 
 def _rule_traffic_manager_name(pool, record):
@@ -290,6 +299,47 @@ def _get_monitor(record):
     return monitor
 
 
+def _check_valid_dynamic(record):
+    typ = record._type
+    dynamic = record.dynamic
+    if typ in ['A', 'AAAA']:
+        # A/AAAA records cannot be aliased to Traffic Managers that contain
+        # other nested Traffic Managers. Due to this limitation, A/AAAA
+        # dynamic records can do only one of geo-fencing, fallback and
+        # weighted RR. So let's validate that the record adheres to this
+        # limitation.
+        data = dynamic._data()
+        values = set(record.values)
+        pools = data['pools'].values()
+        seen_values = set()
+        rr = False
+        fallback = False
+        for pool in pools:
+            vals = pool['values']
+            if len(vals) > 1:
+                rr = True
+            pool_values = set(val['value'] for val in vals)
+            if pool.get('fallback'):
+                fallback = True
+            seen_values.update(pool_values)
+
+        if values != seen_values:
+            msg = ('{} {}: All pool values of A/AAAA dynamic records must be '
+                   'included in top-level \'values\'.')
+            raise AzureException(msg.format(record.fqdn, record._type))
+
+        geo = any(r.get('geos') for r in data['rules'])
+
+        if [rr, fallback, geo].count(True) > 1:
+            msg = ('{} {}: A/AAAA dynamic records must use at most one  of '
+                   'round-robin, fallback and geo-fencing')
+            raise AzureException(msg.format(record.fqdn, record._type))
+    elif typ != 'CNAME':
+        # dynamic records of unsupported type
+        msg = '{}: Dynamic records in Azure must be of type A/AAAA/CNAME'
+        raise AzureException(msg.format(record.fqdn))
+
+
 def _profile_is_match(have, desired):
     if have is None or desired is None:
         return False
@@ -608,9 +658,33 @@ class AzureProvider(BaseProvider):
                           lenient=lenient)
 
     def _data_for_A(self, azrecord):
+        if azrecord.a_records is None:
+            if azrecord.target_resource.id:
+                return self._data_for_dynamic(azrecord)
+            else:
+                # dynamic record alias is broken, return dummy value and apply
+                # will likely overwrite/fix it
+                self.log.warn('_data_for_A: Missing Traffic Manager '
+                              'alias for dynamic A record %s, forcing '
+                              're-link by setting an invalid value',
+                              azrecord.fqdn)
+                return {'values': ['255.255.255.255']}
+
         return {'values': [ar.ipv4_address for ar in azrecord.a_records]}
 
     def _data_for_AAAA(self, azrecord):
+        if azrecord.aaaa_records is None:
+            if azrecord.target_resource.id:
+                return self._data_for_dynamic(azrecord)
+            else:
+                # dynamic record alias is broken, return dummy value and apply
+                # will likely overwrite/fix it
+                self.log.warn('_data_for_AAAA: Missing Traffic Manager '
+                              'alias for dynamic AAAA record %s, forcing '
+                              're-link by setting an invalid value',
+                              azrecord.fqdn)
+                return {'values': ['::1']}
+
         return {'values': [ar.ipv6_address for ar in azrecord.aaaa_records]}
 
     def _data_for_CAA(self, azrecord):
@@ -667,6 +741,7 @@ class AzureProvider(BaseProvider):
         default = set()
         pools = defaultdict(lambda: {'fallback': None, 'values': []})
         rules = []
+        typ = _parse_azure_type(azrecord.type)
 
         # top level profile
         root_profile = self._get_tm_profile_by_id(azrecord.target_resource.id)
@@ -781,8 +856,10 @@ class AzureProvider(BaseProvider):
 
                 for pool_ep in endpoints:
                     val = pool_ep.target
+                    if typ == 'CNAME':
+                        val = _check_endswith_dot(val)
                     pool['values'].append({
-                        'value': _check_endswith_dot(val),
+                        'value': val,
                         'weight': pool_ep.weight or 1,
                     })
                     if pool_ep.name.endswith('--default--'):
@@ -805,24 +882,17 @@ class AzureProvider(BaseProvider):
                 'pools': pools,
                 'rules': rules,
             },
-            'value': _check_endswith_dot(default[0]),
         }
 
+        if typ == 'CNAME':
+            data['value'] = _check_endswith_dot(default[0])
+        else:
+            data['values'] = default
+
         return data
 
     def _extra_changes(self, existing, desired, changes):
-        changed = set()
-
-        # Abort if there are non-CNAME dynamic records
-        for change in changes:
-            record = change.record
-            changed.add(record)
-            typ = record._type
-            dynamic = getattr(record, 'dynamic', False)
-            if dynamic and typ != 'CNAME':
-                msg = '{}: Dynamic records in Azure must be of type CNAME'
-                msg = msg.format(record.fqdn)
-                raise AzureException(msg)
+        changed = set(c.record for c in changes)
 
         log = self.log.info
         seen_profiles = {}
@@ -832,12 +902,24 @@ class AzureProvider(BaseProvider):
                 # Already changed, or not dynamic, no need to check it
                 continue
 
+            # Abort if there are unsupported dynamic record configurations
+            _check_valid_dynamic(record)
+
             # let's walk through and show what will be changed even if
-            # the record is already be in list of changes
+            # the record is already in list of changes
             added = (record in changed)
 
             active = set()
             profiles = self._generate_traffic_managers(record)
+
+            # this should not happen with above check, check again here to
+            # prevent undesired changes
+            if record._type in ['A', 'AAAA'] and len(profiles) > 1:
+                msg = ('Unknown error: {} {} needs more than 1 Traffic '
+                       'Managers which is not supported for A/AAAA dynamic '
+                       'records').format(record.fqdn, record._type)
+                raise AzureException(msg)
+
             for profile in profiles:
                 name = profile.name
                 if name in seen_profiles:
@@ -871,9 +953,9 @@ class AzureProvider(BaseProvider):
     def _generate_tm_profile(self, routing, endpoints, record, label=None):
         # figure out profile name and Traffic Manager FQDN
         name = _root_traffic_manager_name(record)
-        if routing == 'Weighted':
+        if routing == 'Weighted' and label:
             name = _pool_traffic_manager_name(label, record)
-        elif routing == 'Priority':
+        elif routing == 'Priority' and label:
             name = _rule_traffic_manager_name(label, record)
 
         # set appropriate endpoint types
@@ -895,7 +977,7 @@ class AzureProvider(BaseProvider):
             name=name,
             traffic_routing_method=routing,
             dns_config=DnsConfig(
-                relative_name=name,
+                relative_name=name.lower(),
                 ttl=record.ttl,
             ),
             monitor_config=_get_monitor(record),
@@ -906,7 +988,7 @@ class AzureProvider(BaseProvider):
     def _convert_tm_to_root(self, profile, record):
         profile.name = _root_traffic_manager_name(record)
         profile.id = self._profile_name_to_id(profile.name)
-        profile.dns_config.relative_name = profile.name
+        profile.dns_config.relative_name = profile.name.lower()
 
         return profile
 
@@ -914,8 +996,12 @@ class AzureProvider(BaseProvider):
         traffic_managers = []
         pools = record.dynamic.pools
         rules = record.dynamic.rules
+        typ = record._type
 
-        default = record.value[:-1]
+        if typ == 'CNAME':
+            defaults = [record.value[:-1]]
+        else:
+            defaults = record.values
         profile = self._generate_tm_profile
 
         # a pool can be re-used only with a world pool, record the pool
@@ -977,9 +1063,10 @@ class AzureProvider(BaseProvider):
                         for val in pool['values']:
                             target = val['value']
                             # strip trailing dot from CNAME value
-                            target = target[:-1]
+                            if typ == 'CNAME':
+                                target = target[:-1]
                             ep_name = '{}--{}'.format(pool_name, target)
-                            if target == default:
+                            if target in defaults:
                                 # mark default
                                 ep_name += '--default--'
                                 default_seen = True
@@ -1003,14 +1090,16 @@ class AzureProvider(BaseProvider):
                     # Skip Weighted profile hop for single-value pool
                     # append its value as an external endpoint to fallback
                     # rule profile
-                    target = pool['values'][0]['value'][:-1]
+                    target = pool['values'][0]['value']
+                    if typ == 'CNAME':
+                        target = target[:-1]
                     ep_name = pool_name
-                    if target == default:
+                    if target in defaults:
                         # mark default
                         ep_name += '--default--'
                         default_seen = True
                     rule_endpoints.append(Endpoint(
-                        name=pool_name,
+                        name=ep_name,
                         target=target,
                         priority=priority,
                     ))
@@ -1023,7 +1112,7 @@ class AzureProvider(BaseProvider):
             if not default_seen:
                 rule_endpoints.append(Endpoint(
                     name='--default--',
-                    target=default,
+                    target=defaults[0],
                     priority=priority,
                 ))
 
@@ -1053,7 +1142,7 @@ class AzureProvider(BaseProvider):
                 else:
                     # just add the value of single-value pool
                     geo_endpoints.append(Endpoint(
-                        name=rule_ep.name + '--default--',
+                        name=rule_ep.name,
                         target=rule_ep.target,
                         geo_mapping=geos,
                     ))

octodns/provider/base.py

@@ -44,7 +44,7 @@ class BaseProvider(BaseSource):
         '''
         return []
 
-    def plan(self, desired):
+    def plan(self, desired, processors=[]):
         self.log.info('plan: desired=%s', desired.name)
 
         existing = Zone(desired.name, desired.sub_zones)
@@ -55,6 +55,9 @@ class BaseProvider(BaseSource):
             self.log.warn('Provider %s used in target mode did not return '
                           'exists', self.id)
 
+        for processor in processors:
+            existing = processor.process_target_zone(existing, target=self)
+
         # compute the changes at the zone/record level
         changes = existing.changes(desired, self)
 

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/config/unknown-processor.yaml

@@ -0,0 +1,17 @@
+manager:
+  max_workers: 2
+providers:
+  in:
+    class: octodns.provider.yaml.YamlProvider
+    directory: tests/config
+  dump:
+    class: octodns.provider.yaml.YamlProvider
+    directory: env/YAML_TMP_DIR
+zones:
+  unit.tests.:
+    sources:
+      - in
+    processors:
+      - missing
+    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')
 
@@ -338,6 +339,11 @@ class TestManager(TestCase):
         Manager(get_config_filename('simple-alias-zone.yaml')) \
             .validate_configs()
 
+        with self.assertRaises(ManagerException) as ctx:
+            Manager(get_config_filename('unknown-processor.yaml')) \
+                .validate_configs()
+        self.assertTrue('unknown processor' in text_type(ctx.exception))
+
     def test_get_zone(self):
         Manager(get_config_filename('simple.yaml')).get_zone('unit.tests.')
 
@@ -358,20 +364,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()], [])
+            manager._populate_and_plan('unit.tests.', [], [NoLenient()], [])
 
-            class NoZone(SimpleProvider):
+            class OtherType(SimpleProvider):
 
-                def populate(self, lenient=False):
-                    pass
+                def populate(self, zone, lenient=False):
+                    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))
+
+    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) 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 +425,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_processor_filter.py

@@ -0,0 +1,90 @@
+#
+#
+#
+
+from __future__ import absolute_import, division, print_function, \
+    unicode_literals
+
+from unittest import TestCase
+
+from octodns.processor.filter import TypeAllowlistFilter, TypeRejectlistFilter
+from octodns.record import Record
+from octodns.zone import Zone
+
+zone = Zone('unit.tests.', [])
+for record in [
+    Record.new(zone, 'a', {
+        'ttl': 30,
+        'type': 'A',
+        'value': '1.2.3.4',
+    }),
+    Record.new(zone, 'aaaa', {
+        'ttl': 30,
+        'type': 'AAAA',
+        'value': '::1',
+    }),
+    Record.new(zone, 'txt', {
+        'ttl': 30,
+        'type': 'TXT',
+        'value': 'Hello World!',
+    }),
+    Record.new(zone, 'a2', {
+        'ttl': 30,
+        'type': 'A',
+        'value': '2.3.4.5',
+    }),
+    Record.new(zone, 'txt2', {
+        'ttl': 30,
+        'type': 'TXT',
+        'value': 'That will do',
+    }),
+]:
+    zone.add_record(record)
+
+
+class TestTypeAllowListFilter(TestCase):
+
+    def test_basics(self):
+        filter_a = TypeAllowlistFilter('only-a', set(('A')))
+
+        got = filter_a.process_source_zone(zone)
+        self.assertEquals(['a', 'a2'], sorted([r.name for r in got.records]))
+
+        filter_aaaa = TypeAllowlistFilter('only-aaaa', ('AAAA',))
+        got = filter_aaaa.process_source_zone(zone)
+        self.assertEquals(['aaaa'], sorted([r.name for r in got.records]))
+
+        filter_txt = TypeAllowlistFilter('only-txt', ['TXT'])
+        got = filter_txt.process_target_zone(zone)
+        self.assertEquals(['txt', 'txt2'],
+                          sorted([r.name for r in got.records]))
+
+        filter_a_aaaa = TypeAllowlistFilter('only-aaaa', set(('A', 'AAAA')))
+        got = filter_a_aaaa.process_target_zone(zone)
+        self.assertEquals(['a', 'a2', 'aaaa'],
+                          sorted([r.name for r in got.records]))
+
+
+class TestTypeRejectListFilter(TestCase):
+
+    def test_basics(self):
+        filter_a = TypeRejectlistFilter('not-a', set(('A')))
+
+        got = filter_a.process_source_zone(zone)
+        self.assertEquals(['aaaa', 'txt', 'txt2'],
+                          sorted([r.name for r in got.records]))
+
+        filter_aaaa = TypeRejectlistFilter('not-aaaa', ('AAAA',))
+        got = filter_aaaa.process_source_zone(zone)
+        self.assertEquals(['a', 'a2', 'txt', 'txt2'],
+                          sorted([r.name for r in got.records]))
+
+        filter_txt = TypeRejectlistFilter('not-txt', ['TXT'])
+        got = filter_txt.process_target_zone(zone)
+        self.assertEquals(['a', 'a2', 'aaaa'],
+                          sorted([r.name for r in got.records]))
+
+        filter_a_aaaa = TypeRejectlistFilter('not-a-aaaa', set(('A', 'AAAA')))
+        got = filter_a_aaaa.process_target_zone(zone)
+        self.assertEquals(['txt', 'txt2'],
+                          sorted([r.name for r in got.records]))

tests/test_octodns_processor_ownership.py

@@ -0,0 +1,146 @@
+#
+#
+#
+
+from __future__ import absolute_import, division, print_function, \
+    unicode_literals
+
+from unittest import TestCase
+
+from octodns.processor.ownership import OwnershipProcessor
+from octodns.record import Delete, Record
+from octodns.zone import Zone
+
+from helpers import PlannableProvider
+
+
+zone = Zone('unit.tests.', [])
+records = {}
+for record in [
+    Record.new(zone, '', {
+        'ttl': 30,
+        'type': 'A',
+        'values': [
+            '1.2.3.4',
+            '5.6.7.8',
+        ],
+    }),
+    Record.new(zone, 'the-a', {
+        'ttl': 30,
+        'type': 'A',
+        'value': '1.2.3.4',
+    }),
+    Record.new(zone, 'the-aaaa', {
+        'ttl': 30,
+        'type': 'AAAA',
+        'value': '::1',
+    }),
+    Record.new(zone, 'the-txt', {
+        'ttl': 30,
+        'type': 'TXT',
+        'value': 'Hello World!',
+    }),
+    Record.new(zone, '*', {
+        'ttl': 30,
+        'type': 'A',
+        'value': '4.3.2.1',
+    }),
+]:
+    records[record.name] = record
+    zone.add_record(record)
+
+
+class TestOwnershipProcessor(TestCase):
+
+    def test_process_source_zone(self):
+        ownership = OwnershipProcessor('ownership')
+
+        got = ownership.process_source_zone(zone)
+        self.assertEquals([
+            '',
+            '*',
+            '_owner.a',
+            '_owner.a._wildcard',
+            '_owner.a.the-a',
+            '_owner.aaaa.the-aaaa',
+            '_owner.txt.the-txt',
+            'the-a',
+            'the-aaaa',
+            'the-txt',
+        ], sorted([r.name for r in got.records]))
+
+        found = False
+        for record in got.records:
+            if record.name.startswith(ownership.txt_name):
+                self.assertEquals([ownership.txt_value], record.values)
+                # test _is_ownership while we're in here
+                self.assertTrue(ownership._is_ownership(record))
+                found = True
+            else:
+                self.assertFalse(ownership._is_ownership(record))
+        self.assertTrue(found)
+
+    def test_process_plan(self):
+        ownership = OwnershipProcessor('ownership')
+        provider = PlannableProvider('helper')
+
+        # No plan, is a quick noop
+        self.assertFalse(ownership.process_plan(None))
+
+        # Nothing exists create both records and ownership
+        ownership_added = ownership.process_source_zone(zone)
+        plan = provider.plan(ownership_added)
+        self.assertTrue(plan)
+        # Double the number of records
+        self.assertEquals(len(records) * 2, len(plan.changes))
+        # Now process the plan, shouldn't make any changes, we're creating
+        # everything
+        got = ownership.process_plan(plan)
+        self.assertTrue(got)
+        self.assertEquals(len(records) * 2, len(got.changes))
+
+        # Something extra exists and doesn't have ownership TXT, leave it
+        # alone, we don't own it.
+        extra_a = Record.new(zone, 'extra-a', {
+            'ttl': 30,
+            'type': 'A',
+            'value': '4.4.4.4',
+        })
+        plan.existing.add_record(extra_a)
+        # If we'd done a "real" plan we'd have a delete for the extra thing.
+        plan.changes.append(Delete(extra_a))
+        # Process the plan, shouldn't make any changes since the extra bit is
+        # something we don't own
+        got = ownership.process_plan(plan)
+        self.assertTrue(got)
+        self.assertEquals(len(records) * 2, len(got.changes))
+
+        # Something extra exists and does have an ownership record so we will
+        # delete it...
+        copy = Zone('unit.tests.', [])
+        for record in records.values():
+            if record.name != 'the-a':
+                copy.add_record(record)
+        # New ownership, without the `the-a`
+        ownership_added = ownership.process_source_zone(copy)
+        self.assertEquals(len(records) * 2 - 2, len(ownership_added.records))
+        plan = provider.plan(ownership_added)
+        # Fake the extra existing by adding the record, its ownership, and the
+        # two delete changes.
+        the_a = records['the-a']
+        plan.existing.add_record(the_a)
+        name = '{}.a.the-a'.format(ownership.txt_name)
+        the_a_ownership = Record.new(zone, name, {
+            'ttl': 30,
+            'type': 'TXT',
+            'value': ownership.txt_value,
+        })
+        plan.existing.add_record(the_a_ownership)
+        plan.changes.append(Delete(the_a))
+        plan.changes.append(Delete(the_a_ownership))
+        # Finally process the plan, should be a noop and we should get the same
+        # plan out, meaning the planned deletes were allowed to happen.
+        got = ownership.process_plan(plan)
+        self.assertTrue(got)
+        self.assertEquals(plan, got)
+        self.assertEquals(len(plan.changes), len(got.changes))

tests/test_octodns_provider_azuredns.py

@@ -4,6 +4,7 @@
 
 from __future__ import absolute_import, division, print_function, \
     unicode_literals
+from logging import debug
 
 from octodns.record import Create, Update, Delete, Record
 from octodns.provider.azuredns import _AzureRecord, AzureProvider, \
@@ -883,12 +884,13 @@ class TestAzureDnsProvider(TestCase):
 
         # test simple records produce no extra changes
         desired = Zone(name=existing.name, sub_zones=[])
-        desired.add_record(Record.new(desired, 'simple', data={
+        simple = Record.new(desired, 'simple', data={
             'type': record._type,
             'ttl': record.ttl,
             'value': record.value,
-        }))
-        extra = provider._extra_changes(desired, desired, [])
+        })
+        desired.add_record(simple)
+        extra = provider._extra_changes(desired, desired, [Create(simple)])
         self.assertEqual(len(extra), 0)
 
         # test an unchanged dynamic record produces no extra changes
@@ -952,28 +954,28 @@ class TestAzureDnsProvider(TestCase):
         self.assertIsInstance(extra, Update)
         self.assertEqual(extra.new, update_dynamic)
 
-        # test non-CNAME dynamic record throws exception
-        a_dynamic = Record.new(desired, record.name + '3', data={
-            'type': 'A',
+        # test dynamic record of unsupported type throws exception
+        unsupported_dynamic = Record.new(desired, record.name + '3', data={
+            'type': 'DNAME',
             'ttl': record.ttl,
-            'values': ['1.1.1.1'],
+            'value': 'default.unit.tests.',
             'dynamic': {
                 'pools': {
-                    'one': {'values': [{'value': '2.2.2.2'}]},
+                    'one': {'values': [{'value': 'one.unit.tests.'}]},
                 },
                 'rules': [
                     {'pool': 'one'},
                 ],
             },
         })
-        desired.add_record(a_dynamic)
-        changes.append(Create(a_dynamic))
+        desired.add_record(unsupported_dynamic)
+        changes = [Create(unsupported_dynamic)]
         with self.assertRaises(AzureException) as ctx:
             provider._extra_changes(existing, desired, changes)
             self.assertTrue(text_type(ctx).endswith(
                 'must be of type CNAME'
             ))
-        desired._remove_record(a_dynamic)
+        desired._remove_record(unsupported_dynamic)
 
         # test colliding ATM names throws exception
         record1 = Record.new(desired, 'sub.www', data={
@@ -997,6 +999,129 @@ class TestAzureDnsProvider(TestCase):
                 'Collision in Traffic Manager'
             ))
 
+    def test_extra_changes_invalid_dynamic_A(self):
+        provider = self._get_provider()
+
+        # too many test case combinations, here's a method to generate them
+        def record_data(all_values=True, rr=True, fallback=True, geo=True):
+            data = {
+                'type': 'A',
+                'ttl': 60,
+                'dynamic': {
+                    'pools': {
+                        'one': {
+                            'values': [
+                                {'value': '11.11.11.11'},
+                                {'value': '12.12.12.12'},
+                            ],
+                            'fallback': 'two',
+                        },
+                        'two': {
+                            'values': [
+                                {'value': '2.2.2.2'},
+                            ],
+                        },
+                    },
+                    'rules': [
+                        {'geos': ['EU'], 'pool': 'two'},
+                        {'pool': 'one'},
+                    ],
+                }
+            }
+            dynamic = data['dynamic']
+            if not rr:
+                dynamic['pools']['one']['values'].pop()
+            if not fallback:
+                dynamic['pools']['one'].pop('fallback')
+            if not geo:
+                rule = dynamic['rules'].pop(0)
+                if not fallback:
+                    dynamic['pools'].pop(rule['pool'])
+            # put all pool values in default
+            data['values'] = [
+                v['value']
+                for p in dynamic['pools'].values()
+                for v in p['values']
+            ]
+            if not all_values:
+                rm = list(dynamic['pools'].values())[0]['values'][0]['value']
+                data['values'].remove(rm)
+            return data
+
+        # test all combinations
+        values = [True, False]
+        combos = [
+            [arg1, arg2, arg3, arg4]
+            for arg1 in values
+            for arg2 in values
+            for arg3 in values
+            for arg4 in values
+        ]
+        for all_values, rr, fallback, geo in combos:
+            args = [all_values, rr, fallback, geo]
+
+            if not any(args):
+                # all False, invalid use-case
+                continue
+
+            debug('[all_values, rr, fallback, geo] = %s', args)
+            data = record_data(*args)
+            desired = Zone(name=zone.name, sub_zones=[])
+            record = Record.new(desired, 'foo', data)
+            desired.add_record(record)
+
+            features = args[1:]
+            if all_values and features.count(True) <= 1:
+                # assert does not raise exception
+                provider._extra_changes(zone, desired, [Create(record)])
+                continue
+
+            with self.assertRaises(AzureException) as ctx:
+                msg = text_type(ctx)
+                provider._extra_changes(zone, desired, [Create(record)])
+                if not all_values:
+                    self.assertTrue('included in top-level \'values\'' in msg)
+                else:
+                    self.assertTrue('at most one of' in msg)
+
+    @patch('octodns.provider.azuredns._check_valid_dynamic')
+    def test_extra_changes_dynamic_A_multiple_profiles(self, mock_cvd):
+        provider = self._get_provider()
+
+        # bypass validity check to trigger mutliple-profiles check
+        mock_cvd.return_value = True
+
+        desired = Zone(name=zone.name, sub_zones=[])
+        record = Record.new(desired, 'foo', {
+            'type': 'A',
+            'ttl': 60,
+            'values': ['11.11.11.11', '12.12.12.12', '2.2.2.2'],
+            'dynamic': {
+                'pools': {
+                    'one': {
+                        'values': [
+                            {'value': '11.11.11.11'},
+                            {'value': '12.12.12.12'},
+                        ],
+                        'fallback': 'two',
+                    },
+                    'two': {
+                        'values': [
+                            {'value': '2.2.2.2'},
+                        ],
+                    },
+                },
+                'rules': [
+                    {'geos': ['EU'], 'pool': 'two'},
+                    {'pool': 'one'},
+                ],
+            }
+        })
+        desired.add_record(record)
+        with self.assertRaises(AzureException) as ctx:
+            provider._extra_changes(zone, desired, [Create(record)])
+            self.assertTrue('more than 1 Traffic Managers' in text_type(ctx))
+
     def test_generate_tm_profile(self):
         provider, zone, record = self._get_dynamic_package()
         profile_gen = provider._generate_tm_profile
@@ -1573,6 +1698,300 @@ class TestAzureDnsProvider(TestCase):
         record2 = provider._populate_record(zone, azrecord)
         self.assertEqual(record2.dynamic._data(), record.dynamic._data())
 
+    def test_dynamic_A_geo(self):
+        provider = self._get_provider()
+        external = 'Microsoft.Network/trafficManagerProfiles/externalEndpoints'
+
+        record = Record.new(zone, 'foo', data={
+            'type': 'A',
+            'ttl': 60,
+            'values': ['1.1.1.1', '2.2.2.2', '3.3.3.3'],
+            'dynamic': {
+                'pools': {
+                    'one': {
+                        'values': [
+                            {'value': '1.1.1.1'},
+                        ],
+                    },
+                    'two': {
+                        'values': [
+                            {'value': '2.2.2.2'},
+                        ],
+                    },
+                    'three': {
+                        'values': [
+                            {'value': '3.3.3.3'},
+                        ],
+                    },
+                },
+                'rules': [
+                    {'geos': ['AS'], 'pool': 'one'},
+                    {'geos': ['AF'], 'pool': 'two'},
+                    {'pool': 'three'},
+                ],
+            }
+        })
+
+        # test that extra_changes doesn't complain
+        changes = [Create(record)]
+        provider._extra_changes(zone, zone, changes)
+
+        profiles = provider._generate_traffic_managers(record)
+
+        self.assertEqual(len(profiles), 1)
+        self.assertTrue(_profile_is_match(profiles[0], Profile(
+            name='foo--unit--tests-A',
+            traffic_routing_method='Geographic',
+            dns_config=DnsConfig(
+                relative_name='foo--unit--tests-a', ttl=record.ttl),
+            monitor_config=_get_monitor(record),
+            endpoints=[
+                Endpoint(
+                    name='one--default--',
+                    type=external,
+                    target='1.1.1.1',
+                    geo_mapping=['GEO-AS'],
+                ),
+                Endpoint(
+                    name='two--default--',
+                    type=external,
+                    target='2.2.2.2',
+                    geo_mapping=['GEO-AF'],
+                ),
+                Endpoint(
+                    name='three--default--',
+                    type=external,
+                    target='3.3.3.3',
+                    geo_mapping=['WORLD'],
+                ),
+            ],
+        )))
+
+        # test that the record and ATM profile gets created
+        tm_sync = provider._tm_client.profiles.create_or_update
+        create = provider._dns_client.record_sets.create_or_update
+        provider._apply_Create(changes[0])
+        # A dynamic record can only have 1 profile
+        tm_sync.assert_called_once()
+        create.assert_called_once()
+
+        # test broken alias
+        azrecord = RecordSet(
+            ttl=60, target_resource=SubResource(id=None))
+        azrecord.name = record.name or '@'
+        azrecord.type = 'Microsoft.Network/dnszones/{}'.format(record._type)
+        record2 = provider._populate_record(zone, azrecord)
+        self.assertEqual(record2.values, ['255.255.255.255'])
+
+        # test that same record gets populated back from traffic managers
+        tm_list = provider._tm_client.profiles.list_by_resource_group
+        tm_list.return_value = profiles
+        azrecord = RecordSet(
+            ttl=60,
+            target_resource=SubResource(id=profiles[-1].id),
+        )
+        azrecord.name = record.name or '@'
+        azrecord.type = 'Microsoft.Network/dnszones/{}'.format(record._type)
+        record2 = provider._populate_record(zone, azrecord)
+        self.assertEqual(record2.dynamic._data(), record.dynamic._data())
+
+    def test_dynamic_A_fallback(self):
+        provider = self._get_provider()
+        external = 'Microsoft.Network/trafficManagerProfiles/externalEndpoints'
+
+        record = Record.new(zone, 'foo', data={
+            'type': 'A',
+            'ttl': 60,
+            'values': ['8.8.8.8'],
+            'dynamic': {
+                'pools': {
+                    'one': {
+                        'values': [
+                            {'value': '1.1.1.1'},
+                        ],
+                        'fallback': 'two',
+                    },
+                    'two': {
+                        'values': [
+                            {'value': '2.2.2.2'},
+                        ],
+                    },
+                },
+                'rules': [
+                    {'pool': 'one'},
+                ],
+            }
+        })
+        profiles = provider._generate_traffic_managers(record)
+
+        self.assertEqual(len(profiles), 1)
+        self.assertTrue(_profile_is_match(profiles[0], Profile(
+            name='foo--unit--tests-A',
+            traffic_routing_method='Priority',
+            dns_config=DnsConfig(
+                relative_name='foo--unit--tests-a', ttl=record.ttl),
+            monitor_config=_get_monitor(record),
+            endpoints=[
+                Endpoint(
+                    name='one',
+                    type=external,
+                    target='1.1.1.1',
+                    priority=1,
+                ),
+                Endpoint(
+                    name='two',
+                    type=external,
+                    target='2.2.2.2',
+                    priority=2,
+                ),
+                Endpoint(
+                    name='--default--',
+                    type=external,
+                    target='8.8.8.8',
+                    priority=3,
+                ),
+            ],
+        )))
+
+        # test that same record gets populated back from traffic managers
+        tm_list = provider._tm_client.profiles.list_by_resource_group
+        tm_list.return_value = profiles
+        azrecord = RecordSet(
+            ttl=60,
+            target_resource=SubResource(id=profiles[-1].id),
+        )
+        azrecord.name = record.name or '@'
+        azrecord.type = 'Microsoft.Network/dnszones/{}'.format(record._type)
+        record2 = provider._populate_record(zone, azrecord)
+        self.assertEqual(record2.dynamic._data(), record.dynamic._data())
+
+    def test_dynamic_A_weighted_rr(self):
+        provider = self._get_provider()
+        external = 'Microsoft.Network/trafficManagerProfiles/externalEndpoints'
+
+        record = Record.new(zone, 'foo', data={
+            'type': 'A',
+            'ttl': 60,
+            'values': ['1.1.1.1', '8.8.8.8'],
+            'dynamic': {
+                'pools': {
+                    'one': {
+                        'values': [
+                            {'value': '1.1.1.1', 'weight': 11},
+                            {'value': '8.8.8.8', 'weight': 8},
+                        ],
+                    },
+                },
+                'rules': [
+                    {'pool': 'one'},
+                ],
+            }
+        })
+        profiles = provider._generate_traffic_managers(record)
+
+        self.assertEqual(len(profiles), 1)
+        self.assertTrue(_profile_is_match(profiles[0], Profile(
+            name='foo--unit--tests-A',
+            traffic_routing_method='Weighted',
+            dns_config=DnsConfig(
+                relative_name='foo--unit--tests-a', ttl=record.ttl),
+            monitor_config=_get_monitor(record),
+            endpoints=[
+                Endpoint(
+                    name='one--1.1.1.1--default--',
+                    type=external,
+                    target='1.1.1.1',
+                    weight=11,
+                ),
+                Endpoint(
+                    name='one--8.8.8.8--default--',
+                    type=external,
+                    target='8.8.8.8',
+                    weight=8,
+                ),
+            ],
+        )))
+
+        # test that same record gets populated back from traffic managers
+        tm_list = provider._tm_client.profiles.list_by_resource_group
+        tm_list.return_value = profiles
+        azrecord = RecordSet(
+            ttl=60,
+            target_resource=SubResource(id=profiles[-1].id),
+        )
+        azrecord.name = record.name or '@'
+        azrecord.type = 'Microsoft.Network/dnszones/{}'.format(record._type)
+        record2 = provider._populate_record(zone, azrecord)
+        self.assertEqual(record2.dynamic._data(), record.dynamic._data())
+
+    def test_dynamic_AAAA(self):
+        provider = self._get_provider()
+        external = 'Microsoft.Network/trafficManagerProfiles/externalEndpoints'
+
+        record = Record.new(zone, 'foo', data={
+            'type': 'AAAA',
+            'ttl': 60,
+            'values': ['1::1'],
+            'dynamic': {
+                'pools': {
+                    'one': {
+                        'values': [
+                            {'value': '1::1'},
+                        ],
+                    },
+                },
+                'rules': [
+                    {'pool': 'one'},
+                ],
+            }
+        })
+        profiles = provider._generate_traffic_managers(record)
+
+        self.assertEqual(len(profiles), 1)
+        self.assertTrue(_profile_is_match(profiles[0], Profile(
+            name='foo--unit--tests-AAAA',
+            traffic_routing_method='Geographic',
+            dns_config=DnsConfig(
+                relative_name='foo--unit--tests-aaaa', ttl=record.ttl),
+            monitor_config=_get_monitor(record),
+            endpoints=[
+                Endpoint(
+                    name='one--default--',
+                    type=external,
+                    target='1::1',
+                    geo_mapping=['WORLD'],
+                ),
+            ],
+        )))
+
+        # test that the record and ATM profile gets created
+        tm_sync = provider._tm_client.profiles.create_or_update
+        create = provider._dns_client.record_sets.create_or_update
+        provider._apply_Create(Create(record))
+        # A dynamic record can only have 1 profile
+        tm_sync.assert_called_once()
+        create.assert_called_once()
+
+        # test broken alias
+        azrecord = RecordSet(
+            ttl=60, target_resource=SubResource(id=None))
+        azrecord.name = record.name or '@'
+        azrecord.type = 'Microsoft.Network/dnszones/{}'.format(record._type)
+        record2 = provider._populate_record(zone, azrecord)
+        self.assertEqual(record2.values, ['::1'])
+
+        # test that same record gets populated back from traffic managers
+        tm_list = provider._tm_client.profiles.list_by_resource_group
+        tm_list.return_value = profiles
+        azrecord = RecordSet(
+            ttl=60,
+            target_resource=SubResource(id=profiles[-1].id),
+        )
+        azrecord.name = record.name or '@'
+        azrecord.type = 'Microsoft.Network/dnszones/{}'.format(record._type)
+        record2 = provider._populate_record(zone, azrecord)
+        self.assertEqual(record2.dynamic._data(), record.dynamic._data())
+
     def test_sync_traffic_managers(self):
         provider, zone, record = self._get_dynamic_package()
         provider._populate_traffic_managers()

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.', [])