Add OwnershipProcessor.should_replace to support ownership on alias zones

.changelog/5f9e8b0b6b7f4093b109a56d12239302.md

@@ -0,0 +1,4 @@
+---
+type: minor
+---
+Add OwnershipProcessor.should_replace to support ownership on alias zones

octodns/processor/ownership.py

@@ -14,16 +14,27 @@ from .base import BaseProcessor
 # and thus "own" them going forward.
 class OwnershipProcessor(BaseProcessor):
     def __init__(
-        self, name, txt_name='_owner', txt_value='*octodns*', txt_ttl=60
+        self,
+        name,
+        txt_name='_owner',
+        txt_value='*octodns*',
+        txt_ttl=60,
+        should_replace=False,
     ):
         super().__init__(name)
         self.txt_name = txt_name
         self.txt_value = txt_value
         self.txt_ttl = txt_ttl
         self._txt_values = [txt_value]
+        self.should_replace = should_replace
 
     def process_source_zone(self, desired, *args, **kwargs):
         for record in desired.records:
+            if self._is_ownership(record):
+                # don't apply ownership to existing ownership recorcs, most
+                # likely to see this in an alias zone that will be proccessed
+                # once as the original and a 2nd time as the alias
+                continue
             # Then create and add an ownership TXT for each of them
             record_name = record.name.replace('*', '_wildcard')
             if record.name:
@@ -37,7 +48,7 @@ class OwnershipProcessor(BaseProcessor):
             )
             # add these w/lenient to cover the case when the ownership record
             # for a NS delegation record should technically live in the subzone
-            desired.add_record(txt, lenient=True)
+            desired.add_record(txt, lenient=True, replace=self.should_replace)
 
         return desired
 

tests/test_octodns_processor_ownership.py

@@ -9,7 +9,7 @@ from helpers import PlannableProvider
 from octodns.processor.ownership import OwnershipProcessor
 from octodns.provider.plan import Plan
 from octodns.record import Delete, Record
-from octodns.zone import Zone
+from octodns.zone import DuplicateRecordException, Zone
 
 zone = Zone('unit.tests.', [])
 records = {}
@@ -150,3 +150,30 @@ class TestOwnershipProcessor(TestCase):
         self.assertEqual(1, len(plan.changes))
         plan = ownership.process_plan(plan)
         self.assertFalse(plan)
+
+    def test_should_replace(self):
+        ownership = OwnershipProcessor('ownership')
+        self.assertFalse(ownership.should_replace)
+
+        zone = Zone('unit.tests.', [])
+        record = Record.new(
+            zone, 'a', {'ttl': 30, 'type': 'A', 'value': '4.4.4.4'}
+        )
+        zone.add_record(record)
+
+        got = ownership.process_source_zone(zone.copy())
+        self.assertEqual(
+            ['_owner.a.a', 'a'], sorted([r.name for r in got.records])
+        )
+
+        # will fail w/a duplicate
+        with self.assertRaises(DuplicateRecordException):
+            ownership.process_source_zone(got.copy())
+
+        # enable should_replace, will replace instead of failing
+        ownership.should_replace = True
+        got = ownership.process_source_zone(got.copy())
+        # same expected result
+        self.assertEqual(
+            ['_owner.a.a', 'a'], sorted([r.name for r in got.records])
+        )