diff --git a/CHANGELOG.md b/CHANGELOG.md index d480311..4a463ce 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ * Add YamlProvider.order_mode to allow picking between natural (human) the default when enforce_order=True and simple `sort`. * Fix type-o in _build_kwargs handler notification +* Add support for configuring OwnershipProcessor TXT record's TTL ## v1.10.0 - 2024-10-06 - Lots of little stuff diff --git a/octodns/processor/ownership.py b/octodns/processor/ownership.py index ed8f34a..d17996a 100644 --- a/octodns/processor/ownership.py +++ b/octodns/processor/ownership.py @@ -13,10 +13,13 @@ from .base import BaseProcessor # 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*'): + def __init__( + self, name, txt_name='_owner', txt_value='*octodns*', txt_ttl=60 + ): super().__init__(name) self.txt_name = txt_name self.txt_value = txt_value + self.txt_ttl = txt_ttl self._txt_values = [txt_value] def process_source_zone(self, desired, *args, **kwargs): @@ -30,7 +33,7 @@ class OwnershipProcessor(BaseProcessor): txt = Record.new( desired, name, - {'type': 'TXT', 'ttl': 60, 'value': self.txt_value}, + {'type': 'TXT', 'ttl': self.txt_ttl, 'value': self.txt_value}, ) # add these w/lenient to cover the case when the ownership record # for a NS delegation record should technically live in the subzone diff --git a/tests/test_octodns_processor_ownership.py b/tests/test_octodns_processor_ownership.py index bd0a630..95682e5 100644 --- a/tests/test_octodns_processor_ownership.py +++ b/tests/test_octodns_processor_ownership.py @@ -55,11 +55,21 @@ class TestOwnershipProcessor(TestCase): self.assertEqual([ownership.txt_value], record.values) # test _is_ownership while we're in here self.assertTrue(ownership._is_ownership(record)) + # default ttl value + self.assertEqual(60, record.ttl) found = True else: self.assertFalse(ownership._is_ownership(record)) self.assertTrue(found) + # change the ttl from the default + ownership.txt_ttl = 300 + got = ownership.process_source_zone(zone.copy()) + record = next( + r for r in got.records if r.name.startswith(ownership.txt_name) + ) + self.assertEqual(300, record.ttl) + def test_process_plan(self): ownership = OwnershipProcessor('ownership') provider = PlannableProvider('helper')