From 31be2c5de58dd2c1300315b68c0e2d35e598fdf7 Mon Sep 17 00:00:00 2001 From: Ross McFarland Date: Mon, 16 Jun 2025 19:48:17 -0700 Subject: [PATCH] Add context to Templating, cleanly handle records w/o template method, doc --- octodns/processor/templating.py | 64 +++++++++++- tests/test_octodns_processor_templating.py | 114 +++++++++++++++++++-- 2 files changed, 171 insertions(+), 7 deletions(-) diff --git a/octodns/processor/templating.py b/octodns/processor/templating.py index 144264c..c514bec 100644 --- a/octodns/processor/templating.py +++ b/octodns/processor/templating.py @@ -6,9 +6,58 @@ from octodns.processor.base import BaseProcessor class Templating(BaseProcessor): + ''' + Record templating using python format. For simple records like TXT and CAA + that is the value itself. For multi-field records like MX or SRV it's the + text portions, exchange and target respectively. - def __init__(self, id, *args, **kwargs): + Example Processor Config: + + templating: + class: octodns.processor.templating.Templating + # Any k/v present in context will be passed into the .format method and + # thus be available as additional variables in the template. This is all + # optional. + context: + key: value + another: 42 + + Example Records: + + foo: + type: TXT + value: The zone this record lives in is {zone_name}. There are {zone_num_records} record(s). + + bar: + type: MX + values: + - preference: 1 + exchange: mx1.{zone_name}.mail.mx. + - preference: 1 + exchange: mx2.{zone_name}.mail.mx. + + Note that validations for some types reject values with {}. When + encountered the best option is to use record level `lenient: true` + https://github.com/octodns/octodns/blob/main/docs/records.md#lenience + + Note that if you need to add dynamic context you can create a custom + processor that inherits from Templating and passes them into the call to + super, e.g. + + class MyTemplating(Templating): + def __init__(self, *args, context={}, **kwargs): + context['year'] = lambda desired, sources: datetime.now().strftime('%Y') + super().__init__(*args, context, **kwargs) + + See https://docs.python.org/3/library/string.html#custom-string-formatting + for details on formatting options. Anything possible in an `f-string` or + `.format` should work here. + + ''' + + def __init__(self, id, *args, context={}, **kwargs): super().__init__(id, *args, **kwargs) + self.context = context def process_source_zone(self, desired, sources): sources = sources or [] @@ -18,6 +67,13 @@ class Templating(BaseProcessor): 'zone_encoded_name': desired.name, 'zone_num_records': len(desired.records), 'zone_source_ids': ', '.join(s.id for s in sources), + # add any extra context provided to us, if the value is a callable + # object call it passing our params so that arbitrary dynamic + # context can be added for use in formatting + **{ + k: (v(desired, sources) if callable(v) else v) + for k, v in self.context.items() + }, } def params(record): @@ -36,12 +92,18 @@ class Templating(BaseProcessor): for record in desired.records: if hasattr(record, 'values'): + if record.values and not hasattr(record.values[0], 'template'): + # the (custom) value type does not support templating + continue 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: + if not hasattr(record.value, 'template'): + # the (custom) value type does not support templating + continue new_value = record.value.template(params(record)) if record.value != new_value: new = record.copy() diff --git a/tests/test_octodns_processor_templating.py b/tests/test_octodns_processor_templating.py index 80a591a..94472d8 100644 --- a/tests/test_octodns_processor_templating.py +++ b/tests/test_octodns_processor_templating.py @@ -6,10 +6,59 @@ from unittest import TestCase from unittest.mock import call, patch from octodns.processor.templating import Templating -from octodns.record import Record +from octodns.record import Record, ValueMixin, ValuesMixin from octodns.zone import Zone +class DummySource: + + def __init__(self, id): + self.id = str(id) + + +class CustomValue(str): + + @classmethod + def validate(cls, *args, **kwargs): + return [] + + @classmethod + def process(cls, v): + if isinstance(v, (list, tuple)): + return (CustomValue(i) for i in v) + return CustomValue(v) + + @classmethod + def parse_rdata_text(cls, *args, **kwargs): + pass + + def __init__(self, *args, **kwargs): + self._asked_for = set() + + def rdata_text(self): + pass + + def __getattr__(self, item): + self._asked_for.add(item) + raise AttributeError('nope') + + +class Single(ValueMixin, Record): + _type = 'S' + _value_type = CustomValue + + +Record.register_type(Single, 'S') + + +class Multiple(ValuesMixin, Record): + _type = 'M' + _value_type = CustomValue + + +Record.register_type(Multiple, 'M') + + def _find(zone, name): return next(r for r in zone.records if r.name == name) @@ -75,14 +124,29 @@ class TemplatingTest(TestCase): noop = _find(got, 'noop') self.assertEqual('Nothing to template here.', noop.values[0]) - @patch('octodns.record.TxtValue.template') - def test_params(self, mock_template): + def test_no_template(self): templ = Templating('test') - class DummySource: + zone = Zone('unit.tests.', []) + s = Record.new(zone, 's', {'type': 'S', 'ttl': 42, 'value': 'string'}) + zone.add_record(s) - def __init__(self, id): - self.id = id + m = Record.new( + zone, 'm', {'type': 'M', 'ttl': 43, 'values': ('string', 'another')} + ) + zone.add_record(m) + + # this should check for the template method on our values that don't + # have one + templ.process_source_zone(zone, None) + # and these should make sure that the value types were asked if they + # have a template method + self.assertEqual({'template'}, s.value._asked_for) + self.assertEqual({'template'}, m.values[0]._asked_for) + + @patch('octodns.record.TxtValue.template') + def test_params(self, mock_template): + templ = Templating('test') zone = Zone('unit.tests.', []) record_source = DummySource('record') @@ -123,3 +187,41 @@ class TemplatingTest(TestCase): ), mock_template.call_args, ) + + def test_context(self): + templ = Templating( + 'test', + context={ + # static + 'the_answer': 42, + # dynamic + 'the_date': lambda _, __: 'today', + # uses a param + 'num_sources': lambda z, ss: len(ss), + }, + ) + + zone = Zone('unit.tests.', []) + txt = Record.new( + zone, + 'txt', + { + 'type': 'TXT', + 'ttl': 42, + 'values': ( + 'the_answer: {the_answer}', + 'the_date: {the_date}', + 'num_sources: {num_sources}', + ), + }, + ) + zone.add_record(txt) + + got = templ.process_source_zone( + zone, tuple(DummySource(i) for i in range(3)) + ) + txt = _find(got, 'txt') + self.assertEqual(3, len(txt.values)) + self.assertEqual('num_sources: 3', txt.values[0]) + self.assertEqual('the_answer: 42', txt.values[1]) + self.assertEqual('the_date: today', txt.values[2])