From 2a159bf93bb38d9d1b3862841c9b390508de3524 Mon Sep 17 00:00:00 2001 From: Ross McFarland Date: Fri, 12 Jun 2020 09:36:57 -0700 Subject: [PATCH] Refactor PowerDNS version handling to be transparently cached properties --- octodns/provider/powerdns.py | 91 +++++++++++-------------- tests/test_octodns_provider_powerdns.py | 73 ++++++++++---------- 2 files changed, 75 insertions(+), 89 deletions(-) diff --git a/octodns/provider/powerdns.py b/octodns/provider/powerdns.py index c9b5e01..bcb6980 100644 --- a/octodns/provider/powerdns.py +++ b/octodns/provider/powerdns.py @@ -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, diff --git a/tests/test_octodns_provider_powerdns.py b/tests/test_octodns_provider_powerdns.py index 9ba28af..fd877ef 100644 --- a/tests/test_octodns_provider_powerdns.py +++ b/tests/test_octodns_provider_powerdns.py @@ -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))