mirror of
https://github.com/certbot/certbot.git
synced 2026-07-30 16:14:44 +02:00
Migrate certbot-dns-cloudflare to cloudflare 4.x SDK (#10587)
## Summary - Migrate `certbot-dns-cloudflare` from the archived `python-cloudflare` 2.x library (`import CloudFlare`) to the actively maintained Cloudflare Python SDK 4.x (`import cloudflare`) - Update all API calls to the new SDK surface: `dns.records.create/list/delete`, `zones.list`, typed response objects instead of dicts - Replace `CloudFlare.exceptions.CloudFlareAPIError` with `cloudflare.APIStatusError` and extract CF error codes from `response.json()` - Bump dependency from `cloudflare>=2.19, <2.20` to `cloudflare>=4.0` - Update oldest pinning from `cloudflare 2.19` to `4.0.0` - Update all test mocks and assertions accordingly Fixes #9938 ## API Migration | Operation | Old 2.x | New 4.x | |---|---|---| | Import | `import CloudFlare` | `import cloudflare` | | Client (token) | `CloudFlare.CloudFlare(token=t)` | `cloudflare.Cloudflare(api_token=t)` | | Client (key) | `CloudFlare.CloudFlare(email, key)` | `cloudflare.Cloudflare(api_email=e, api_key=k)` | | List zones | `cf.zones.get(params={...})` → `list[dict]` | `cf.zones.list(name=n)` → iterable of Zone objects | | Create record | `cf.zones.dns_records.post(zone_id, data={...})` | `cf.dns.records.create(zone_id=id, **data)` | | List records | `cf.zones.dns_records.get(zone_id, params={...})` | `cf.dns.records.list(zone_id=id, type=..., ...)` | | Delete record | `cf.zones.dns_records.delete(zone_id, record_id)` | `cf.dns.records.delete(dns_record_id=rid, zone_id=zid)` | | Exceptions | `CloudFlare.exceptions.CloudFlareAPIError` | `cloudflare.APIStatusError` | ## Test plan - [x] All 20 existing tests pass with updated mocks - [x] Credentials INI file format is unchanged — no user-facing config changes - [x] Live dry-run renewal tested successfully across 5 domains --------- Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com> Co-authored-by: Ember <BeigeBox@users.noreply.github.com> Co-authored-by: Brad Warren <bmw@users.noreply.github.com>
This commit is contained in:
Mike Fara
co-authored by
Claude Sonnet 4.6
Ember
Brad Warren
Claude Sonnet 4.6
Ember
Brad Warren
parent
9ec4105ff3
commit
c8ebcb49bd
@@ -1,11 +1,21 @@
|
||||
"""DNS Authenticator for Cloudflare."""
|
||||
import logging
|
||||
import warnings
|
||||
from typing import Any
|
||||
from typing import Callable
|
||||
from typing import Literal
|
||||
from typing import Optional
|
||||
from typing import cast
|
||||
from typing import TypedDict
|
||||
|
||||
import CloudFlare
|
||||
# cloudflare 4.x includes a pydantic v1 compatibility shim that emits a
|
||||
# UserWarning on Python 3.14+. Suppress it here so that this internal-detail
|
||||
# warning is not shown to users during plugin discovery. In our test suite it
|
||||
# is filtered out via pytest.ini so it does not affect filterwarnings=error.
|
||||
with warnings.catch_warnings():
|
||||
warnings.filterwarnings('ignore', message='Core Pydantic V1 functionality',
|
||||
category=UserWarning)
|
||||
import cloudflare
|
||||
from cloudflare.types.zones import Zone
|
||||
|
||||
from certbot import errors
|
||||
from certbot.plugins import dns_common
|
||||
@@ -95,15 +105,15 @@ class _CloudflareClient:
|
||||
api_token: Optional[str] = None) -> None:
|
||||
if email:
|
||||
# If an email was specified, we're using an email/key combination and not a token.
|
||||
# We can't use named arguments in this case, as it would break compatibility with
|
||||
# the Cloudflare library since version 2.10.1, as the `token` argument was used for
|
||||
# tokens and keys alike and the `key` argument did not exist in earlier versions.
|
||||
self.cf = CloudFlare.CloudFlare(email, api_key)
|
||||
# We use named arguments here to match the cloudflare 4.x SDK's explicit parameter
|
||||
# names (api_email and api_key), which correspond to the Global API Key credentials
|
||||
# found in the Cloudflare dashboard under My Profile > API Tokens.
|
||||
self.cf = cloudflare.Cloudflare(api_email=email, api_key=api_key)
|
||||
else:
|
||||
# If no email was specified, we're using just a token. Let's use the named argument
|
||||
# for simplicity, which is compatible with all (current) versions of the Cloudflare
|
||||
# library.
|
||||
self.cf = CloudFlare.CloudFlare(token=api_token)
|
||||
# If no email was specified, we're using just an API token. We use the named argument
|
||||
# for clarity. API Tokens are the recommended authentication method as they support
|
||||
# fine-grained permissions scoped to specific zones and operations.
|
||||
self.cf = cloudflare.Cloudflare(api_token=api_token)
|
||||
|
||||
def add_txt_record(self, domain: str, record_name: str, record_content: str,
|
||||
record_ttl: int) -> None:
|
||||
@@ -119,24 +129,30 @@ class _CloudflareClient:
|
||||
|
||||
zone_id = self._find_zone_id(domain)
|
||||
|
||||
data = {'type': 'TXT',
|
||||
'name': record_name,
|
||||
'content': record_content,
|
||||
'ttl': record_ttl}
|
||||
data: _RecordData = {
|
||||
'type': 'TXT',
|
||||
'name': record_name,
|
||||
'content': record_content,
|
||||
'ttl': record_ttl,
|
||||
}
|
||||
|
||||
try:
|
||||
logger.debug('Attempting to add record to zone %s: %s', zone_id, data)
|
||||
self.cf.zones.dns_records.post(zone_id, data=data) # zones | pylint: disable=no-member
|
||||
except CloudFlare.exceptions.CloudFlareAPIError as e:
|
||||
code = int(e)
|
||||
self.cf.dns.records.create(zone_id=zone_id, **data)
|
||||
except cloudflare.APIStatusError as e:
|
||||
code = _cf_error_code(e)
|
||||
hint = None
|
||||
|
||||
if code == 1009:
|
||||
hint = 'Does your API token have "Zone:DNS:Edit" permissions?'
|
||||
|
||||
logger.error('Encountered CloudFlareAPIError adding TXT record: %d %s', e, e)
|
||||
logger.error('Encountered Cloudflare API error adding TXT record: %s', e)
|
||||
raise errors.PluginError('Error communicating with the Cloudflare API: {0}{1}'
|
||||
.format(e, ' ({0})'.format(hint) if hint else ''))
|
||||
except cloudflare.APIConnectionError as e:
|
||||
logger.error('Network error talking to the Cloudflare API: %s', e)
|
||||
raise errors.PluginError('Network error communicating with the Cloudflare API '
|
||||
'while adding a TXT record: {0}'.format(e))
|
||||
|
||||
record_id = self._find_txt_record_id(zone_id, record_name, record_content)
|
||||
logger.debug('Successfully added TXT record with record_id: %s', record_id)
|
||||
@@ -159,52 +175,48 @@ class _CloudflareClient:
|
||||
zone_id = self._find_zone_id(domain)
|
||||
except errors.PluginError as e:
|
||||
logger.debug('Encountered error finding zone_id during deletion: %s', e)
|
||||
logger.debug('Zone not found; no cleanup needed.')
|
||||
return
|
||||
|
||||
if zone_id:
|
||||
record_id = self._find_txt_record_id(zone_id, record_name, record_content)
|
||||
if record_id:
|
||||
try:
|
||||
# zones | pylint: disable=no-member
|
||||
self.cf.zones.dns_records.delete(zone_id, record_id)
|
||||
logger.debug('Successfully deleted TXT record.')
|
||||
except CloudFlare.exceptions.CloudFlareAPIError as e:
|
||||
logger.warning('Encountered CloudFlareAPIError deleting TXT record: %s', e)
|
||||
else:
|
||||
logger.debug('TXT record not found; no cleanup needed.')
|
||||
record_id = self._find_txt_record_id(zone_id, record_name, record_content)
|
||||
if record_id:
|
||||
try:
|
||||
self.cf.dns.records.delete(dns_record_id=record_id, zone_id=zone_id)
|
||||
logger.debug('Successfully deleted TXT record.')
|
||||
except cloudflare.APIStatusError as e:
|
||||
logger.warning('Encountered Cloudflare API error deleting TXT record: %s', e)
|
||||
except cloudflare.APIConnectionError as e:
|
||||
logger.warning('Network error deleting TXT record from Cloudflare: %s', e)
|
||||
else:
|
||||
logger.debug('Zone not found; no cleanup needed.')
|
||||
logger.debug('TXT record not found; no cleanup needed.')
|
||||
|
||||
def _find_zone_id(self, domain: str) -> str:
|
||||
"""
|
||||
Find the zone_id for a given domain.
|
||||
|
||||
:param str domain: The domain for which to find the zone_id.
|
||||
:returns: The zone_id, if found.
|
||||
:returns: The zone_id for the first matching zone that has a non-empty
|
||||
identifier. A zone with an empty/invalid id is treated as if no zone
|
||||
were found, so this method never returns an empty string.
|
||||
:rtype: str
|
||||
:raises certbot.errors.PluginError: if no zone_id is found.
|
||||
"""
|
||||
|
||||
zone_name_guesses = dns_common.base_domain_name_guesses(domain)
|
||||
zones: list[dict[str, Any]] = []
|
||||
zone: Zone | None = None
|
||||
code = msg = None
|
||||
|
||||
for zone_name in zone_name_guesses:
|
||||
params = {'name': zone_name,
|
||||
'per_page': 1}
|
||||
|
||||
try:
|
||||
zones = self.cf.zones.get(params=params) # zones | pylint: disable=no-member
|
||||
except CloudFlare.exceptions.CloudFlareAPIError as e:
|
||||
code = int(e)
|
||||
zone = next(iter(self.cf.zones.list(name=zone_name, per_page=1)), None)
|
||||
except cloudflare.APIStatusError as e:
|
||||
code = _cf_error_code(e)
|
||||
msg = str(e)
|
||||
hint = None
|
||||
|
||||
if code == 6003:
|
||||
hint = ('Did you copy your entire API token/key? To use Cloudflare tokens, '
|
||||
'you\'ll need the python package cloudflare>=2.3.1.{}'
|
||||
.format(' This certbot is running cloudflare ' + str(CloudFlare.__version__)
|
||||
if hasattr(CloudFlare, '__version__') else ''))
|
||||
hint = ('Did you copy your entire API token/key? '
|
||||
'See {} to manage your API tokens.'.format(ACCOUNT_URL))
|
||||
elif code == 9103:
|
||||
hint = 'Did you enter the correct email address and Global key?'
|
||||
elif code == 9109:
|
||||
@@ -215,13 +227,19 @@ class _CloudflareClient:
|
||||
'that you have supplied valid Cloudflare API credentials. ({2})'
|
||||
.format(code, msg, hint))
|
||||
else:
|
||||
logger.debug('Unrecognised CloudFlareAPIError while finding zone_id: %d %s. '
|
||||
'Continuing with next zone guess...', e, e)
|
||||
logger.debug('Unrecognised Cloudflare API error while finding zone_id: %s. '
|
||||
'Continuing with next zone guess...', e)
|
||||
except cloudflare.APIConnectionError as e:
|
||||
raise errors.PluginError('Network error contacting the Cloudflare API while '
|
||||
'looking up the zone for {0}: {1}'.format(domain, e))
|
||||
|
||||
if zones:
|
||||
zone_id: str = zones[0]['id']
|
||||
logger.debug('Found zone_id of %s for %s using name %s', zone_id, domain, zone_name)
|
||||
return zone_id
|
||||
if zone:
|
||||
zone_id = zone.id
|
||||
if zone_id:
|
||||
logger.debug('Found zone_id of %s for %s using name %s',
|
||||
zone_id, domain, zone_name)
|
||||
return zone_id
|
||||
break # Found a zone but it has no usable ID; stop searching
|
||||
|
||||
if msg is not None:
|
||||
if 'com.cloudflare.api.account.zone.list' in msg:
|
||||
@@ -252,20 +270,38 @@ class _CloudflareClient:
|
||||
:rtype: str
|
||||
"""
|
||||
|
||||
params = {'type': 'TXT',
|
||||
'name': record_name,
|
||||
'content': record_content,
|
||||
'per_page': 1}
|
||||
try:
|
||||
# zones | pylint: disable=no-member
|
||||
records = self.cf.zones.dns_records.get(zone_id, params=params)
|
||||
except CloudFlare.exceptions.CloudFlareAPIError as e:
|
||||
logger.debug('Encountered CloudFlareAPIError getting TXT record_id: %s', e)
|
||||
records = list(self.cf.dns.records.list(
|
||||
zone_id=zone_id, type='TXT', name={'exact': record_name},
|
||||
content={'exact': record_content}, per_page=1))
|
||||
except cloudflare.APIStatusError as e:
|
||||
logger.debug('Encountered Cloudflare API error getting TXT record_id: %s', e)
|
||||
records = []
|
||||
except cloudflare.APIConnectionError as e:
|
||||
logger.debug('Network error getting TXT record_id from Cloudflare: %s', e)
|
||||
records = []
|
||||
|
||||
if records:
|
||||
# Cleanup is returning the system to the state we found it. If, for some reason,
|
||||
# there are multiple matching records, we only delete one because we only added one.
|
||||
return cast(str, records[0]['id'])
|
||||
return records[0].id
|
||||
logger.debug('Unable to find TXT record.')
|
||||
return None
|
||||
|
||||
|
||||
class _RecordData(TypedDict):
|
||||
"""Offers type hints for dictionaries of Cloudflare API parameters."""
|
||||
|
||||
type: Literal['TXT']
|
||||
name: str
|
||||
content: str
|
||||
ttl: int
|
||||
|
||||
|
||||
def _cf_error_code(e: cloudflare.APIStatusError) -> int | None:
|
||||
"""Extract the first Cloudflare error code from an API error response."""
|
||||
try:
|
||||
body = e.response.json()
|
||||
return int(body['errors'][0]['code'])
|
||||
except (ValueError, KeyError, IndexError, TypeError): # pragma: no cover
|
||||
return None
|
||||
|
||||
+135
-62
@@ -2,9 +2,10 @@
|
||||
|
||||
import sys
|
||||
import unittest
|
||||
from typing import Any
|
||||
from unittest import mock
|
||||
|
||||
import CloudFlare
|
||||
import cloudflare
|
||||
import pytest
|
||||
|
||||
from certbot import errors
|
||||
@@ -13,7 +14,23 @@ from certbot.plugins import dns_test_common
|
||||
from certbot.plugins.dns_test_common import DOMAIN
|
||||
from certbot.tests import util as test_util
|
||||
|
||||
API_ERROR = CloudFlare.exceptions.CloudFlareAPIError(1000, '', '')
|
||||
|
||||
def _make_api_error(cf_code: int, msg: str = '', http_status: int = 400
|
||||
) -> cloudflare.APIStatusError:
|
||||
"""Build a cloudflare.APIStatusError with a Cloudflare error code in the body."""
|
||||
body: Any = {'success': False, 'errors': [{'code': cf_code, 'message': msg}]}
|
||||
response = mock.Mock(status_code=http_status, request=mock.Mock())
|
||||
response.json.return_value = body
|
||||
return cloudflare.APIStatusError(message=msg or str(cf_code), response=response, body=body)
|
||||
|
||||
|
||||
def _make_connection_error(msg: str = 'Connection error.') -> cloudflare.APIConnectionError:
|
||||
"""Build a cloudflare.APIConnectionError (e.g. transient network failure)."""
|
||||
return cloudflare.APIConnectionError(message=msg, request=mock.Mock())
|
||||
|
||||
|
||||
API_ERROR = _make_api_error(1000)
|
||||
CONNECTION_ERROR = _make_connection_error()
|
||||
|
||||
API_TOKEN = 'an-api-token'
|
||||
|
||||
@@ -96,13 +113,43 @@ class AuthenticatorTest(test_util.TempDirTestCase, dns_test_common.BaseAuthentic
|
||||
with pytest.raises(errors.PluginError):
|
||||
self.auth.perform([self.achall])
|
||||
|
||||
def test_get_cloudflare_client_with_api_token(self):
|
||||
from certbot_dns_cloudflare._internal.dns_cloudflare import Authenticator
|
||||
from certbot_dns_cloudflare._internal.dns_cloudflare import _CloudflareClient
|
||||
mock_auth = mock.MagicMock()
|
||||
mock_auth.credentials.conf.return_value = API_TOKEN
|
||||
client = Authenticator._get_cloudflare_client(mock_auth)
|
||||
self.assertIsInstance(client, _CloudflareClient)
|
||||
|
||||
def test_get_cloudflare_client_with_email_key(self):
|
||||
from certbot_dns_cloudflare._internal.dns_cloudflare import Authenticator
|
||||
from certbot_dns_cloudflare._internal.dns_cloudflare import _CloudflareClient
|
||||
mock_auth = mock.MagicMock()
|
||||
mock_auth.credentials.conf.side_effect = lambda k: None if k == 'api-token' else 'some_value'
|
||||
client = Authenticator._get_cloudflare_client(mock_auth)
|
||||
self.assertIsInstance(client, _CloudflareClient)
|
||||
|
||||
|
||||
def _mock_zone(zone_id: str | None) -> mock.MagicMock:
|
||||
"""Create a mock zone object with an .id attribute."""
|
||||
zone = mock.MagicMock()
|
||||
zone.id = zone_id
|
||||
return zone
|
||||
|
||||
|
||||
def _mock_record(record_id: str | None) -> mock.MagicMock:
|
||||
"""Create a mock DNS record object with an .id attribute."""
|
||||
record = mock.MagicMock()
|
||||
record.id = record_id
|
||||
return record
|
||||
|
||||
|
||||
class CloudflareClientTest(unittest.TestCase):
|
||||
record_name = "foo"
|
||||
record_content = "bar"
|
||||
record_ttl = 42
|
||||
zone_id = 1
|
||||
record_id = 2
|
||||
zone_id = "zone-id-1"
|
||||
record_id = "record-id-2"
|
||||
|
||||
def setUp(self):
|
||||
from certbot_dns_cloudflare._internal.dns_cloudflare import _CloudflareClient
|
||||
@@ -113,119 +160,145 @@ class CloudflareClientTest(unittest.TestCase):
|
||||
self.cloudflare_client.cf = self.cf
|
||||
|
||||
def test_add_txt_record(self):
|
||||
self.cf.zones.get.return_value = [{'id': self.zone_id}]
|
||||
self.cf.zones.list.return_value = [_mock_zone(self.zone_id)]
|
||||
|
||||
self.cloudflare_client.add_txt_record(DOMAIN, self.record_name, self.record_content,
|
||||
self.record_ttl)
|
||||
|
||||
self.cf.zones.dns_records.post.assert_called_with(self.zone_id, data=mock.ANY)
|
||||
|
||||
post_data = self.cf.zones.dns_records.post.call_args[1]['data']
|
||||
|
||||
assert 'TXT' == post_data['type']
|
||||
assert self.record_name == post_data['name']
|
||||
assert self.record_content == post_data['content']
|
||||
assert self.record_ttl == post_data['ttl']
|
||||
self.cf.dns.records.create.assert_called_with(
|
||||
zone_id=self.zone_id, type='TXT', name=self.record_name,
|
||||
content=self.record_content, ttl=self.record_ttl)
|
||||
|
||||
def test_add_txt_record_error(self):
|
||||
self.cf.zones.get.return_value = [{'id': self.zone_id}]
|
||||
self.cf.zones.list.return_value = [_mock_zone(self.zone_id)]
|
||||
|
||||
self.cf.zones.dns_records.post.side_effect = CloudFlare.exceptions.CloudFlareAPIError(1009, '', '')
|
||||
self.cf.dns.records.create.side_effect = _make_api_error(1009)
|
||||
|
||||
with pytest.raises(errors.PluginError):
|
||||
self.cloudflare_client.add_txt_record(DOMAIN, self.record_name, self.record_content, self.record_ttl)
|
||||
self.cloudflare_client.add_txt_record(DOMAIN, self.record_name, self.record_content,
|
||||
self.record_ttl)
|
||||
|
||||
def test_add_txt_record_error_during_zone_lookup(self):
|
||||
self.cf.zones.get.side_effect = API_ERROR
|
||||
self.cf.zones.list.side_effect = API_ERROR
|
||||
|
||||
with pytest.raises(errors.PluginError):
|
||||
self.cloudflare_client.add_txt_record(DOMAIN, self.record_name, self.record_content, self.record_ttl)
|
||||
self.cloudflare_client.add_txt_record(DOMAIN, self.record_name, self.record_content,
|
||||
self.record_ttl)
|
||||
|
||||
def test_add_txt_record_zone_not_found(self):
|
||||
self.cf.zones.get.return_value = []
|
||||
self.cf.zones.list.return_value = []
|
||||
|
||||
with pytest.raises(errors.PluginError):
|
||||
self.cloudflare_client.add_txt_record(DOMAIN, self.record_name, self.record_content, self.record_ttl)
|
||||
self.cloudflare_client.add_txt_record(DOMAIN, self.record_name, self.record_content,
|
||||
self.record_ttl)
|
||||
|
||||
def test_add_txt_record_connection_error_on_create(self):
|
||||
self.cf.zones.list.return_value = [_mock_zone(self.zone_id)]
|
||||
self.cf.dns.records.create.side_effect = CONNECTION_ERROR
|
||||
|
||||
with pytest.raises(errors.PluginError, match='Network error'):
|
||||
self.cloudflare_client.add_txt_record(DOMAIN, self.record_name, self.record_content,
|
||||
self.record_ttl)
|
||||
|
||||
def test_add_txt_record_connection_error_during_zone_lookup(self):
|
||||
self.cf.zones.list.side_effect = CONNECTION_ERROR
|
||||
|
||||
with pytest.raises(errors.PluginError, match='Network error'):
|
||||
self.cloudflare_client.add_txt_record(DOMAIN, self.record_name, self.record_content,
|
||||
self.record_ttl)
|
||||
|
||||
def test_add_txt_record_bad_creds(self):
|
||||
self.cf.zones.get.side_effect = CloudFlare.exceptions.CloudFlareAPIError(6003, '', '')
|
||||
self.cf.zones.list.side_effect = _make_api_error(6003)
|
||||
with pytest.raises(errors.PluginError):
|
||||
self.cloudflare_client.add_txt_record(DOMAIN, self.record_name, self.record_content, self.record_ttl)
|
||||
self.cloudflare_client.add_txt_record(DOMAIN, self.record_name, self.record_content,
|
||||
self.record_ttl)
|
||||
|
||||
self.cf.zones.get.side_effect = CloudFlare.exceptions.CloudFlareAPIError(9103, '', '')
|
||||
self.cf.zones.list.side_effect = _make_api_error(9103)
|
||||
with pytest.raises(errors.PluginError):
|
||||
self.cloudflare_client.add_txt_record(DOMAIN, self.record_name, self.record_content, self.record_ttl)
|
||||
self.cloudflare_client.add_txt_record(DOMAIN, self.record_name, self.record_content,
|
||||
self.record_ttl)
|
||||
|
||||
self.cf.zones.get.side_effect = CloudFlare.exceptions.CloudFlareAPIError(9109, '', '')
|
||||
self.cf.zones.list.side_effect = _make_api_error(9109)
|
||||
with pytest.raises(errors.PluginError):
|
||||
self.cloudflare_client.add_txt_record(DOMAIN, self.record_name, self.record_content, self.record_ttl)
|
||||
self.cloudflare_client.add_txt_record(DOMAIN, self.record_name, self.record_content,
|
||||
self.record_ttl)
|
||||
|
||||
self.cf.zones.get.side_effect = CloudFlare.exceptions.CloudFlareAPIError(0, 'com.cloudflare.api.account.zone.list', '')
|
||||
self.cf.zones.list.side_effect = _make_api_error(0, 'com.cloudflare.api.account.zone.list')
|
||||
with pytest.raises(errors.PluginError):
|
||||
self.cloudflare_client.add_txt_record(DOMAIN, self.record_name, self.record_content, self.record_ttl)
|
||||
self.cloudflare_client.add_txt_record(DOMAIN, self.record_name, self.record_content,
|
||||
self.record_ttl)
|
||||
|
||||
def test_del_txt_record(self):
|
||||
self.cf.zones.get.return_value = [{'id': self.zone_id}]
|
||||
self.cf.zones.dns_records.get.return_value = [{'id': self.record_id}]
|
||||
self.cf.zones.list.return_value = [_mock_zone(self.zone_id)]
|
||||
self.cf.dns.records.list.return_value = [_mock_record(self.record_id)]
|
||||
|
||||
self.cloudflare_client.del_txt_record(DOMAIN, self.record_name, self.record_content)
|
||||
|
||||
expected = [mock.call.zones.get(params=mock.ANY),
|
||||
mock.call.zones.dns_records.get(self.zone_id, params=mock.ANY),
|
||||
mock.call.zones.dns_records.delete(self.zone_id, self.record_id)]
|
||||
|
||||
assert expected == self.cf.mock_calls
|
||||
|
||||
get_data = self.cf.zones.dns_records.get.call_args[1]['params']
|
||||
|
||||
assert 'TXT' == get_data['type']
|
||||
assert self.record_name == get_data['name']
|
||||
assert self.record_content == get_data['content']
|
||||
self.cf.zones.list.assert_called_once()
|
||||
self.cf.dns.records.list.assert_called_once_with(
|
||||
zone_id=self.zone_id, type='TXT', name={'exact': self.record_name},
|
||||
content={'exact': self.record_content}, per_page=1)
|
||||
self.cf.dns.records.delete.assert_called_once_with(
|
||||
dns_record_id=self.record_id, zone_id=self.zone_id)
|
||||
|
||||
def test_del_txt_record_error_during_zone_lookup(self):
|
||||
self.cf.zones.get.side_effect = API_ERROR
|
||||
self.cf.zones.list.side_effect = API_ERROR
|
||||
|
||||
self.cloudflare_client.del_txt_record(DOMAIN, self.record_name, self.record_content)
|
||||
|
||||
def test_del_txt_record_error_during_delete(self):
|
||||
self.cf.zones.get.return_value = [{'id': self.zone_id}]
|
||||
self.cf.zones.dns_records.get.return_value = [{'id': self.record_id}]
|
||||
self.cf.zones.dns_records.delete.side_effect = API_ERROR
|
||||
self.cf.zones.list.return_value = [_mock_zone(self.zone_id)]
|
||||
self.cf.dns.records.list.return_value = [_mock_record(self.record_id)]
|
||||
self.cf.dns.records.delete.side_effect = API_ERROR
|
||||
|
||||
self.cloudflare_client.del_txt_record(DOMAIN, self.record_name, self.record_content)
|
||||
expected = [mock.call.zones.get(params=mock.ANY),
|
||||
mock.call.zones.dns_records.get(self.zone_id, params=mock.ANY),
|
||||
mock.call.zones.dns_records.delete(self.zone_id, self.record_id)]
|
||||
|
||||
assert expected == self.cf.mock_calls
|
||||
self.cf.dns.records.delete.assert_called_once_with(
|
||||
dns_record_id=self.record_id, zone_id=self.zone_id)
|
||||
|
||||
def test_del_txt_record_error_during_get(self):
|
||||
self.cf.zones.get.return_value = [{'id': self.zone_id}]
|
||||
self.cf.zones.dns_records.get.side_effect = API_ERROR
|
||||
self.cf.zones.list.return_value = [_mock_zone(self.zone_id)]
|
||||
self.cf.dns.records.list.side_effect = API_ERROR
|
||||
|
||||
self.cloudflare_client.del_txt_record(DOMAIN, self.record_name, self.record_content)
|
||||
expected = [mock.call.zones.get(params=mock.ANY),
|
||||
mock.call.zones.dns_records.get(self.zone_id, params=mock.ANY)]
|
||||
|
||||
assert expected == self.cf.mock_calls
|
||||
self.cf.dns.records.list.assert_called_once()
|
||||
self.cf.dns.records.delete.assert_not_called()
|
||||
|
||||
def test_del_txt_record_no_record(self):
|
||||
self.cf.zones.get.return_value = [{'id': self.zone_id}]
|
||||
self.cf.zones.dns_records.get.return_value = []
|
||||
self.cf.zones.list.return_value = [_mock_zone(self.zone_id)]
|
||||
self.cf.dns.records.list.return_value = []
|
||||
|
||||
self.cloudflare_client.del_txt_record(DOMAIN, self.record_name, self.record_content)
|
||||
expected = [mock.call.zones.get(params=mock.ANY),
|
||||
mock.call.zones.dns_records.get(self.zone_id, params=mock.ANY)]
|
||||
|
||||
assert expected == self.cf.mock_calls
|
||||
self.cf.dns.records.list.assert_called_once()
|
||||
self.cf.dns.records.delete.assert_not_called()
|
||||
|
||||
def test_del_txt_record_no_zone(self):
|
||||
self.cf.zones.get.return_value = [{'id': None}]
|
||||
self.cf.zones.list.return_value = [_mock_zone(None)]
|
||||
|
||||
self.cloudflare_client.del_txt_record(DOMAIN, self.record_name, self.record_content)
|
||||
expected = [mock.call.zones.get(params=mock.ANY)]
|
||||
|
||||
assert expected == self.cf.mock_calls
|
||||
self.cf.zones.list.assert_called_once()
|
||||
|
||||
def test_del_txt_record_connection_error_on_delete(self):
|
||||
self.cf.zones.list.return_value = [_mock_zone(self.zone_id)]
|
||||
self.cf.dns.records.list.return_value = [_mock_record(self.record_id)]
|
||||
self.cf.dns.records.delete.side_effect = CONNECTION_ERROR
|
||||
|
||||
self.cloudflare_client.del_txt_record(DOMAIN, self.record_name, self.record_content)
|
||||
|
||||
self.cf.dns.records.delete.assert_called_once_with(
|
||||
dns_record_id=self.record_id, zone_id=self.zone_id)
|
||||
|
||||
def test_del_txt_record_connection_error_on_get(self):
|
||||
self.cf.zones.list.return_value = [_mock_zone(self.zone_id)]
|
||||
self.cf.dns.records.list.side_effect = CONNECTION_ERROR
|
||||
|
||||
self.cloudflare_client.del_txt_record(DOMAIN, self.record_name, self.record_content)
|
||||
|
||||
self.cf.dns.records.list.assert_called_once()
|
||||
self.cf.dns.records.delete.assert_not_called()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user