mirror of
https://github.com/github/octodns.git
synced 2024-05-11 05:55:00 +00:00
130 lines
4.5 KiB
Python
130 lines
4.5 KiB
Python
#
|
|
#
|
|
#
|
|
|
|
from __future__ import (
|
|
absolute_import,
|
|
division,
|
|
print_function,
|
|
unicode_literals,
|
|
)
|
|
|
|
from unittest import TestCase
|
|
|
|
from octodns.processor.ownership import OwnershipProcessor
|
|
from octodns.record import Delete, Record
|
|
from octodns.zone import Zone
|
|
|
|
from helpers import PlannableProvider
|
|
|
|
|
|
zone = Zone('unit.tests.', [])
|
|
records = {}
|
|
for record in [
|
|
Record.new(
|
|
zone, '', {'ttl': 30, 'type': 'A', 'values': ['1.2.3.4', '5.6.7.8']}
|
|
),
|
|
Record.new(zone, 'the-a', {'ttl': 30, 'type': 'A', 'value': '1.2.3.4'}),
|
|
Record.new(zone, 'the-aaaa', {'ttl': 30, 'type': 'AAAA', 'value': '::1'}),
|
|
Record.new(
|
|
zone, 'the-txt', {'ttl': 30, 'type': 'TXT', 'value': 'Hello World!'}
|
|
),
|
|
Record.new(zone, '*', {'ttl': 30, 'type': 'A', 'value': '4.3.2.1'}),
|
|
]:
|
|
records[record.name] = record
|
|
zone.add_record(record)
|
|
|
|
|
|
class TestOwnershipProcessor(TestCase):
|
|
def test_process_source_zone(self):
|
|
ownership = OwnershipProcessor('ownership')
|
|
|
|
got = ownership.process_source_zone(zone.copy())
|
|
self.assertEqual(
|
|
[
|
|
'',
|
|
'*',
|
|
'_owner.a',
|
|
'_owner.a._wildcard',
|
|
'_owner.a.the-a',
|
|
'_owner.aaaa.the-aaaa',
|
|
'_owner.txt.the-txt',
|
|
'the-a',
|
|
'the-aaaa',
|
|
'the-txt',
|
|
],
|
|
sorted([r.name for r in got.records]),
|
|
)
|
|
|
|
found = False
|
|
for record in got.records:
|
|
if record.name.startswith(ownership.txt_name):
|
|
self.assertEqual([ownership.txt_value], record.values)
|
|
# test _is_ownership while we're in here
|
|
self.assertTrue(ownership._is_ownership(record))
|
|
found = True
|
|
else:
|
|
self.assertFalse(ownership._is_ownership(record))
|
|
self.assertTrue(found)
|
|
|
|
def test_process_plan(self):
|
|
ownership = OwnershipProcessor('ownership')
|
|
provider = PlannableProvider('helper')
|
|
|
|
# No plan, is a quick noop
|
|
self.assertFalse(ownership.process_plan(None))
|
|
|
|
# Nothing exists create both records and ownership
|
|
ownership_added = ownership.process_source_zone(zone.copy())
|
|
plan = provider.plan(ownership_added)
|
|
self.assertTrue(plan)
|
|
# Double the number of records
|
|
self.assertEqual(len(records) * 2, len(plan.changes))
|
|
# Now process the plan, shouldn't make any changes, we're creating
|
|
# everything
|
|
got = ownership.process_plan(plan)
|
|
self.assertTrue(got)
|
|
self.assertEqual(len(records) * 2, len(got.changes))
|
|
|
|
# Something extra exists and doesn't have ownership TXT, leave it
|
|
# alone, we don't own it.
|
|
extra_a = Record.new(
|
|
zone, 'extra-a', {'ttl': 30, 'type': 'A', 'value': '4.4.4.4'}
|
|
)
|
|
plan.existing.add_record(extra_a)
|
|
# If we'd done a "real" plan we'd have a delete for the extra thing.
|
|
plan.changes.append(Delete(extra_a))
|
|
# Process the plan, shouldn't make any changes since the extra bit is
|
|
# something we don't own
|
|
got = ownership.process_plan(plan)
|
|
self.assertTrue(got)
|
|
self.assertEqual(len(records) * 2, len(got.changes))
|
|
|
|
# Something extra exists and does have an ownership record so we will
|
|
# delete it...
|
|
copy = Zone('unit.tests.', [])
|
|
for record in records.values():
|
|
if record.name != 'the-a':
|
|
copy.add_record(record)
|
|
# New ownership, without the `the-a`
|
|
ownership_added = ownership.process_source_zone(copy)
|
|
self.assertEqual(len(records) * 2 - 2, len(ownership_added.records))
|
|
plan = provider.plan(ownership_added)
|
|
# Fake the extra existing by adding the record, its ownership, and the
|
|
# two delete changes.
|
|
the_a = records['the-a']
|
|
plan.existing.add_record(the_a)
|
|
name = f'{ownership.txt_name}.a.the-a'
|
|
the_a_ownership = Record.new(
|
|
zone, name, {'ttl': 30, 'type': 'TXT', 'value': ownership.txt_value}
|
|
)
|
|
plan.existing.add_record(the_a_ownership)
|
|
plan.changes.append(Delete(the_a))
|
|
plan.changes.append(Delete(the_a_ownership))
|
|
# Finally process the plan, should be a noop and we should get the same
|
|
# plan out, meaning the planned deletes were allowed to happen.
|
|
got = ownership.process_plan(plan)
|
|
self.assertTrue(got)
|
|
self.assertEqual(plan, got)
|
|
self.assertEqual(len(plan.changes), len(got.changes))
|