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

Refactor PowerDNS version handling to be transparently cached properties

Browse Source
This commit is contained in:
Ross McFarland
2020-06-12 09:36:57 -07:00
parent 836d6daee2
commit 2a159bf93b
2 changed files with 75 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
octodns/provider/powerdns.py
View File

@@ -25,12 +25,11 @@ class PowerDnsBaseProvider(BaseProvider):
self.host = host
self.port = port
self.version_detected = ''
self.soa_edit_api = "INCEPTION-INCREMENT"
self.check_status_not_found = False
self.scheme = scheme
self.timeout = timeout
self._powerdns_version = None
sess = Session()
sess.headers.update({'X-API-Key': api_key})
self._sess = sess
@@ -169,56 +168,46 @@ class PowerDnsBaseProvider(BaseProvider):
'ttl': rrset['ttl']
}
def detect_version(self):
# Only detect version once
if self.version_detected != '':
self.log.debug('detect_version: version %s allready detected',
self.version_detected)
return self.version_detected
try:
self.log.debug('detect_version: getting version from server')
resp = self._get('')
except HTTPError as e:
if e.response.status_code == 401:
# Nicer error message for auth problems
raise Exception('PowerDNS unauthorized host={}'
.format(self.host))
else:
@property
def powerdns_version(self):
if self._powerdns_version is None:
try:
resp = self._get('')
except HTTPError as e:
if e.response.status_code == 401:
# Nicer error message for auth problems
raise Exception('PowerDNS unauthorized host={}'
.format(self.host))
raise
self.version_detected = resp.json()["version"]
self.log.debug('detect_version: got version %s from server',
self.version_detected)
self.configure_for_version(self.version_detected)
version = resp.json()['version']
self.log.debug('powerdns_version: got version %s from server',
version)
self._powerdns_version = [int(p) for p in version.split('.')]
def configure_for_version(self, version):
major, minor, patch = version.split('.', 2)
major, minor, patch = int(major), int(minor), int(patch)
self.log.debug('configure_for_version: configure for '
'major: %s, minor: %s, patch: %s',
major, minor, patch)
return self._powerdns_version
# Defaults for v4.0.0
self.soa_edit_api = "INCEPTION-INCREMENT"
self.check_status_not_found = False
@property
def soa_edit_api(self):
# >>> [4, 4, 3] >= [4, 3]
# True
# >>> [4, 3, 3] >= [4, 3]
# True
# >>> [4, 1, 3] >= [4, 3]
# False
if self.powerdns_version >= [4, 3]:
return 'DEFAULT'
return 'INCEPTION-INCREMENT'
if major == 4 and minor >= 2:
self.log.debug("configure_for_version: Version >= 4.2")
self.soa_edit_api = "INCEPTION-INCREMENT"
self.check_status_not_found = True
if major == 4 and minor >= 3:
self.log.debug("configure_for_version: Version >= 4.3")
self.soa_edit_api = "DEFAULT"
self.check_status_not_found = True
@property
def check_status_not_found(self):
# >=4.2.x returns 404 when not found
return self.powerdns_version >= [4, 2]
def populate(self, zone, target=False, lenient=False):
self.log.debug('populate: name=%s, target=%s, lenient=%s', zone.name,
target, lenient)
self.detect_version()
resp = None
try:
resp = self._get('zones/{}'.format(zone.name))
@@ -231,16 +220,16 @@ class PowerDnsBaseProvider(BaseProvider):
.format(self.host))
elif e.response.status_code == 404 \
and self.check_status_not_found:
# 404 or 422 means powerdns doesn't know anything about the
# requested domain. We'll just ignore it here and leave the
# zone untouched.
# 404 means powerdns doesn't know anything about the requested
# domain. We'll just ignore it here and leave the zone
# untouched.
pass
elif e.response.status_code == 422 \
and error.startswith('Could not find domain ') \
and not self.check_status_not_found:
# 404 or 422 means powerdns doesn't know anything about the
# requested domain. We'll just ignore it here and leave the
# zone untouched.
# 422 means powerdns doesn't know anything about the requested
# domain. We'll just ignore it here and leave the zone
# untouched.
pass
else:
# just re-throw
@@ -458,10 +447,6 @@ class PowerDnsProvider(PowerDnsBaseProvider):
- 1.2.3.5.
# The nameserver record TTL when managed, (optional, default 600)
nameserver_ttl: 600
# The SOA-EDIT-API value to set for newly created zones (optional)
# Defaults to INCEPTION-INCREMENT. PowerDNS >=4.3.x users should use
# 'DEFAULT' instead, as INCEPTION-INCREMENT is no longer a valid value.
soa_edit_api: INCEPTION-INCREMENT
'''
def __init__(self, id, host, api_key, port=8081, nameserver_values=None,

73
tests/test_octodns_provider_powerdns.py
View File

@@ -50,7 +50,7 @@ class TestPowerDnsProvider(TestCase):
mock.get(ANY, status_code=401, text='Unauthorized')
with self.assertRaises(Exception) as ctx:
provider.detect_version()
provider.powerdns_version
self.assertTrue('unauthorized' in text_type(ctx.exception))
# Api not found
@@ -58,23 +58,20 @@ class TestPowerDnsProvider(TestCase):
mock.get(ANY, status_code=404, text='Not Found')
with self.assertRaises(Exception) as ctx:
provider.detect_version()
provider.powerdns_version
self.assertTrue('404' in text_type(ctx.exception))
# Test version detection
with requests_mock() as mock:
mock.get('http://non.existent:8081/api/v1/servers/localhost',
status_code=200, json={'version': "4.1.10"})
provider.detect_version()
self.assertEquals(provider.version_detected, '4.1.10')
self.assertEquals(provider.powerdns_version, [4, 1, 10])
# Test version detection for second time (should stay at 4.1.10)
with requests_mock() as mock:
mock.get('http://non.existent:8081/api/v1/servers/localhost',
status_code=200, json={'version': "4.2.0"})
provider.detect_version()
self.assertEquals(provider.version_detected, '4.1.10')
self.assertEquals(provider.powerdns_version, [4, 1, 10])
# Test version detection
with requests_mock() as mock:
@@ -82,39 +79,44 @@ class TestPowerDnsProvider(TestCase):
status_code=200, json={'version': "4.2.0"})
# Reset version, so detection will try again
provider.version_detected = ''
provider.detect_version()
self.assertNotEquals(provider.version_detected, '4.1.10')
provider._powerdns_version = None
self.assertNotEquals(provider.powerdns_version, [4, 1, 10])
def test_provider_version_config(self):
provider = PowerDnsProvider('test', 'non.existent', 'api-key',
nameserver_values=['8.8.8.8.',
'9.9.9.9.'])
provider.check_status_not_found = None
provider.soa_edit_api = 'something else'
# Test version 4.1.0
provider.configure_for_version("4.1.0")
self.assertEquals(provider.soa_edit_api, 'INCEPTION-INCREMENT')
self.assertFalse(
provider.check_status_not_found,
'check_status_not_found should be false '
'for version 4.1.x and below')
provider._powerdns_version = None
with requests_mock() as mock:
mock.get('http://non.existent:8081/api/v1/servers/localhost',
status_code=200, json={'version': "4.1.10"})
self.assertEquals(provider.soa_edit_api, 'INCEPTION-INCREMENT')
self.assertFalse(
provider.check_status_not_found,
'check_status_not_found should be false '
'for version 4.1.x and below')
# Test version 4.2.0
provider.configure_for_version("4.2.0")
self.assertEquals(provider.soa_edit_api, 'INCEPTION-INCREMENT')
self.assertTrue(
provider.check_status_not_found,
'check_status_not_found should be true for version 4.2.x')
provider._powerdns_version = None
with requests_mock() as mock:
mock.get('http://non.existent:8081/api/v1/servers/localhost',
status_code=200, json={'version': "4.2.0"})
self.assertEquals(provider.soa_edit_api, 'INCEPTION-INCREMENT')
self.assertTrue(
provider.check_status_not_found,
'check_status_not_found should be true for version 4.2.x')
# Test version 4.3.0
provider.configure_for_version("4.3.0")
self.assertEquals(provider.soa_edit_api, 'DEFAULT')
self.assertTrue(
provider.check_status_not_found,
'check_status_not_found should be true for version 4.3.x')
provider._powerdns_version = None
with requests_mock() as mock:
mock.get('http://non.existent:8081/api/v1/servers/localhost',
status_code=200, json={'version': "4.3.0"})
self.assertEquals(provider.soa_edit_api, 'DEFAULT')
self.assertTrue(
provider.check_status_not_found,
'check_status_not_found should be true for version 4.3.x')
def test_provider(self):
provider = PowerDnsProvider('test', 'non.existent', 'api-key',
@@ -125,9 +127,7 @@ class TestPowerDnsProvider(TestCase):
with requests_mock() as mock:
mock.get('http://non.existent:8081/api/v1/servers/localhost',
status_code=200, json={'version': "4.1.10"})
provider.detect_version()
self.assertEquals(provider.version_detected, '4.1.10')
self.assertEquals(provider.powerdns_version, [4, 1, 10])
# Bad auth
with requests_mock() as mock:
@@ -157,13 +157,14 @@ class TestPowerDnsProvider(TestCase):
# Non-existent zone in PowerDNS >=4.2.0 doesn't populate anything
provider._powerdns_version = [4, 2, 0]
with requests_mock() as mock:
mock.get(ANY, status_code=404, text='Not Found')
provider.configure_for_version("4.2.0")
zone = Zone('unit.tests.', [])
provider.populate(zone)
self.assertEquals(set(), zone.records)
provider.configure_for_version("4.1.0")
provider._powerdns_version = [4, 1, 0]
# The rest of this is messy/complicated b/c it's dealing with mocking
@@ -219,8 +220,8 @@ class TestPowerDnsProvider(TestCase):
self.assertEquals(expected_n, provider.apply(plan))
self.assertFalse(plan.exists)
provider._powerdns_version = [4, 2, 0]
with requests_mock() as mock:
provider.configure_for_version('4.2.0')
# get 404's, unknown zone
mock.get(ANY, status_code=404, text='')
# patch 404's, unknown zone
@@ -232,8 +233,8 @@ class TestPowerDnsProvider(TestCase):
self.assertEquals(expected_n, len(plan.changes))
self.assertEquals(expected_n, provider.apply(plan))
self.assertFalse(plan.exists)
provider.configure_for_version('4.1.0')
provider._powerdns_version = [4, 1, 0]
with requests_mock() as mock:
# get 422's, unknown zone
mock.get(ANY, status_code=422, text=dumps(not_found))