You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
260 lines
7.3 KiB
260 lines
7.3 KiB
#
|
|
#
|
|
#
|
|
|
|
from unittest import TestCase
|
|
from unittest.mock import call, patch
|
|
|
|
from octodns.processor.templating import Templating, TemplatingError
|
|
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_and_target_zones(zone, None, 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_and_target_zones(zone, None, 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_and_target_zones(zone, None, 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_trailing_dots(self, mock_template):
|
|
templ = Templating('test', trailing_dots=False)
|
|
|
|
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_and_target_zones(zone, None, None)
|
|
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_source_id': 'record',
|
|
'record_type': 'TXT',
|
|
'record_ttl': 42,
|
|
'zone_name': 'unit.tests',
|
|
'zone_decoded_name': 'unit.tests',
|
|
'zone_encoded_name': 'unit.tests',
|
|
'zone_num_records': 1,
|
|
}
|
|
),
|
|
mock_template.call_args,
|
|
)
|
|
|
|
def test_context(self):
|
|
templ = Templating(
|
|
'test',
|
|
context={
|
|
# static
|
|
'the_answer': 42,
|
|
# dynamic
|
|
'the_date': lambda _, __: 'today',
|
|
# uses a param
|
|
'provider': lambda _, pro: pro,
|
|
},
|
|
)
|
|
|
|
zone = Zone('unit.tests.', [])
|
|
txt = Record.new(
|
|
zone,
|
|
'txt',
|
|
{
|
|
'type': 'TXT',
|
|
'ttl': 42,
|
|
'values': (
|
|
'the_answer: {the_answer}',
|
|
'the_date: {the_date}',
|
|
'provider: {provider}',
|
|
),
|
|
},
|
|
)
|
|
zone.add_record(txt)
|
|
|
|
got, _ = templ.process_source_and_target_zones(zone, None, 'da-pro')
|
|
txt = _find(got, 'txt')
|
|
self.assertEqual(3, len(txt.values))
|
|
self.assertEqual('provider: da-pro', txt.values[0])
|
|
self.assertEqual('the_answer: 42', txt.values[1])
|
|
self.assertEqual('the_date: today', txt.values[2])
|
|
|
|
def test_bad_key(self):
|
|
templ = Templating('test')
|
|
|
|
zone = Zone('unit.tests.', [])
|
|
txt = Record.new(
|
|
zone,
|
|
'txt',
|
|
{'type': 'TXT', 'ttl': 42, 'value': 'this {bad} does not exist'},
|
|
)
|
|
zone.add_record(txt)
|
|
|
|
with self.assertRaises(TemplatingError) as ctx:
|
|
templ.process_source_and_target_zones(zone, None, None)
|
|
self.assertEqual(
|
|
'Invalid record "txt.unit.tests.", undefined template parameter "bad" in value',
|
|
str(ctx.exception),
|
|
)
|
|
|
|
zone = Zone('unit.tests.', [])
|
|
cname = Record.new(
|
|
zone,
|
|
'cname',
|
|
{
|
|
'type': 'CNAME',
|
|
'ttl': 42,
|
|
'value': '_cname.{bad}something.else.',
|
|
},
|
|
lenient=True,
|
|
)
|
|
zone.add_record(cname)
|
|
|
|
with self.assertRaises(TemplatingError) as ctx:
|
|
templ.process_source_and_target_zones(zone, None, None)
|
|
self.assertEqual(
|
|
'Invalid record "cname.unit.tests.", undefined template parameter "bad" in value',
|
|
str(ctx.exception),
|
|
)
|