From c8672cbb301c65148746c5c158e9b48c28d291e9 Mon Sep 17 00:00:00 2001 From: Ross McFarland Date: Tue, 10 Jun 2025 10:37:29 -0700 Subject: [PATCH] Helps if you add the new files --- .../4af5a11fb21842ffb627d5ee4d80fb14.md | 4 + octodns/processor/templating.py | 51 +++++++ tests/test_octodns_processor_templating.py | 125 ++++++++++++++++++ 3 files changed, 180 insertions(+) create mode 100644 .changelog/4af5a11fb21842ffb627d5ee4d80fb14.md create mode 100644 octodns/processor/templating.py create mode 100644 tests/test_octodns_processor_templating.py diff --git a/.changelog/4af5a11fb21842ffb627d5ee4d80fb14.md b/.changelog/4af5a11fb21842ffb627d5ee4d80fb14.md new file mode 100644 index 0000000..3b7ed8f --- /dev/null +++ b/.changelog/4af5a11fb21842ffb627d5ee4d80fb14.md @@ -0,0 +1,4 @@ +--- +type: minor +--- +Add new Templating proccors \ No newline at end of file diff --git a/octodns/processor/templating.py b/octodns/processor/templating.py new file mode 100644 index 0000000..144264c --- /dev/null +++ b/octodns/processor/templating.py @@ -0,0 +1,51 @@ +# +# +# + +from octodns.processor.base import BaseProcessor + + +class Templating(BaseProcessor): + + def __init__(self, id, *args, **kwargs): + super().__init__(id, *args, **kwargs) + + def process_source_zone(self, desired, sources): + sources = sources or [] + zone_params = { + 'zone_name': desired.decoded_name, + 'zone_decoded_name': desired.decoded_name, + 'zone_encoded_name': desired.name, + 'zone_num_records': len(desired.records), + 'zone_source_ids': ', '.join(s.id for s in sources), + } + + def params(record): + return { + 'record_name': record.decoded_name, + 'record_decoded_name': record.decoded_name, + 'record_encoded_name': record.name, + 'record_fqdn': record.decoded_fqdn, + 'record_decoded_fqdn': record.decoded_fqdn, + 'record_encoded_fqdn': record.fqdn, + 'record_type': record._type, + 'record_ttl': record.ttl, + 'record_source_id': record.source.id if record.source else None, + **zone_params, + } + + for record in desired.records: + if hasattr(record, 'values'): + new_values = [v.template(params(record)) for v in record.values] + if record.values != new_values: + new = record.copy() + new.values = new_values + desired.add_record(new, replace=True) + else: + new_value = record.value.template(params(record)) + if record.value != new_value: + new = record.copy() + new.value = new_value + desired.add_record(new, replace=True) + + return desired diff --git a/tests/test_octodns_processor_templating.py b/tests/test_octodns_processor_templating.py new file mode 100644 index 0000000..80a591a --- /dev/null +++ b/tests/test_octodns_processor_templating.py @@ -0,0 +1,125 @@ +# +# +# + +from unittest import TestCase +from unittest.mock import call, patch + +from octodns.processor.templating import Templating +from octodns.record import Record +from octodns.zone import Zone + + +def _find(zone, name): + return next(r for r in zone.records if r.name == name) + + +class TemplatingTest(TestCase): + def test_cname(self): + templ = Templating('test') + + zone = Zone('unit.tests.', []) + cname = Record.new( + zone, + 'cname', + { + 'type': 'CNAME', + 'ttl': 42, + 'value': '_cname.{zone_name}something.else.', + }, + lenient=True, + ) + zone.add_record(cname) + noop = Record.new( + zone, + 'noop', + { + 'type': 'CNAME', + 'ttl': 42, + 'value': '_noop.nothing_to_do.something.else.', + }, + lenient=True, + ) + zone.add_record(noop) + + got = templ.process_source_zone(zone, None) + cname = _find(got, 'cname') + self.assertEqual('_cname.unit.tests.something.else.', cname.value) + noop = _find(got, 'noop') + self.assertEqual('_noop.nothing_to_do.something.else.', noop.value) + + def test_txt(self): + templ = Templating('test') + + zone = Zone('unit.tests.', []) + txt = Record.new( + zone, + 'txt', + { + 'type': 'TXT', + 'ttl': 42, + 'value': 'There are {zone_num_records} record(s) in {zone_name}', + }, + ) + zone.add_record(txt) + noop = Record.new( + zone, + 'noop', + {'type': 'TXT', 'ttl': 43, 'value': 'Nothing to template here.'}, + ) + zone.add_record(noop) + + got = templ.process_source_zone(zone, None) + txt = _find(got, 'txt') + self.assertEqual('There are 2 record(s) in unit.tests.', txt.values[0]) + noop = _find(got, 'noop') + self.assertEqual('Nothing to template here.', noop.values[0]) + + @patch('octodns.record.TxtValue.template') + def test_params(self, mock_template): + templ = Templating('test') + + class DummySource: + + def __init__(self, id): + self.id = id + + zone = Zone('unit.tests.', []) + record_source = DummySource('record') + txt = Record.new( + zone, + 'txt', + { + 'type': 'TXT', + 'ttl': 42, + 'value': 'There are {zone_num_records} record(s) in {zone_name}', + }, + source=record_source, + ) + zone.add_record(txt) + + templ.process_source_zone( + zone, sources=[record_source, DummySource('other')] + ) + mock_template.assert_called_once() + self.assertEqual( + call( + { + 'record_name': 'txt', + 'record_decoded_name': 'txt', + 'record_encoded_name': 'txt', + 'record_fqdn': 'txt.unit.tests.', + 'record_decoded_fqdn': 'txt.unit.tests.', + 'record_encoded_fqdn': 'txt.unit.tests.', + 'record_type': 'TXT', + 'record_ttl': 42, + 'record_source_id': 'record', + 'zone_name': 'unit.tests.', + 'zone_decoded_name': 'unit.tests.', + 'zone_encoded_name': 'unit.tests.', + 'zone_num_records': 1, + 'zone_source_ids': 'record, other', + } + ), + mock_template.call_args, + )