mirror/github-octodns
1
0
Fork 0
mirror of https://github.com/github/octodns.git synced 2024-05-11 05:55:00 +00:00
Code Issues Releases Activity

Merge branch 'main' into global-processors

Browse Source
This commit is contained in:
Ross McFarland
2022-09-14 09:52:10 -07:00
committed by GitHub
parent 5498a3b9c9 e1ce226298
commit bf296cd5ff
6 changed files with 421 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_octodns_processor_filter.py
+86 -1
View File
@@ -11,7 +11,12 @@ from __future__ import (
from unittest import TestCase
from octodns.processor.filter import TypeAllowlistFilter, TypeRejectlistFilter
from octodns.processor.filter import (
NameAllowlistFilter,
NameRejectlistFilter,
TypeAllowlistFilter,
TypeRejectlistFilter,
)
from octodns.record import Record
from octodns.zone import Zone
@@ -76,3 +81,83 @@ class TestTypeRejectListFilter(TestCase):
filter_a_aaaa = TypeRejectlistFilter('not-a-aaaa', set(('A', 'AAAA')))
got = filter_a_aaaa.process_target_zone(zone.copy())
self.assertEqual(['txt', 'txt2'], sorted([r.name for r in got.records]))
class TestNameAllowListFilter(TestCase):
zone = Zone('unit.tests.', [])
matches = Record.new(
zone, 'matches', {'type': 'A', 'ttl': 42, 'value': '1.2.3.4'}
)
zone.add_record(matches)
doesnt = Record.new(
zone, 'doesnt', {'type': 'A', 'ttl': 42, 'value': '2.3.4.5'}
)
zone.add_record(doesnt)
matchable1 = Record.new(
zone, 'start-f43ad96-end', {'type': 'A', 'ttl': 42, 'value': '3.4.5.6'}
)
zone.add_record(matchable1)
matchable2 = Record.new(
zone, 'start-a3b444c-end', {'type': 'A', 'ttl': 42, 'value': '4.5.6.7'}
)
zone.add_record(matchable2)
def test_exact(self):
allows = NameAllowlistFilter('exact', ('matches',))
self.assertEqual(4, len(self.zone.records))
filtered = allows.process_source_zone(self.zone.copy())
self.assertEqual(1, len(filtered.records))
self.assertEqual(['matches'], [r.name for r in filtered.records])
def test_regex(self):
allows = NameAllowlistFilter('exact', ('/^start-.+-end$/',))
self.assertEqual(4, len(self.zone.records))
filtered = allows.process_source_zone(self.zone.copy())
self.assertEqual(2, len(filtered.records))
self.assertEqual(
['start-a3b444c-end', 'start-f43ad96-end'],
sorted([r.name for r in filtered.records]),
)
class TestNameRejectListFilter(TestCase):
zone = Zone('unit.tests.', [])
matches = Record.new(
zone, 'matches', {'type': 'A', 'ttl': 42, 'value': '1.2.3.4'}
)
zone.add_record(matches)
doesnt = Record.new(
zone, 'doesnt', {'type': 'A', 'ttl': 42, 'value': '2.3.4.5'}
)
zone.add_record(doesnt)
matchable1 = Record.new(
zone, 'start-f43ad96-end', {'type': 'A', 'ttl': 42, 'value': '3.4.5.6'}
)
zone.add_record(matchable1)
matchable2 = Record.new(
zone, 'start-a3b444c-end', {'type': 'A', 'ttl': 42, 'value': '4.5.6.7'}
)
zone.add_record(matchable2)
def test_exact(self):
rejects = NameRejectlistFilter('exact', ('matches',))
self.assertEqual(4, len(self.zone.records))
filtered = rejects.process_source_zone(self.zone.copy())
self.assertEqual(3, len(filtered.records))
self.assertEqual(
['doesnt', 'start-a3b444c-end', 'start-f43ad96-end'],
sorted([r.name for r in filtered.records]),
)
def test_regex(self):
rejects = NameRejectlistFilter('exact', ('/^start-.+-end$/',))
self.assertEqual(4, len(self.zone.records))
filtered = rejects.process_source_zone(self.zone.copy())
self.assertEqual(2, len(filtered.records))
self.assertEqual(
['doesnt', 'matches'], sorted([r.name for r in filtered.records])
)
tests/test_octodns_processor_restrict.py
+113
View File
@@ -0,0 +1,113 @@
from unittest import TestCase
from octodns.processor.restrict import (
RestrictionException,
TtlRestrictionFilter,
)
from octodns.record import Record
from octodns.zone import Zone
class TestTtlRestrictionFilter(TestCase):
def test_restrict_ttl(self):
# configured values
restrictor = TtlRestrictionFilter('test', min_ttl=32, max_ttl=1024)
zone = Zone('unit.tests.', [])
good = Record.new(
zone, 'good', {'type': 'A', 'ttl': 42, 'value': '1.2.3.4'}
)
zone.add_record(good)
restricted = restrictor.process_source_zone(zone)
self.assertEqual(zone.records, restricted.records)
# too low
low = Record.new(
zone, 'low', {'type': 'A', 'ttl': 16, 'value': '1.2.3.4'}
)
copy = zone.copy()
copy.add_record(low)
with self.assertRaises(RestrictionException) as ctx:
restrictor.process_source_zone(copy)
self.assertEqual(
'low.unit.tests. ttl=16 too low, min_ttl=32', str(ctx.exception)
)
# with lenient set, we can go lower
lenient = Record.new(
zone,
'low',
{
'octodns': {'lenient': True},
'type': 'A',
'ttl': 16,
'value': '1.2.3.4',
},
)
copy = zone.copy()
copy.add_record(lenient)
restricted = restrictor.process_source_zone(copy)
self.assertEqual(copy.records, restricted.records)
# too high
high = Record.new(
zone, 'high', {'type': 'A', 'ttl': 2048, 'value': '1.2.3.4'}
)
copy = zone.copy()
copy.add_record(high)
with self.assertRaises(RestrictionException) as ctx:
restrictor.process_source_zone(copy)
self.assertEqual(
'high.unit.tests. ttl=2048 too high, max_ttl=1024',
str(ctx.exception),
)
# too low defaults
restrictor = TtlRestrictionFilter('test')
low = Record.new(
zone, 'low', {'type': 'A', 'ttl': 0, 'value': '1.2.3.4'}
)
copy = zone.copy()
copy.add_record(low)
with self.assertRaises(RestrictionException) as ctx:
restrictor.process_source_zone(copy)
self.assertEqual(
'low.unit.tests. ttl=0 too low, min_ttl=1', str(ctx.exception)
)
# too high defaults
high = Record.new(
zone, 'high', {'type': 'A', 'ttl': 999999, 'value': '1.2.3.4'}
)
copy = zone.copy()
copy.add_record(high)
with self.assertRaises(RestrictionException) as ctx:
restrictor.process_source_zone(copy)
self.assertEqual(
'high.unit.tests. ttl=999999 too high, max_ttl=604800',
str(ctx.exception),
)
# allowed_ttls
restrictor = TtlRestrictionFilter('test', allowed_ttls=[42, 300])
# add 300 (42 is already there)
another = Record.new(
zone, 'another', {'type': 'A', 'ttl': 300, 'value': '4.5.6.7'}
)
zone.add_record(another)
# 42 and 300 are allowed through
restricted = restrictor.process_source_zone(zone)
self.assertEqual(zone.records, restricted.records)
# 16 is not
copy = zone.copy()
copy.add_record(low)
with self.assertRaises(RestrictionException) as ctx:
restrictor.process_source_zone(copy)
self.assertEqual(
'low.unit.tests. ttl=0 not an allowed value, allowed_ttls={42, 300}',
str(ctx.exception),
)