Merge pull request #1259 from octodns/templating-processor

Implment Templating processor and prototype Value.template methods

.changelog/4af5a11fb21842ffb627d5ee4d80fb14.md

@@ -0,0 +1,4 @@
+---
+type: minor
+---
+Templating processor added

.git_hooks_pre-commit

@@ -7,7 +7,7 @@ GIT=$(dirname "$HOOKS")
 ROOT=$(dirname "$GIT")
 
 . "$ROOT/env/bin/activate"
+"$ROOT/script/changelog" check
 "$ROOT/script/lint"
 "$ROOT/script/format" --check --quiet || (echo "Formatting check failed, run ./script/format" && exit 1)
 "$ROOT/script/coverage"
-"$ROOT/script/changelog" check

octodns/processor/templating.py

@@ -0,0 +1,113 @@
+#
+#
+#
+
+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.
+
+    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 []
+        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),
+            # 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):
+            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'):
+                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()
+                    new.value = new_value
+                    desired.add_record(new, replace=True)
+
+        return desired

octodns/record/caa.py

@@ -87,6 +87,13 @@ class CaaValue(EqualityTupleMixin, dict):
     def rdata_text(self):
         return f'{self.flags} {self.tag} {self.value}'
 
+    def template(self, params):
+        if '{' not in self.value:
+            return self
+        new = self.__class__(self)
+        new.value = new.value.format(**params)
+        return new
+
     def _equality_tuple(self):
         return (self.flags, self.tag, self.value)
 

octodns/record/chunked.py

@@ -80,3 +80,8 @@ class _ChunkedValue(str):
     @property
     def rdata_text(self):
         return self
+
+    def template(self, params):
+        if '{' not in self:
+            return self
+        return self.__class__(self.format(**params))

octodns/record/ds.py

@@ -164,6 +164,13 @@ class DsValue(EqualityTupleMixin, dict):
             f'{self.key_tag} {self.algorithm} {self.digest_type} {self.digest}'
         )
 
+    def template(self, params):
+        if '{' not in self.digest:
+            return self
+        new = self.__class__(self)
+        new.digest = new.digest.format(**params)
+        return new
+
     def _equality_tuple(self):
         return (self.key_tag, self.algorithm, self.digest_type, self.digest)
 

octodns/record/ip.py

@@ -45,5 +45,8 @@ class _IpValue(str):
     def rdata_text(self):
         return self
 
+    def template(self, params):
+        return self
+
 
 _IpAddress = _IpValue

octodns/record/loc.py

@@ -305,6 +305,9 @@ class LocValue(EqualityTupleMixin, dict):
     def rdata_text(self):
         return f'{self.lat_degrees} {self.lat_minutes} {self.lat_seconds} {self.lat_direction} {self.long_degrees} {self.long_minutes} {self.long_seconds} {self.long_direction} {self.altitude}m {self.size}m {self.precision_horz}m {self.precision_vert}m'
 
+    def template(self, params):
+        return self
+
     def __hash__(self):
         return hash(
             (

octodns/record/mx.py

@@ -101,6 +101,13 @@ class MxValue(EqualityTupleMixin, dict):
     def rdata_text(self):
         return f'{self.preference} {self.exchange}'
 
+    def template(self, params):
+        if '{' not in self.exchange:
+            return self
+        new = self.__class__(self)
+        new.exchange = new.exchange.format(**params)
+        return new
+
     def __hash__(self):
         return hash((self.preference, self.exchange))
 

octodns/record/naptr.py

@@ -138,6 +138,19 @@ class NaptrValue(EqualityTupleMixin, dict):
     def rdata_text(self):
         return f'{self.order} {self.preference} {self.flags} {self.service} {self.regexp} {self.replacement}'
 
+    def template(self, params):
+        if (
+            '{' not in self.service
+            and '{' not in self.regexp
+            and '{' not in self.replacement
+        ):
+            return self
+        new = self.__class__(self)
+        new.service = new.service.format(**params)
+        new.regexp = new.regexp.format(**params)
+        new.replacement = new.replacement.format(**params)
+        return new
+
     def __hash__(self):
         return hash(self.__repr__())
 

octodns/record/srv.py

@@ -135,6 +135,13 @@ class SrvValue(EqualityTupleMixin, dict):
     def rdata_text(self):
         return f"{self.priority} {self.weight} {self.port} {self.target}"
 
+    def template(self, params):
+        if '{' not in self.target:
+            return self
+        new = self.__class__(self)
+        new.target = new.target.format(**params)
+        return new
+
     def __hash__(self):
         return hash(self.__repr__())
 

octodns/record/sshfp.py

@@ -105,6 +105,13 @@ class SshfpValue(EqualityTupleMixin, dict):
     def rdata_text(self):
         return f'{self.algorithm} {self.fingerprint_type} {self.fingerprint}'
 
+    def template(self, params):
+        if '{' not in self.fingerprint:
+            return self
+        new = self.__class__(self)
+        new.fingerprint = new.fingerprint.format(**params)
+        return new
+
     def __hash__(self):
         return hash(self.__repr__())
 

octodns/record/svcb.py

@@ -287,6 +287,14 @@ class SvcbValue(EqualityTupleMixin, dict):
                     params += f'={svcparamvalue}'
         return f'{self.svcpriority} {self.targetname}{params}'
 
+    def template(self, params):
+        if '{' not in self.targetname:
+            return self
+        new = self.__class__(self)
+        new.targetname = new.targetname.format(**params)
+        # TODO: what, if any of the svcparams should be templated
+        return new
+
     def __hash__(self):
         return hash(self.__repr__())
 

octodns/record/target.py

@@ -41,6 +41,11 @@ class _TargetValue(str):
     def rdata_text(self):
         return self
 
+    def template(self, params):
+        if '{' not in self:
+            return self
+        return self.__class__(self.format(**params))
+
 
 #
 # much like _TargetValue, but geared towards multiple values
@@ -75,3 +80,8 @@ class _TargetsValue(str):
     @property
     def rdata_text(self):
         return self
+
+    def template(self, params):
+        if '{' not in self:
+            return self
+        return self.__class__(self.format(**params))

octodns/record/tlsa.py

@@ -136,6 +136,15 @@ class TlsaValue(EqualityTupleMixin, dict):
     def rdata_text(self):
         return f'{self.certificate_usage} {self.selector} {self.matching_type} {self.certificate_association_data}'
 
+    def template(self, params):
+        if '{' not in self.certificate_association_data:
+            return self
+        new = self.__class__(self)
+        new.certificate_association_data = (
+            new.certificate_association_data.format(**params)
+        )
+        return new
+
     def _equality_tuple(self):
         return (
             self.certificate_usage,

octodns/record/urlfwd.py

@@ -132,6 +132,14 @@ class UrlfwdValue(EqualityTupleMixin, dict):
     def rdata_text(self):
         return f'"{self.path}" "{self.target}" {self.code} {self.masking} {self.query}'
 
+    def template(self, params):
+        if '{' not in self.path and '{' not in self.target:
+            return self
+        new = self.__class__(self)
+        new.path = new.path.format(**params)
+        new.target = new.target.format(**params)
+        return new
+
     def _equality_tuple(self):
         return (self.path, self.target, self.code, self.masking, self.query)
 

tests/test_octodns_processor_templating.py

@@ -0,0 +1,227 @@
+#
+#
+#
+
+from unittest import TestCase
+from unittest.mock import call, patch
+
+from octodns.processor.templating import Templating
+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)
+
+
+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])
+
+    def test_no_template(self):
+        templ = Templating('test')
+
+        zone = Zone('unit.tests.', [])
+        s = Record.new(zone, 's', {'type': 'S', 'ttl': 42, 'value': 'string'})
+        zone.add_record(s)
+
+        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')
+        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,
+        )
+
+    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])

tests/test_octodns_record_caa.py

@@ -285,3 +285,20 @@ class TestRecordCaa(TestCase):
                 {'type': 'CAA', 'ttl': 600, 'value': {'tag': 'iodef'}},
             )
         self.assertEqual(['missing value'], ctx.exception.reasons)
+
+
+class TestCaaValue(TestCase):
+
+    def test_template(self):
+        value = CaaValue(
+            {'flags': 0, 'tag': 'issue', 'value': 'ca.example.net'}
+        )
+        got = value.template({'needle': 42})
+        self.assertIs(value, got)
+
+        value = CaaValue(
+            {'flags': 0, 'tag': 'issue', 'value': 'ca.{needle}.net'}
+        )
+        got = value.template({'needle': 42})
+        self.assertIsNot(value, got)
+        self.assertEqual('ca.42.net', got.value)

tests/test_octodns_record_chunked.py

@@ -115,3 +115,15 @@ class TestChunkedValue(TestCase):
 
         sc = self.SmallerChunkedMixin(['0123456789'])
         self.assertEqual(['"01234567" "89"'], sc.chunked_values)
+
+    def test_template(self):
+        s = 'this.has.no.templating.'
+        value = _ChunkedValue(s)
+        got = value.template({'needle': 42})
+        self.assertIs(value, got)
+
+        s = 'this.does.{needle}.have.templating.'
+        value = _ChunkedValue(s)
+        got = value.template({'needle': 42})
+        self.assertIsNot(value, got)
+        self.assertEqual('this.does.42.have.templating.', got)

tests/test_octodns_record_ds.py

@@ -259,3 +259,30 @@ class TestRecordDs(TestCase):
         self.assertEqual(DsValue(values[1]), a.values[1].data)
         self.assertEqual('1 2 3 99148c44', a.values[1].rdata_text)
         self.assertEqual('1 2 3 99148c44', a.values[1].__repr__())
+
+
+class TestDsValue(TestCase):
+
+    def test_template(self):
+        value = DsValue(
+            {
+                'key_tag': 0,
+                'algorithm': 1,
+                'digest_type': 2,
+                'digest': 'abcdef0123456',
+            }
+        )
+        got = value.template({'needle': 42})
+        self.assertIs(value, got)
+
+        value = DsValue(
+            {
+                'key_tag': 0,
+                'algorithm': 1,
+                'digest_type': 2,
+                'digest': 'abcd{needle}ef0123456',
+            }
+        )
+        got = value.template({'needle': 42})
+        self.assertIsNot(value, got)
+        self.assertEqual('abcd42ef0123456', got.digest)

tests/test_octodns_record_ip.py

@@ -27,3 +27,11 @@ class TestRecordIp(TestCase):
         zone = Zone('unit.tests.', [])
         a = ARecord(zone, 'a', {'ttl': 42, 'value': '1.2.3.4'})
         self.assertEqual('1.2.3.4', a.values[0].rdata_text)
+
+
+class TestIpValue(TestCase):
+
+    def test_template(self):
+        value = Ipv4Value('1.2.3.4')
+        # template is a noop
+        self.assertIs(value, value.template({'needle': 42}))

tests/test_octodns_record_loc.py

@@ -715,3 +715,26 @@ class TestRecordLoc(TestCase):
         self.assertEqual(
             ['invalid value for size "99999999.99"'], ctx.exception.reasons
         )
+
+
+class TestLocValue(TestCase):
+
+    def test_template(self):
+        value = LocValue(
+            {
+                'lat_degrees': 31,
+                'lat_minutes': 58,
+                'lat_seconds': 52.1,
+                'lat_direction': 'S',
+                'long_degrees': 115,
+                'long_minutes': 49,
+                'long_seconds': 11.7,
+                'long_direction': 'E',
+                'altitude': 20,
+                'size': 10,
+                'precision_horz': 10,
+                'precision_vert': 2,
+            }
+        )
+        # loc value template is a noop
+        self.assertIs(value, value.template({}))

tests/test_octodns_record_mx.py

@@ -268,3 +268,16 @@ class TestRecordMx(TestCase):
             },
         )
         self.assertEqual('.', record.values[0].exchange)
+
+
+class TestMxValue(TestCase):
+
+    def test_template(self):
+        value = MxValue({'preference': 10, 'exchange': 'smtp1.'})
+        got = value.template({'needle': 42})
+        self.assertIs(value, got)
+
+        value = MxValue({'preference': 10, 'exchange': 'smtp1.{needle}.'})
+        got = value.template({'needle': 42})
+        self.assertIsNot(value, got)
+        self.assertEqual('smtp1.42.', got.exchange)

tests/test_octodns_record_naptr.py

@@ -449,3 +449,36 @@ class TestRecordNaptr(TestCase):
         with self.assertRaises(ValidationError) as ctx:
             Record.new(self.zone, '', {'type': 'NAPTR', 'ttl': 600, 'value': v})
         self.assertEqual(['unrecognized flags "X"'], ctx.exception.reasons)
+
+
+class TestNaptrValue(TestCase):
+
+    def test_template(self):
+        value = NaptrValue(
+            {
+                'order': 10,
+                'preference': 11,
+                'flags': 'X',
+                'service': 'Y',
+                'regexp': 'Z',
+                'replacement': '.',
+            }
+        )
+        got = value.template({'needle': 42})
+        self.assertIs(value, got)
+
+        value = NaptrValue(
+            {
+                'order': 10,
+                'preference': 11,
+                'flags': 'X',
+                'service': 'Y{needle}',
+                'regexp': 'Z{needle}',
+                'replacement': '.{needle}',
+            }
+        )
+        got = value.template({'needle': 42})
+        self.assertIsNot(value, got)
+        self.assertEqual('Y42', got.service)
+        self.assertEqual('Z42', got.regexp)
+        self.assertEqual('.42', got.replacement)

tests/test_octodns_record_srv.py

@@ -450,3 +450,30 @@ class TestRecordSrv(TestCase):
             ['Invalid SRV target "100 foo.bar.com." is not a valid FQDN.'],
             ctx.exception.reasons,
         )
+
+
+class TestSrvValue(TestCase):
+
+    def test_template(self):
+        value = SrvValue(
+            {
+                'priority': 10,
+                'weight': 11,
+                'port': 12,
+                'target': 'no_placeholders',
+            }
+        )
+        got = value.template({'needle': 42})
+        self.assertIs(value, got)
+
+        value = SrvValue(
+            {
+                'priority': 10,
+                'weight': 11,
+                'port': 12,
+                'target': 'has_{needle}_placeholder',
+            }
+        )
+        got = value.template({'needle': 42})
+        self.assertIsNot(value, got)
+        self.assertEqual('has_42_placeholder', got.target)

tests/test_octodns_record_sshfp.py

@@ -333,3 +333,24 @@ class TestRecordSshfp(TestCase):
                 },
             )
         self.assertEqual(['missing fingerprint'], ctx.exception.reasons)
+
+
+class TestSshFpValue(TestCase):
+
+    def test_template(self):
+        value = SshfpValue(
+            {'algorithm': 10, 'fingerprint_type': 11, 'fingerprint': 'abc123'}
+        )
+        got = value.template({'needle': 42})
+        self.assertIs(value, got)
+
+        value = SshfpValue(
+            {
+                'algorithm': 10,
+                'fingerprint_type': 11,
+                'fingerprint': 'ab{needle}c123',
+            }
+        )
+        got = value.template({'needle': 42})
+        self.assertIsNot(value, got)
+        self.assertEqual('ab42c123', got.fingerprint)

tests/test_octodns_record_svcb.py

@@ -673,3 +673,18 @@ class TestRecordSvcb(TestCase):
             ],
             ctx.exception.reasons,
         )
+
+
+class TestSrvValue(TestCase):
+
+    def test_template(self):
+        value = SvcbValue({'svcpriority': 0, 'targetname': 'foo.example.com.'})
+        got = value.template({'needle': 42})
+        self.assertIs(value, got)
+
+        value = SvcbValue(
+            {'svcpriority': 0, 'targetname': 'foo.{needle}.example.com.'}
+        )
+        got = value.template({'needle': 42})
+        self.assertIsNot(value, got)
+        self.assertEqual('foo.42.example.com.', got.targetname)

tests/test_octodns_record_target.py

@@ -5,7 +5,7 @@
 from unittest import TestCase
 
 from octodns.record.alias import AliasRecord
-from octodns.record.target import _TargetValue
+from octodns.record.target import _TargetsValue, _TargetValue
 from octodns.zone import Zone
 
 
@@ -28,3 +28,33 @@ class TestRecordTarget(TestCase):
         zone = Zone('unit.tests.', [])
         a = AliasRecord(zone, 'a', {'ttl': 42, 'value': 'some.target.'})
         self.assertEqual('some.target.', a.value.rdata_text)
+
+
+class TestTargetValue(TestCase):
+
+    def test_template(self):
+        s = 'this.has.no.templating.'
+        value = _TargetValue(s)
+        got = value.template({'needle': 42})
+        self.assertIs(value, got)
+
+        s = 'this.does.{needle}.have.templating.'
+        value = _TargetValue(s)
+        got = value.template({'needle': 42})
+        self.assertIsNot(value, got)
+        self.assertEqual('this.does.42.have.templating.', got)
+
+
+class TestTargetsValue(TestCase):
+
+    def test_template(self):
+        s = 'this.has.no.templating.'
+        value = _TargetsValue(s)
+        got = value.template({'needle': 42})
+        self.assertIs(value, got)
+
+        s = 'this.does.{needle}.have.templating.'
+        value = _TargetsValue(s)
+        got = value.template({'needle': 42})
+        self.assertIsNot(value, got)
+        self.assertEqual('this.does.42.have.templating.', got)

tests/test_octodns_record_tlsa.py

@@ -429,3 +429,32 @@ class TestRecordTlsa(TestCase):
                 'invalid matching_type "{value["matching_type"]}"',
                 ctx.exception.reasons,
             )
+
+
+class TestTlsaValue(TestCase):
+
+    def test_template(self):
+        value = TlsaValue(
+            {
+                'certificate_usage': 1,
+                'selector': 1,
+                'matching_type': 1,
+                'certificate_association_data': 'ABABABABABABABABAB',
+            }
+        )
+        got = value.template({'needle': 42})
+        self.assertIs(value, got)
+
+        value = TlsaValue(
+            {
+                'certificate_usage': 1,
+                'selector': 1,
+                'matching_type': 1,
+                'certificate_association_data': 'ABAB{needle}ABABABABABABAB',
+            }
+        )
+        got = value.template({'needle': 42})
+        self.assertIsNot(value, got)
+        self.assertEqual(
+            'ABAB42ABABABABABABAB', got.certificate_association_data
+        )

tests/test_octodns_record_urlfwd.py

@@ -483,3 +483,33 @@ class TestRecordUrlfwd(TestCase):
             {'ttl': 32, 'value': UrlfwdValue.parse_rdata_text(rdata)},
         )
         self.assertEqual(rdata, record.values[0].rdata_text)
+
+
+class TestUrlfwdValue(TestCase):
+
+    def test_template(self):
+        value = UrlfwdValue(
+            {
+                'path': '/',
+                'target': 'http://foo',
+                'code': 301,
+                'masking': 2,
+                'query': 0,
+            }
+        )
+        got = value.template({'needle': 42})
+        self.assertIs(value, got)
+
+        value = UrlfwdValue(
+            {
+                'path': '/{needle}',
+                'target': 'http://foo.{needle}',
+                'code': 301,
+                'masking': 2,
+                'query': 0,
+            }
+        )
+        got = value.template({'needle': 42})
+        self.assertIsNot(value, got)
+        self.assertEqual('/42', got.path)
+        self.assertEqual('http://foo.42', got.target)