Merge pull request #955 from octodns/ignore-root-ns-filter

Add IgnoreRootNsFilter w/tests

CHANGELOG.md

@@ -1,3 +1,11 @@
+## v1.0.0.rc0 - 2022-??-?? - First of the ones
+
+#### Noteworthy changes
+
+#### Stuff
+
+* Added simple IgnoreRootNsFilter
+
 ## v0.9.21 - 2022-10-16 - Last of the oughts
 
 * Shim AxfrSource and ZoneFileSource post extraction into

octodns/processor/filter.py

@@ -188,3 +188,33 @@ class NameRejectlistFilter(_NameBaseFilter):
 
     process_source_zone = _process
     process_target_zone = _process
+
+
+class IgnoreRootNsFilter(BaseProcessor):
+    '''Do not manage Root NS Records.
+
+    Example usage:
+
+    processors:
+      no-root-ns:
+        class: octodns.processor.filter.IgnoreRootNsFilter
+
+    zones:
+      exxampled.com.:
+        sources:
+          - config
+        processors:
+          - no-root-ns
+        targets:
+          - ns1
+    '''
+
+    def _process(self, zone, *args, **kwargs):
+        for record in zone.records:
+            if record._type == 'NS' and not record.name:
+                zone.remove_record(record)
+
+        return zone
+
+    process_source_zone = _process
+    process_target_zone = _process

tests/test_octodns_processor_filter.py

@@ -5,6 +5,7 @@
 from unittest import TestCase
 
 from octodns.processor.filter import (
+    IgnoreRootNsFilter,
     NameAllowlistFilter,
     NameRejectlistFilter,
     TypeAllowlistFilter,
@@ -154,3 +155,28 @@ class TestNameRejectListFilter(TestCase):
         self.assertEqual(
             ['doesnt', 'matches'], sorted([r.name for r in filtered.records])
         )
+
+
+class TestIgnoreRootNsFilter(TestCase):
+    zone = Zone('unit.tests.', [])
+    root = Record.new(
+        zone, '', {'type': 'NS', 'ttl': 42, 'value': 'ns1.unit.tests.'}
+    )
+    zone.add_record(root)
+    not_root = Record.new(
+        zone, 'sub', {'type': 'NS', 'ttl': 43, 'value': 'ns2.unit.tests.'}
+    )
+    zone.add_record(not_root)
+    not_ns = Record.new(zone, '', {'type': 'A', 'ttl': 42, 'value': '3.4.5.6'})
+    zone.add_record(not_ns)
+
+    def test_filtering(self):
+        proc = IgnoreRootNsFilter('no-root')
+
+        self.assertEqual(3, len(self.zone.records))
+        filtered = proc.process_source_zone(self.zone.copy())
+        self.assertEqual(2, len(filtered.records))
+        self.assertEqual(
+            [('A', ''), ('NS', 'sub')],
+            sorted([(r._type, r.name) for r in filtered.records]),
+        )