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 remote-tracking branch 'origin/master' into extract-googlecloud

Browse Source
This commit is contained in:
Ross McFarland
2022-01-13 09:59:10 -08:00
parent c9a1be2f48 0ce60061e2
commit ecee50289b
16 changed files with 77 additions and 6945 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.github/workflows/main.yml vendored
View File

@ -7,7 +7,7 @@ jobs:
strategy:
matrix:
# Tested versions based on dates in https://devguide.python.org/devcycle/#end-of-life-branches,
python-version: [3.6, 3.7, 3.8, 3.9]
python-version: ['3.7', '3.8', '3.9']
steps:
- uses: actions/checkout@master
- name: Setup python

3
CHANGELOG.md
View File

@ -6,6 +6,7 @@
https://github.com/octodns/octodns/issues/622 &
https://github.com/octodns/octodns/pull/822 for more information. Providers
that have been extracted in this release include:
* [AzureProvider](https://github.com/octodns/octodns-azure/)
* [AkamaiProvider](https://github.com/octodns/octodns-edgedns/)
* [CloudflareProvider](https://github.com/octodns/octodns-cloudflare/)
* [ConstellixProvider](https://github.com/octodns/octodns-constellix/)
@ -15,6 +16,8 @@
* [DynProvider](https://github.com/octodns/octodns-dynprovider/)
* [EasyDnsProvider](https://github.com/octodns/octodns-easydns/)
* [EtcHostsProvider](https://github.com/octodns/octodns-etchosts/)
* [GandiProvider](https://github.com/octodns/octodns-gandi/)
* [GcoreProvider](https://github.com/octodns/octodns-gcore/)
* [GoogleCloudProvider](https://github.com/octodns/octodns-googlecloud/)
* [Ns1Provider](https://github.com/octodns/octodns-ns1/)
* [PowerDnsProvider](https://github.com/octodns/octodns-powerdns/)

6
README.md
View File

@ -192,7 +192,7 @@ The table below lists the providers octoDNS supports. We're currently in the pro
| Provider | Module | Requirements | Record Support | Dynamic | Notes |
|--|--|--|--|--|--|
| [AzureProvider](/octodns/provider/azuredns.py) | | azure-identity, azure-mgmt-dns, azure-mgmt-trafficmanager | A, AAAA, CAA, CNAME, MX, NS, PTR, SRV, TXT | Alpha (A, AAAA, CNAME) | |
| [AzureProvider](https://github.com/octodns/octodns-azure/) | [octodns_azure](https://github.com/octodns/octodns-azure/) | | | | |
| [AkamaiProvider](https://github.com/octodns/octodns-edgedns/) | [octodns_edgedns](https://github.com/octodns/octodns-edgedns/) | | | | |
| [CloudflareProvider](https://github.com/octodns/octodns-cloudflare/) | [octodns_cloudflare](https://github.com/octodns/octodns-cloudflare/) | | | | |
| [ConstellixProvider](https://github.com/octodns/octodns-constellix/) | [octodns_constellix](https://github.com/octodns/octodns-constellix/) | | | | |
@ -203,8 +203,8 @@ The table below lists the providers octoDNS supports. We're currently in the pro
| [EasyDnsProvider](https://github.com/octodns/octodns-easydns/) | [octodns_easydns](https://github.com/octodns/octodns-easydns/) | | | | |
| [EtcHostsProvider](https://github.com/octodns/octodns-etchosts/) | [octodns_etchosts](https://github.com/octodns/octodns-etchosts/) | | | | |
| [EnvVarSource](/octodns/source/envvar.py) | | | TXT | No | read-only environment variable injection |
| [GandiProvider](/octodns/provider/gandi.py) | | | A, AAAA, ALIAS, CAA, CNAME, DNAME, MX, NS, PTR, SPF, SRV, SSHFP, TXT | No | |
| [GCoreProvider](/octodns/provider/gcore.py) | | | A, AAAA, NS, MX, TXT, SRV, CNAME, PTR | Dynamic | |
| [GandiProvider](https://github.com/octodns/octodns-gandi/) | [octodns_gandi](https://github.com/octodns/octodns-gandi/) | | | | |
| [GCoreProvider](https://github.com/octodns/octodns-gcore/) | [octodns_gcore](https://github.com/octodns/octodns-gcore/) | | | | |
| [GoogleCloudProvider](https://github.com/octodns/octodns-googlecloud/) | [octodns_googlecloud](https://github.com/octodns/octodns-googlecloud/) | | | | |
| [HetznerProvider](/octodns/provider/hetzner.py) | | | A, AAAA, CAA, CNAME, MX, NS, SRV, TXT | No | |
| [MythicBeastsProvider](/octodns/provider/mythicbeasts.py) | | Mythic Beasts | A, AAAA, ALIAS, CNAME, MX, NS, SRV, SSHFP, CAA, TXT | No | |

1511
octodns/provider/azuredns.py
View File

File diff suppressed because it is too large Load Diff

386
octodns/provider/gandi.py
View File

@ -5,373 +5,19 @@
from __future__ import absolute_import, division, print_function, \
unicode_literals
from collections import defaultdict
from requests import Session
import logging
from ..record import Record
from . import ProviderException
from .base import BaseProvider
class GandiClientException(ProviderException):
pass
class GandiClientBadRequest(GandiClientException):
def __init__(self, r):
super(GandiClientBadRequest, self).__init__(r.text)
class GandiClientUnauthorized(GandiClientException):
def __init__(self, r):
super(GandiClientUnauthorized, self).__init__(r.text)
class GandiClientForbidden(GandiClientException):
def __init__(self, r):
super(GandiClientForbidden, self).__init__(r.text)
class GandiClientNotFound(GandiClientException):
def __init__(self, r):
super(GandiClientNotFound, self).__init__(r.text)
class GandiClientUnknownDomainName(GandiClientException):
def __init__(self, msg):
super(GandiClientUnknownDomainName, self).__init__(msg)
class GandiClient(object):
def __init__(self, token):
session = Session()
session.headers.update({'Authorization': f'Apikey {token}'})
self._session = session
self.endpoint = 'https://api.gandi.net/v5'
def _request(self, method, path, params={}, data=None):
url = f'{self.endpoint}{path}'
r = self._session.request(method, url, params=params, json=data)
if r.status_code == 400:
raise GandiClientBadRequest(r)
if r.status_code == 401:
raise GandiClientUnauthorized(r)
elif r.status_code == 403:
raise GandiClientForbidden(r)
elif r.status_code == 404:
raise GandiClientNotFound(r)
r.raise_for_status()
return r
def zone(self, zone_name):
return self._request('GET', f'/livedns/domains/{zone_name}').json()
def zone_create(self, zone_name):
return self._request('POST', '/livedns/domains', data={
'fqdn': zone_name,
'zone': {}
}).json()
def zone_records(self, zone_name):
records = self._request('GET',
f'/livedns/domains/{zone_name}/records').json()
for record in records:
if record['rrset_name'] == '@':
record['rrset_name'] = ''
# Change relative targets to absolute ones.
if record['rrset_type'] in ['ALIAS', 'CNAME', 'DNAME', 'MX',
'NS', 'SRV']:
for i, value in enumerate(record['rrset_values']):
if not value.endswith('.'):
record['rrset_values'][i] = f'{value}.{zone_name}.'
return records
def record_create(self, zone_name, data):
self._request('POST', f'/livedns/domains/{zone_name}/records',
data=data)
def record_delete(self, zone_name, record_name, record_type):
self._request('DELETE', f'/livedns/domains/{zone_name}/records/'
f'{record_name}/{record_type}')
class GandiProvider(BaseProvider):
'''
Gandi provider using API v5.
gandi:
class: octodns.provider.gandi.GandiProvider
# Your API key (required)
token: XXXXXXXXXXXX
'''
SUPPORTS_GEO = False
SUPPORTS_DYNAMIC = False
SUPPORTS = set((['A', 'AAAA', 'ALIAS', 'CAA', 'CNAME', 'DNAME',
'MX', 'NS', 'PTR', 'SPF', 'SRV', 'SSHFP', 'TXT']))
def __init__(self, id, token, *args, **kwargs):
self.log = logging.getLogger(f'GandiProvider[{id}]')
self.log.debug('__init__: id=%s, token=***', id)
super(GandiProvider, self).__init__(id, *args, **kwargs)
self._client = GandiClient(token)
self._zone_records = {}
def _data_for_multiple(self, _type, records):
return {
'ttl': records[0]['rrset_ttl'],
'type': _type,
'values': [v.replace(';', '\\;') for v in
records[0]['rrset_values']] if _type == 'TXT' else
records[0]['rrset_values']
}
_data_for_A = _data_for_multiple
_data_for_AAAA = _data_for_multiple
_data_for_TXT = _data_for_multiple
_data_for_SPF = _data_for_multiple
_data_for_NS = _data_for_multiple
def _data_for_CAA(self, _type, records):
values = []
for record in records[0]['rrset_values']:
flags, tag, value = record.split(' ')
values.append({
'flags': flags,
'tag': tag,
# Remove quotes around value.
'value': value[1:-1],
})
return {
'ttl': records[0]['rrset_ttl'],
'type': _type,
'values': values
}
def _data_for_single(self, _type, records):
return {
'ttl': records[0]['rrset_ttl'],
'type': _type,
'value': records[0]['rrset_values'][0]
}
_data_for_ALIAS = _data_for_single
_data_for_CNAME = _data_for_single
_data_for_DNAME = _data_for_single
_data_for_PTR = _data_for_single
def _data_for_MX(self, _type, records):
values = []
for record in records[0]['rrset_values']:
priority, server = record.split(' ')
values.append({
'preference': priority,
'exchange': server
})
return {
'ttl': records[0]['rrset_ttl'],
'type': _type,
'values': values
}
def _data_for_SRV(self, _type, records):
values = []
for record in records[0]['rrset_values']:
priority, weight, port, target = record.split(' ', 3)
values.append({
'priority': priority,
'weight': weight,
'port': port,
'target': target
})
return {
'ttl': records[0]['rrset_ttl'],
'type': _type,
'values': values
}
def _data_for_SSHFP(self, _type, records):
values = []
for record in records[0]['rrset_values']:
algorithm, fingerprint_type, fingerprint = record.split(' ', 2)
values.append({
'algorithm': algorithm,
'fingerprint': fingerprint,
'fingerprint_type': fingerprint_type
})
return {
'ttl': records[0]['rrset_ttl'],
'type': _type,
'values': values
}
def zone_records(self, zone):
if zone.name not in self._zone_records:
try:
self._zone_records[zone.name] = \
self._client.zone_records(zone.name[:-1])
except GandiClientNotFound:
return []
return self._zone_records[zone.name]
def populate(self, zone, target=False, lenient=False):
self.log.debug('populate: name=%s, target=%s, lenient=%s', zone.name,
target, lenient)
values = defaultdict(lambda: defaultdict(list))
for record in self.zone_records(zone):
_type = record['rrset_type']
if _type not in self.SUPPORTS:
continue
values[record['rrset_name']][record['rrset_type']].append(record)
before = len(zone.records)
for name, types in values.items():
for _type, records in types.items():
data_for = getattr(self, f'_data_for_{_type}')
record = Record.new(zone, name, data_for(_type, records),
source=self, lenient=lenient)
zone.add_record(record, lenient=lenient)
exists = zone.name in self._zone_records
self.log.info('populate: found %s records, exists=%s',
len(zone.records) - before, exists)
return exists
def _record_name(self, name):
return name if name else '@'
def _params_for_multiple(self, record):
return {
'rrset_name': self._record_name(record.name),
'rrset_ttl': record.ttl,
'rrset_type': record._type,
'rrset_values': [v.replace('\\;', ';') for v in
record.values] if record._type == 'TXT'
else record.values
}
_params_for_A = _params_for_multiple
_params_for_AAAA = _params_for_multiple
_params_for_NS = _params_for_multiple
_params_for_TXT = _params_for_multiple
_params_for_SPF = _params_for_multiple
def _params_for_CAA(self, record):
return {
'rrset_name': self._record_name(record.name),
'rrset_ttl': record.ttl,
'rrset_type': record._type,
'rrset_values': [f'{v.flags} {v.tag} "{v.value}"'
for v in record.values]
}
def _params_for_single(self, record):
return {
'rrset_name': self._record_name(record.name),
'rrset_ttl': record.ttl,
'rrset_type': record._type,
'rrset_values': [record.value]
}
_params_for_ALIAS = _params_for_single
_params_for_CNAME = _params_for_single
_params_for_DNAME = _params_for_single
_params_for_PTR = _params_for_single
def _params_for_MX(self, record):
return {
'rrset_name': self._record_name(record.name),
'rrset_ttl': record.ttl,
'rrset_type': record._type,
'rrset_values': [f'{v.preference} {v.exchange}'
for v in record.values]
}
def _params_for_SRV(self, record):
return {
'rrset_name': self._record_name(record.name),
'rrset_ttl': record.ttl,
'rrset_type': record._type,
'rrset_values': [f'{v.priority} {v.weight} {v.port} {v.target}'
for v in record.values]
}
def _params_for_SSHFP(self, record):
return {
'rrset_name': self._record_name(record.name),
'rrset_ttl': record.ttl,
'rrset_type': record._type,
'rrset_values': [f'{v.algorithm} {v.fingerprint_type} '
f'{v.fingerprint}' for v in record.values]
}
def _apply_create(self, change):
new = change.new
data = getattr(self, f'_params_for_{new._type}')(new)
self._client.record_create(new.zone.name[:-1], data)
def _apply_update(self, change):
self._apply_delete(change)
self._apply_create(change)
def _apply_delete(self, change):
existing = change.existing
zone = existing.zone
self._client.record_delete(zone.name[:-1],
self._record_name(existing.name),
existing._type)
def _apply(self, plan):
desired = plan.desired
changes = plan.changes
zone = desired.name[:-1]
self.log.debug('_apply: zone=%s, len(changes)=%d', desired.name,
len(changes))
try:
self._client.zone(zone)
except GandiClientNotFound:
self.log.info('_apply: no existing zone, trying to create it')
try:
self._client.zone_create(zone)
self.log.info('_apply: zone has been successfully created')
except GandiClientNotFound:
# We suppress existing exception before raising
# GandiClientUnknownDomainName.
e = GandiClientUnknownDomainName('This domain is not '
'registered at Gandi. '
'Please register or '
'transfer it here '
'to be able to manage its '
'DNS zone.')
e.__cause__ = None
raise e
# Force records deletion to be done before creation in order to avoid
# "CNAME record must be the only record" error when an existing CNAME
# record is replaced by an A/AAAA record.
changes.reverse()
for change in changes:
class_name = change.__class__.__name__
getattr(self, f'_apply_{class_name.lower()}')(change)
# Clear out the cache if any
self._zone_records.pop(desired.name, None)
from logging import getLogger
logger = getLogger('Gandi')
try:
logger.warn('octodns_gandi shimmed. Update your provider class to '
'octodns_gandi.GandiProvider. '
'Shim will be removed in 1.0')
from octodns_gandi import GandiProvider
GandiProvider # pragma: no cover
except ModuleNotFoundError:
logger.exception('GandiProvider has been moved into a seperate module, '
'octodns_gandi is now required. Provider class should '
'be updated to octodns_gandi.GandiProvider. See '
'https://github.com/octodns/octodns/README.md#updating-'
'to-use-extracted-providers for more information.')
raise

629
octodns/provider/gcore.py
View File

@ -2,615 +2,20 @@
#
#
from __future__ import (
absolute_import,
division,
print_function,
unicode_literals,
)
from collections import defaultdict
from requests import Session
import http
import logging
import urllib.parse
from ..record import GeoCodes
from ..record import Record
from . import ProviderException
from .base import BaseProvider
class GCoreClientException(ProviderException):
def __init__(self, r):
super(GCoreClientException, self).__init__(r.text)
class GCoreClientBadRequest(GCoreClientException):
def __init__(self, r):
super(GCoreClientBadRequest, self).__init__(r)
class GCoreClientNotFound(GCoreClientException):
def __init__(self, r):
super(GCoreClientNotFound, self).__init__(r)
class GCoreClient(object):
ROOT_ZONES = "zones"
def __init__(
self,
log,
api_url,
auth_url,
token=None,
token_type=None,
login=None,
password=None,
):
self.log = log
self._session = Session()
self._api_url = api_url
if token is not None and token_type is not None:
self._session.headers.update(
{"Authorization": f"{token_type} {token}"}
)
elif login is not None and password is not None:
token = self._auth(auth_url, login, password)
self._session.headers.update(
{"Authorization": f"Bearer {token}"}
)
else:
raise ValueError("either token or login & password must be set")
def _auth(self, url, login, password):
# well, can't use _request, since API returns 400 if credentials
# invalid which will be logged, but we don't want do this
r = self._session.request(
"POST",
self._build_url(url, "auth", "jwt", "login"),
json={"username": login, "password": password},
)
r.raise_for_status()
return r.json()["access"]
def _request(self, method, url, params=None, data=None):
r = self._session.request(
method, url, params=params, json=data, timeout=30.0
)
if r.status_code == http.HTTPStatus.BAD_REQUEST:
self.log.error(
"bad request %r has been sent to %r: %s", data, url, r.text
)
raise GCoreClientBadRequest(r)
elif r.status_code == http.HTTPStatus.NOT_FOUND:
self.log.error("resource %r not found: %s", url, r.text)
raise GCoreClientNotFound(r)
elif r.status_code == http.HTTPStatus.INTERNAL_SERVER_ERROR:
self.log.error("server error no %r to %r: %s", data, url, r.text)
raise GCoreClientException(r)
r.raise_for_status()
return r
def zone(self, zone_name):
return self._request(
"GET", self._build_url(self._api_url, self.ROOT_ZONES, zone_name)
).json()
def zone_create(self, zone_name):
return self._request(
"POST",
self._build_url(self._api_url, self.ROOT_ZONES),
data={"name": zone_name},
).json()
def zone_records(self, zone_name):
url = self._build_url(self._api_url, self.ROOT_ZONES, zone_name,
"rrsets")
rrsets = self._request("GET", url, params={"all": "true"}).json()
records = rrsets["rrsets"]
return records
def record_create(self, zone_name, rrset_name, type_, data):
self._request(
"POST", self._rrset_url(zone_name, rrset_name, type_), data=data
)
def record_update(self, zone_name, rrset_name, type_, data):
self._request(
"PUT", self._rrset_url(zone_name, rrset_name, type_), data=data
)
def record_delete(self, zone_name, rrset_name, type_):
self._request("DELETE", self._rrset_url(zone_name, rrset_name, type_))
def _rrset_url(self, zone_name, rrset_name, type_):
return self._build_url(
self._api_url, self.ROOT_ZONES, zone_name, rrset_name, type_
)
@staticmethod
def _build_url(base, *items):
for i in items:
base = base.strip("/") + "/"
base = urllib.parse.urljoin(base, i)
return base
class GCoreProvider(BaseProvider):
"""
GCore provider using API v2.
gcore:
class: octodns.provider.gcore.GCoreProvider
# Your API key
token: XXXXXXXXXXXX
# token_type: APIKey
# or login + password
login: XXXXXXXXXXXX
password: XXXXXXXXXXXX
# auth_url: https://api.gcdn.co
# url: https://dnsapi.gcorelabs.com/v2
# records_per_response: 1
"""
SUPPORTS_GEO = False
SUPPORTS_DYNAMIC = True
SUPPORTS = set(("A", "AAAA", "NS", "MX", "TXT", "SRV", "CNAME", "PTR"))
def __init__(self, id, *args, **kwargs):
token = kwargs.pop("token", None)
token_type = kwargs.pop("token_type", "APIKey")
login = kwargs.pop("login", None)
password = kwargs.pop("password", None)
api_url = kwargs.pop("url", "https://dnsapi.gcorelabs.com/v2")
auth_url = kwargs.pop("auth_url", "https://api.gcdn.co")
self.records_per_response = kwargs.pop("records_per_response", 1)
self.log = logging.getLogger(f"GCoreProvider[{id}]")
self.log.debug("__init__: id=%s", id)
super(GCoreProvider, self).__init__(id, *args, **kwargs)
self._client = GCoreClient(
self.log,
api_url,
auth_url,
token=token,
token_type=token_type,
login=login,
password=password,
)
def _add_dot_if_need(self, value):
return f"{value}." if not value.endswith(".") else value
def _build_pools(self, record, default_pool_name, value_transform_fn):
defaults = []
geo_sets, pool_idx = dict(), 0
pools = defaultdict(lambda: {"values": []})
for rr in record["resource_records"]:
meta = rr.get("meta", {}) or {}
value = {"value": value_transform_fn(rr["content"][0])}
countries = meta.get("countries", []) or []
continents = meta.get("continents", []) or []
if meta.get("default", False):
pools[default_pool_name]["values"].append(value)
defaults.append(value["value"])
continue
# defaults is false or missing and no conties or continents
elif len(continents) == 0 and len(countries) == 0:
defaults.append(value["value"])
continue
# RR with the same set of countries and continents are
# combined in single pool
geo_set = frozenset(
[GeoCodes.country_to_code(cc.upper()) for cc in countries]
) | frozenset(cc.upper() for cc in continents)
if geo_set not in geo_sets:
geo_sets[geo_set] = f"pool-{pool_idx}"
pool_idx += 1
pools[geo_sets[geo_set]]["values"].append(value)
return pools, geo_sets, defaults
def _build_rules(self, pools, geo_sets):
rules = []
for name, _ in pools.items():
rule = {"pool": name}
geo_set = next(
(
geo_set
for geo_set, pool_name in geo_sets.items()
if pool_name == name
),
{},
)
if len(geo_set) > 0:
rule["geos"] = list(geo_set)
rules.append(rule)
return sorted(rules, key=lambda x: x["pool"])
def _data_for_dynamic(self, record, value_transform_fn=lambda x: x):
default_pool = "other"
pools, geo_sets, defaults = self._build_pools(
record, default_pool, value_transform_fn
)
if len(pools) == 0:
raise RuntimeError(
f"filter is enabled, but no pools where built for {record}"
)
# defaults can't be empty, so use first pool values
if len(defaults) == 0:
defaults = [
value_transform_fn(v["value"])
for v in next(iter(pools.values()))["values"]
]
# if at least one default RR was found then setup fallback for
# other pools to default
if default_pool in pools:
for pool_name, pool in pools.items():
if pool_name == default_pool:
continue
pool["fallback"] = default_pool
rules = self._build_rules(pools, geo_sets)
return pools, rules, defaults
def _data_for_single(self, _type, record):
return {
"ttl": record["ttl"],
"type": _type,
"value": self._add_dot_if_need(
record["resource_records"][0]["content"][0]
),
}
_data_for_PTR = _data_for_single
def _data_for_CNAME(self, _type, record):
if record.get("filters") is None:
return self._data_for_single(_type, record)
pools, rules, defaults = self._data_for_dynamic(
record, self._add_dot_if_need
)
return {
"ttl": record["ttl"],
"type": _type,
"dynamic": {"pools": pools, "rules": rules},
"value": self._add_dot_if_need(defaults[0]),
}
def _data_for_multiple(self, _type, record):
extra = dict()
if record.get("filters") is not None:
pools, rules, defaults = self._data_for_dynamic(record)
extra = {
"dynamic": {"pools": pools, "rules": rules},
"values": defaults,
}
else:
extra = {
"values": [
rr_value
for resource_record in record["resource_records"]
for rr_value in resource_record["content"]
]
}
return {
"ttl": record["ttl"],
"type": _type,
**extra,
}
_data_for_A = _data_for_multiple
_data_for_AAAA = _data_for_multiple
def _data_for_TXT(self, _type, record):
return {
"ttl": record["ttl"],
"type": _type,
"values": [
rr_value.replace(";", "\\;")
for resource_record in record["resource_records"]
for rr_value in resource_record["content"]
],
}
def _data_for_MX(self, _type, record):
return {
"ttl": record["ttl"],
"type": _type,
"values": [
dict(
preference=preference,
exchange=self._add_dot_if_need(exchange),
)
for preference, exchange in map(
lambda x: x["content"], record["resource_records"]
)
],
}
def _data_for_NS(self, _type, record):
return {
"ttl": record["ttl"],
"type": _type,
"values": [
self._add_dot_if_need(rr_value)
for resource_record in record["resource_records"]
for rr_value in resource_record["content"]
],
}
def _data_for_SRV(self, _type, record):
return {
"ttl": record["ttl"],
"type": _type,
"values": [
dict(
priority=priority,
weight=weight,
port=port,
target=self._add_dot_if_need(target),
)
for priority, weight, port, target in map(
lambda x: x["content"], record["resource_records"]
)
],
}
def zone_records(self, zone):
try:
return self._client.zone_records(zone.name[:-1]), True
except GCoreClientNotFound:
return [], False
def populate(self, zone, target=False, lenient=False):
self.log.debug(
"populate: name=%s, target=%s, lenient=%s",
zone.name,
target,
lenient,
)
values = defaultdict(defaultdict)
records, exists = self.zone_records(zone)
for record in records:
_type = record["type"].upper()
if _type not in self.SUPPORTS:
continue
if self._should_ignore(record):
continue
rr_name = zone.hostname_from_fqdn(record["name"])
values[rr_name][_type] = record
before = len(zone.records)
for name, types in values.items():
for _type, record in types.items():
data_for = getattr(self, f"_data_for_{_type}")
record = Record.new(
zone,
name,
data_for(_type, record),
source=self,
lenient=lenient,
)
zone.add_record(record, lenient=lenient)
self.log.info(
"populate: found %s records, exists=%s",
len(zone.records) - before,
exists,
)
return exists
def _should_ignore(self, record):
name = record.get("name", "name-not-defined")
if record.get("filters") is None:
return False
want_filters = 3
filters = record.get("filters", [])
if len(filters) != want_filters:
self.log.info(
"ignore %s has filters and their count is not %d",
name,
want_filters,
)
return True
types = [v.get("type") for v in filters]
for i, want_type in enumerate(["geodns", "default", "first_n"]):
if types[i] != want_type:
self.log.info(
"ignore %s, filters.%d.type is %s, want %s",
name,
i,
types[i],
want_type,
)
return True
limits = [filters[i].get("limit", 1) for i in [1, 2]]
if limits[0] != limits[1]:
self.log.info(
"ignore %s, filters.1.limit (%d) != filters.2.limit (%d)",
name,
limits[0],
limits[1],
)
return True
return False
def _params_for_dymanic(self, record):
records = []
default_pool_found = False
default_values = set(
record.values if hasattr(record, "values") else [record.value]
)
for rule in record.dynamic.rules:
meta = dict()
# build meta tags if geos information present
if len(rule.data.get("geos", [])) > 0:
for geo_code in rule.data["geos"]:
geo = GeoCodes.parse(geo_code)
country = geo["country_code"]
continent = geo["continent_code"]
if country is not None:
meta.setdefault("countries", []).append(country)
else:
meta.setdefault("continents", []).append(continent)
else:
meta["default"] = True
pool_values = set()
pool_name = rule.data["pool"]
for value in record.dynamic.pools[pool_name].data["values"]:
v = value["value"]
records.append({"content": [v], "meta": meta})
pool_values.add(v)
default_pool_found |= default_values == pool_values
# if default values doesn't match any pool values, then just add this
# values with no any meta
if not default_pool_found:
for value in default_values:
records.append({"content": [value]})
return records
def _params_for_single(self, record):
return {
"ttl": record.ttl,
"resource_records": [{"content": [record.value]}],
}
_params_for_PTR = _params_for_single
def _params_for_CNAME(self, record):
if not record.dynamic:
return self._params_for_single(record)
return {
"ttl": record.ttl,
"resource_records": self._params_for_dymanic(record),
"filters": [
{"type": "geodns"},
{
"type": "default",
"limit": self.records_per_response,
"strict": False,
},
{"type": "first_n", "limit": self.records_per_response},
],
}
def _params_for_multiple(self, record):
extra = dict()
if record.dynamic:
extra["resource_records"] = self._params_for_dymanic(record)
extra["filters"] = [
{"type": "geodns"},
{
"type": "default",
"limit": self.records_per_response,
"strict": False,
},
{"type": "first_n", "limit": self.records_per_response},
]
else:
extra["resource_records"] = [
{"content": [value]} for value in record.values
]
return {
"ttl": record.ttl,
**extra,
}
_params_for_A = _params_for_multiple
_params_for_AAAA = _params_for_multiple
def _params_for_NS(self, record):
return {
"ttl": record.ttl,
"resource_records": [
{"content": [value]} for value in record.values
],
}
def _params_for_TXT(self, record):
return {
"ttl": record.ttl,
"resource_records": [
{"content": [value.replace("\\;", ";")]}
for value in record.values
],
}
def _params_for_MX(self, record):
return {
"ttl": record.ttl,
"resource_records": [
{"content": [rec.preference, rec.exchange]}
for rec in record.values
],
}
def _params_for_SRV(self, record):
return {
"ttl": record.ttl,
"resource_records": [
{"content": [rec.priority, rec.weight, rec.port, rec.target]}
for rec in record.values
],
}
def _apply_create(self, change):
self.log.info("creating: %s", change)
new = change.new
data = getattr(self, f"_params_for_{new._type}")(new)
self._client.record_create(
new.zone.name[:-1], new.fqdn, new._type, data
)
def _apply_update(self, change):
self.log.info("updating: %s", change)
new = change.new
data = getattr(self, f"_params_for_{new._type}")(new)
self._client.record_update(
new.zone.name[:-1], new.fqdn, new._type, data
)
def _apply_delete(self, change):
self.log.info("deleting: %s", change)
existing = change.existing
self._client.record_delete(
existing.zone.name[:-1], existing.fqdn, existing._type
)
def _apply(self, plan):
desired = plan.desired
changes = plan.changes
zone = desired.name[:-1]
self.log.debug(
"_apply: zone=%s, len(changes)=%d", desired.name, len(changes)
)
try:
self._client.zone(zone)
except GCoreClientNotFound:
self.log.info("_apply: no existing zone, trying to create it")
self._client.zone_create(zone)
self.log.info("_apply: zone has been successfully created")
changes.reverse()
for change in changes:
class_name = change.__class__.__name__
getattr(self, f"_apply_{class_name.lower()}")(change)
from __future__ import absolute_import, division, print_function, \
unicode_literals
from logging import getLogger
logger = getLogger('GCore')
try:
logger.warn('octodns_gcore shimmed. Update your provider class to '
'octodns_gcore.GCoreProvider. '
'Shim will be removed in 1.0')
from octodns_gcore import GCoreProvider
GCoreProvider # pragma: no cover
except ModuleNotFoundError:
logger.exception('GCoreProvider has been moved into a seperate module, '
'octodns_gcore is now required. Provider class should '
'be updated to octodns_gcore.GCoreProvider')
raise

9
requirements.txt
View File

@ -1,18 +1,13 @@
PyYaml==5.4
azure-common==1.1.27
azure-identity==1.5.0
azure-mgmt-dns==8.0.0
azure-mgmt-trafficmanager==0.51.0
dnspython==1.16.0
docutils==0.16
fqdn==1.5.0
jmespath==0.10.0
msrestazure==0.6.4
natsort==6.2.1
ovh==0.5.0
pycountry-convert==0.7.2
pycountry==20.7.3
pycountry==22.1.10
python-dateutil==2.8.1
requests==2.25.1
setuptools==44.1.1
setuptools==60.5.0
python-transip==0.5.0

154
tests/fixtures/gandi-no-changes.json vendored
View File

@ -1,154 +0,0 @@
[
{
"rrset_type": "A",
"rrset_ttl": 300,
"rrset_name": "@",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/%40/A",
"rrset_values": [
"1.2.3.4",
"1.2.3.5"
]
},
{
"rrset_type": "CAA",
"rrset_ttl": 3600,
"rrset_name": "@",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/%40/CAA",
"rrset_values": [
"0 issue \"ca.unit.tests\""
]
},
{
"rrset_type": "SSHFP",
"rrset_ttl": 3600,
"rrset_name": "@",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/%40/SSHFP",
"rrset_values": [
"1 1 7491973e5f8b39d5327cd4e08bc81b05f7710b49",
"1 1 bf6b6825d2977c511a475bbefb88aad54a92ac73"
]
},
{
"rrset_type": "AAAA",
"rrset_ttl": 600,
"rrset_name": "aaaa",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/aaaa/AAAA",
"rrset_values": [
"2601:644:500:e210:62f8:1dff:feb8:947a"
]
},
{
"rrset_type": "CNAME",
"rrset_ttl": 300,
"rrset_name": "cname",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/cname/CNAME",
"rrset_values": [
"unit.tests."
]
},
{
"rrset_type": "DNAME",
"rrset_ttl": 300,
"rrset_name": "dname",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/dname/DNAME",
"rrset_values": [
"unit.tests."
]
},
{
"rrset_type": "CNAME",
"rrset_ttl": 3600,
"rrset_name": "excluded",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/excluded/CNAME",
"rrset_values": [
"unit.tests."
]
},
{
"rrset_type": "MX",
"rrset_ttl": 300,
"rrset_name": "mx",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/mx/MX",
"rrset_values": [
"10 smtp-4.unit.tests.",
"20 smtp-2.unit.tests.",
"30 smtp-3.unit.tests.",
"40 smtp-1.unit.tests."
]
},
{
"rrset_type": "PTR",
"rrset_ttl": 300,
"rrset_name": "ptr",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/ptr/PTR",
"rrset_values": [
"foo.bar.com."
]
},
{
"rrset_type": "SPF",
"rrset_ttl": 600,
"rrset_name": "spf",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/spf/SPF",
"rrset_values": [
"\"v=spf1 ip4:192.168.0.1/16-all\""
]
},
{
"rrset_type": "TXT",
"rrset_ttl": 600,
"rrset_name": "txt",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/txt/TXT",
"rrset_values": [
"\"Bah bah black sheep\"",
"\"have you any wool.\"",
"\"v=DKIM1;k=rsa;s=email;h=sha256;p=A/kinda+of/long/string+with+numb3rs\""
]
},
{
"rrset_type": "A",
"rrset_ttl": 300,
"rrset_name": "www",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/www/A",
"rrset_values": [
"2.2.3.6"
]
},
{
"rrset_type": "A",
"rrset_ttl": 300,
"rrset_name": "www.sub",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/www.sub/A",
"rrset_values": [
"2.2.3.6"
]
},
{
"rrset_type": "SRV",
"rrset_ttl": 600,
"rrset_name": "_imap._tcp",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/_imap._tcp/SRV",
"rrset_values": [
"0 0 0 ."
]
},
{
"rrset_type": "SRV",
"rrset_ttl": 600,
"rrset_name": "_pop3._tcp",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/_pop3._tcp/SRV",
"rrset_values": [
"0 0 0 ."
]
},
{
"rrset_type": "SRV",
"rrset_ttl": 600,
"rrset_name": "_srv._tcp",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/_srv._tcp/SRV",
"rrset_values": [
"10 20 30 foo-1.unit.tests.",
"12 20 30 foo-2.unit.tests."
]
}
]

111
tests/fixtures/gandi-records.json vendored
View File

@ -1,111 +0,0 @@
[
{
"rrset_type": "A",
"rrset_ttl": 10800,
"rrset_name": "@",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/%40/A",
"rrset_values": [
"217.70.184.38"
]
},
{
"rrset_type": "MX",
"rrset_ttl": 10800,
"rrset_name": "@",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/%40/MX",
"rrset_values": [
"10 spool.mail.gandi.net.",
"50 fb.mail.gandi.net."
]
},
{
"rrset_type": "TXT",
"rrset_ttl": 10800,
"rrset_name": "@",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/%40/TXT",
"rrset_values": [
"\"v=spf1 include:_mailcust.gandi.net ?all\""
]
},
{
"rrset_type": "CNAME",
"rrset_ttl": 10800,
"rrset_name": "webmail",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/webmail/CNAME",
"rrset_values": [
"webmail.gandi.net."
]
},
{
"rrset_type": "CNAME",
"rrset_ttl": 10800,
"rrset_name": "www",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/www/CNAME",
"rrset_values": [
"webredir.vip.gandi.net."
]
},
{
"rrset_type": "SRV",
"rrset_ttl": 10800,
"rrset_name": "_imap._tcp",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/_imap._tcp/SRV",
"rrset_values": [
"0 0 0 ."
]
},
{
"rrset_type": "SRV",
"rrset_ttl": 10800,
"rrset_name": "_imaps._tcp",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/_imaps._tcp/SRV",
"rrset_values": [
"0 1 993 mail.gandi.net."
]
},
{
"rrset_type": "SRV",
"rrset_ttl": 10800,
"rrset_name": "_pop3._tcp",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/_pop3._tcp/SRV",
"rrset_values": [
"0 0 0 ."
]
},
{
"rrset_type": "SRV",
"rrset_ttl": 10800,
"rrset_name": "_pop3s._tcp",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/_pop3s._tcp/SRV",
"rrset_values": [
"10 1 995 mail.gandi.net."
]
},
{
"rrset_type": "SRV",
"rrset_ttl": 10800,
"rrset_name": "_submission._tcp",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/_submission._tcp/SRV",
"rrset_values": [
"0 1 465 mail.gandi.net."
]
},
{
"rrset_type": "CDS",
"rrset_ttl": 10800,
"rrset_name": "sub",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/sub/CDS",
"rrset_values": [
"32128 13 1 6823D9BB1B03DF714DD0EB163E20B341C96D18C0"
]
},
{
"rrset_type": "CNAME",
"rrset_ttl": 10800,
"rrset_name": "relative",
"rrset_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records/relative/CNAME",
"rrset_values": [
"target"
]
}
]

7
tests/fixtures/gandi-zone.json vendored
View File

@ -1,7 +0,0 @@
{
"domain_keys_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/keys",
"fqdn": "unit.tests",
"automatic_snapshots": true,
"domain_records_href": "https://api.gandi.net/v5/livedns/domains/unit.tests/records",
"domain_href": "https://api.gandi.net/v5/livedns/domains/unit.tests"
}

245
tests/fixtures/gcore-no-changes.json vendored
View File

@ -1,245 +0,0 @@
{
"rrsets": [
{
"name": "unit.tests",
"type": "A",
"ttl": 300,
"resource_records": [
{
"content": [
"1.2.3.4"
]
},
{
"content": [
"1.2.3.5"
]
}
]
},
{
"name": "unit.tests",
"type": "NS",
"ttl": 300,
"resource_records": [
{
"content": [
"ns2.gcdn.services"
]
},
{
"content": [
"ns1.gcorelabs.net"
]
}
]
},
{
"name": "_imap._tcp",
"type": "SRV",
"ttl": 600,
"resource_records": [
{
"content": [
0,
0,
0,
"."
]
}
]
},
{
"name": "_pop3._tcp",
"type": "SRV",
"ttl": 600,
"resource_records": [
{
"content": [
0,
0,
0,
"."
]
}
]
},
{
"name": "_srv._tcp",
"type": "SRV",
"ttl": 600,
"resource_records": [
{
"content": [
12,
20,
30,
"foo-2.unit.tests"
]
},
{
"content": [
10,
20,
30,
"foo-1.unit.tests"
]
}
]
},
{
"name": "aaaa.unit.tests",
"type": "AAAA",
"ttl": 600,
"resource_records": [
{
"content": [
"2601:644:500:e210:62f8:1dff:feb8:947a"
]
}
]
},
{
"name": "cname.unit.tests",
"type": "CNAME",
"ttl": 300,
"resource_records": [
{
"content": [
"unit.tests."
]
}
]
},
{
"name": "excluded.unit.tests",
"type": "CNAME",
"ttl": 3600,
"resource_records": [
{
"content": [
"unit.tests."
]
}
]
},
{
"name": "mx.unit.tests",
"type": "MX",
"ttl": 300,
"resource_records": [
{
"content": [
40,
"smtp-1.unit.tests."
]
},
{
"content": [
20,
"smtp-2.unit.tests."
]
},
{
"content": [
30,
"smtp-3.unit.tests."
]
},
{
"content": [
10,
"smtp-4.unit.tests."
]
}
]
},
{
"name": "ptr.unit.tests.",
"type": "PTR",
"ttl": 300,
"resource_records": [
{
"content": [
"foo.bar.com"
]
}
]
},
{
"name": "sub.unit.tests",
"type": "NS",
"ttl": 3600,
"resource_records": [
{
"content": [
"6.2.3.4"
]
},
{
"content": [
"7.2.3.4"
]
}
]
},
{
"name": "txt.unit.tests",
"type": "TXT",
"ttl": 600,
"resource_records": [
{
"content": [
"Bah bah black sheep"
]
},
{
"content": [
"have you any wool."
]
},
{
"content": [
"v=DKIM1;k=rsa;s=email;h=sha256;p=A/kinda+of/long/string+with+numb3rs"
]
}
]
},
{
"name": "www.unit.tests.",
"type": "A",
"ttl": 300,
"resource_records": [
{
"content": [
"2.2.3.6"
]
}
]
},
{
"name": "www.sub.unit.tests.",
"type": "A",
"ttl": 300,
"resource_records": [
{
"content": [
"2.2.3.6"
]
}
]
},
{
"name": "spf.sub.unit.tests.",
"type": "SPF",
"ttl": 600,
"resource_records": [
{
"content": [
"v=spf1 ip4:192.168.0.1/16-all"
]
}
]
}
]
}

428
tests/fixtures/gcore-records.json vendored
View File

@ -1,428 +0,0 @@
{
"rrsets": [
{
"name": "unit.tests",
"type": "A",
"ttl": 300,
"resource_records": [
{
"content": [
"1.2.3.4"
]
}
]
},
{
"name": "unit.tests",
"type": "NS",
"ttl": 300,
"resource_records": [
{
"content": [
"ns2.gcdn.services"
]
},
{
"content": [
"ns1.gcorelabs.net"
]
}
]
},
{
"name": "_imap._tcp",
"type": "SRV",
"ttl": 1200,
"resource_records": [
{
"content": [
0,
0,
0,
"."
]
}
]
},
{
"name": "_pop3._tcp",
"type": "SRV",
"ttl": 1200,
"resource_records": [
{
"content": [
0,
0,
0,
"."
]
}
]
},
{
"name": "_srv._tcp",
"type": "SRV",
"ttl": 1200,
"resource_records": [
{
"content": [
12,
20,
30,
"foo-2.unit.tests."
]
},
{
"content": [
10,
20,
30,
"foo-1.unit.tests."
]
}
]
},
{
"name": "aaaa.unit.tests",
"type": "AAAA",
"ttl": 600,
"resource_records": [
{
"content": [
"2601:644:500:e210:62f8:1dff:feb8:947a"
]
}
]
},
{
"name": "cname.unit.tests",
"type": "CNAME",
"ttl": 300,
"resource_records": [
{
"content": [
"unit.tests."
]
}
]
},
{
"name": "mx.unit.tests",
"type": "MX",
"ttl": 600,
"resource_records": [
{
"content": [
40,
"smtp-1.unit.tests."
]
},
{
"content": [
20,
"smtp-2.unit.tests."
]
}
]
},
{
"name": "ptr.unit.tests.",
"type": "PTR",
"ttl": 300,
"resource_records": [
{
"content": [
"foo.bar.com"
]
}
]
},
{
"name": "sub.unit.tests",
"type": "NS",
"ttl": 300,
"resource_records": [
{
"content": [
"6.2.3.4"
]
},
{
"content": [
"7.2.3.4"
]
}
]
},
{
"name": "txt.unit.tests",
"type": "TXT",
"ttl": 300,
"resource_records": [
{
"content": [
"\"Bah bah black sheep\""
]
},
{
"content": [
"\"have you any wool.\""
]
},
{
"content": [
"\"v=DKIM1;k=rsa;s=email;h=sha256;p=A/kinda+of/long/string+with+numb3rs\""
]
}
]
},
{
"name": "www.unit.tests.",
"type": "A",
"ttl": 300,
"resource_records": [
{
"content": [
"2.2.3.6"
]
}
]
},
{
"name": "www.sub.unit.tests.",
"type": "A",
"ttl": 300,
"resource_records": [
{
"content": [
"2.2.3.6"
]
}
]
},
{
"name": "geo-A-single.unit.tests.",
"type": "A",
"ttl": 300,
"filters": [
{
"type": "geodns"
},
{
"limit": 1,
"strict": false,
"type": "default"
},
{
"limit": 1,
"type": "first_n"
}
],
"resource_records": [
{
"content": [
"7.7.7.7"
],
"meta": {
"countries": [
"RU"
]
}
},
{
"content": [
"8.8.8.8"
],
"meta": {
"countries": [
"RU"
]
}
},
{
"content": [
"9.9.9.9"
],
"meta": {
"continents": [
"EU"
]
}
},
{
"content": [
"10.10.10.10"
],
"meta": {
"default": true
}
}
]
},
{
"name": "geo-no-def.unit.tests.",
"type": "A",
"ttl": 300,
"filters": [
{
"type": "geodns"
},
{
"limit": 1,
"strict": false,
"type": "default"
},
{
"limit": 1,
"type": "first_n"
}
],
"resource_records": [
{
"content": [
"7.7.7.7"
],
"meta": {
"countries": [
"RU"
]
}
}
]
},
{
"name": "geo-CNAME.unit.tests.",
"type": "CNAME",
"ttl": 300,
"filters": [
{
"type": "geodns"
},
{
"limit": 1,
"strict": false,
"type": "default"
},
{
"limit": 1,
"type": "first_n"
}
],
"resource_records": [
{
"content": [
"ru-1.unit.tests"
],
"meta": {
"countries": [
"RU"
]
}
},
{
"content": [
"ru-2.unit.tests"
],
"meta": {
"countries": [
"RU"
]
}
},
{
"content": [
"eu.unit.tests"
],
"meta": {
"continents": [
"EU"
]
}
},
{
"content": [
"any.unit.tests."
],
"meta": {
"default": true
}
}
]
},
{
"name": "geo-ignore-len-filters.unit.tests.",
"type": "A",
"ttl": 300,
"filters": [
{
"limit": 1,
"type": "first_n"
},
{
"limit": 1,
"strict": false,
"type": "default"
}
],
"resource_records": [
{
"content": [
"7.7.7.7"
]
}
]
},
{
"name": "geo-ignore-types.unit.tests.",
"type": "A",
"ttl": 300,
"filters": [
{
"type": "geodistance"
},
{
"limit": 1,
"type": "first_n"
},
{
"limit": 1,
"strict": false,
"type": "default"
}
],
"resource_records": [
{
"content": [
"7.7.7.7"
]
}
]
},
{
"name": "geo-ignore-limits.unit.tests.",
"type": "A",
"ttl": 300,
"filters": [
{
"type": "geodns"
},
{
"limit": 2,
"strict": false,
"type": "default"
},
{
"limit": 1,
"type": "first_n"
}
],
"resource_records": [
{
"content": [
"7.7.7.7"
]
}
]
}
]
}

27
tests/fixtures/gcore-zone.json vendored
View File

@ -1,27 +0,0 @@
{
"id": 27757,
"name": "unit.test",
"nx_ttl": 300,
"retry": 5400,
"refresh": 0,
"expiry": 1209600,
"contact": "support@gcorelabs.com",
"serial": 1614752868,
"primary_server": "ns1.gcorelabs.net",
"records": [
{
"id": 12419,
"name": "unit.test",
"type": "ns",
"ttl": 300,
"short_answers": [
"[ns2.gcdn.services]",
"[ns1.gcorelabs.net]"
]
}
],
"dns_servers": [
"ns1.gcorelabs.net",
"ns2.gcdn.services"
]
}

2462
tests/test_octodns_provider_azuredns.py
View File

File diff suppressed because it is too large Load Diff

373
tests/test_octodns_provider_gandi.py
View File

@ -5,372 +5,17 @@
from __future__ import absolute_import, division, print_function, \
unicode_literals
from mock import Mock, call
from os.path import dirname, join
from requests import HTTPError
from requests_mock import ANY, mock as requests_mock
from unittest import TestCase
from octodns.record import Record
from octodns.provider.gandi import GandiProvider, GandiClientBadRequest, \
GandiClientUnauthorized, GandiClientForbidden, GandiClientNotFound, \
GandiClientUnknownDomainName
from octodns.provider.yaml import YamlProvider
from octodns.zone import Zone
# Just for coverage
import octodns.provider.fastdns
# Quell warnings
octodns.provider.fastdns
class TestGandiProvider(TestCase):
expected = Zone('unit.tests.', [])
source = YamlProvider('test', join(dirname(__file__), 'config'))
source.populate(expected)
class TestGandiShim(TestCase):
# We remove this record from the test zone as Gandi API reject it
# (rightfully).
expected._remove_record(Record.new(expected, 'sub', {
'ttl': 1800,
'type': 'NS',
'values': [
'6.2.3.4.',
'7.2.3.4.'
]
}))
def test_populate(self):
provider = GandiProvider('test_id', 'token')
# 400 - Bad Request.
with requests_mock() as mock:
mock.get(ANY, status_code=400,
text='{"status": "error", "errors": [{"location": '
'"body", "name": "items", "description": '
'"\'6.2.3.4.\': invalid hostname (param: '
'{\'rrset_type\': u\'NS\', \'rrset_ttl\': 3600, '
'\'rrset_name\': u\'sub\', \'rrset_values\': '
'[u\'6.2.3.4.\', u\'7.2.3.4.\']})"}, {"location": '
'"body", "name": "items", "description": '
'"\'7.2.3.4.\': invalid hostname (param: '
'{\'rrset_type\': u\'NS\', \'rrset_ttl\': 3600, '
'\'rrset_name\': u\'sub\', \'rrset_values\': '
'[u\'6.2.3.4.\', u\'7.2.3.4.\']})"}]}')
with self.assertRaises(GandiClientBadRequest) as ctx:
zone = Zone('unit.tests.', [])
provider.populate(zone)
self.assertIn('"status": "error"', str(ctx.exception))
# 401 - Unauthorized.
with requests_mock() as mock:
mock.get(ANY, status_code=401,
text='{"code":401,"message":"The server could not verify '
'that you authorized to access the document you '
'requested. Either you supplied the wrong '
'credentials (e.g., bad api key), or your access '
'token has expired","object":"HTTPUnauthorized",'
'"cause":"Unauthorized"}')
with self.assertRaises(GandiClientUnauthorized) as ctx:
zone = Zone('unit.tests.', [])
provider.populate(zone)
self.assertIn('"cause":"Unauthorized"', str(ctx.exception))
# 403 - Forbidden.
with requests_mock() as mock:
mock.get(ANY, status_code=403,
text='{"code":403,"message":"Access was denied to this '
'resource.","object":"HTTPForbidden","cause":'
'"Forbidden"}')
with self.assertRaises(GandiClientForbidden) as ctx:
zone = Zone('unit.tests.', [])
provider.populate(zone)
self.assertIn('"cause":"Forbidden"', str(ctx.exception))
# 404 - Not Found.
with requests_mock() as mock:
mock.get(ANY, status_code=404,
text='{"code": 404, "message": "The resource could not '
'be found.", "object": "HTTPNotFound", "cause": '
'"Not Found"}')
with self.assertRaises(GandiClientNotFound) as ctx:
zone = Zone('unit.tests.', [])
provider._client.zone(zone)
self.assertIn('"cause": "Not Found"', str(ctx.exception))
# General error
with requests_mock() as mock:
mock.get(ANY, status_code=502, text='Things caught fire')
with self.assertRaises(HTTPError) as ctx:
zone = Zone('unit.tests.', [])
provider.populate(zone)
self.assertEquals(502, ctx.exception.response.status_code)
# No diffs == no changes
with requests_mock() as mock:
base = 'https://api.gandi.net/v5/livedns/domains/unit.tests' \
'/records'
with open('tests/fixtures/gandi-no-changes.json') as fh:
mock.get(base, text=fh.read())
zone = Zone('unit.tests.', [])
provider.populate(zone)
self.assertEquals(16, len(zone.records))
changes = self.expected.changes(zone, provider)
self.assertEquals(0, len(changes))
del provider._zone_records[zone.name]
# Default Gandi zone file.
with requests_mock() as mock:
base = 'https://api.gandi.net/v5/livedns/domains/unit.tests' \
'/records'
with open('tests/fixtures/gandi-records.json') as fh:
mock.get(base, text=fh.read())
zone = Zone('unit.tests.', [])
provider.populate(zone)
self.assertEquals(11, len(zone.records))
changes = self.expected.changes(zone, provider)
self.assertEquals(24, len(changes))
# 2nd populate makes no network calls/all from cache
again = Zone('unit.tests.', [])
provider.populate(again)
self.assertEquals(11, len(again.records))
# bust the cache
del provider._zone_records[zone.name]
def test_apply(self):
provider = GandiProvider('test_id', 'token')
# Zone does not exists but can be created.
with requests_mock() as mock:
mock.get(ANY, status_code=404,
text='{"code": 404, "message": "The resource could not '
'be found.", "object": "HTTPNotFound", "cause": '
'"Not Found"}')
mock.post(ANY, status_code=201,
text='{"message": "Domain Created"}')
plan = provider.plan(self.expected)
provider.apply(plan)
# Zone does not exists and can't be created.
with requests_mock() as mock:
mock.get(ANY, status_code=404,
text='{"code": 404, "message": "The resource could not '
'be found.", "object": "HTTPNotFound", "cause": '
'"Not Found"}')
mock.post(ANY, status_code=404,
text='{"code": 404, "message": "The resource could not '
'be found.", "object": "HTTPNotFound", "cause": '
'"Not Found"}')
with self.assertRaises((GandiClientNotFound,
GandiClientUnknownDomainName)) as ctx:
plan = provider.plan(self.expected)
provider.apply(plan)
self.assertIn('This domain is not registered at Gandi.',
str(ctx.exception))
resp = Mock()
resp.json = Mock()
provider._client._request = Mock(return_value=resp)
with open('tests/fixtures/gandi-zone.json') as fh:
zone = fh.read()
# non-existent domain
resp.json.side_effect = [
GandiClientNotFound(resp), # no zone in populate
GandiClientNotFound(resp), # no domain during apply
zone
]
plan = provider.plan(self.expected)
# No root NS, no ignored, no excluded, no LOC
n = len(self.expected.records) - 6
self.assertEquals(n, len(plan.changes))
self.assertEquals(n, provider.apply(plan))
self.assertFalse(plan.exists)
provider._client._request.assert_has_calls([
call('GET', '/livedns/domains/unit.tests/records'),
call('GET', '/livedns/domains/unit.tests'),
call('POST', '/livedns/domains', data={
'fqdn': 'unit.tests',
'zone': {}
}),
call('POST', '/livedns/domains/unit.tests/records', data={
'rrset_name': 'www.sub',
'rrset_ttl': 300,
'rrset_type': 'A',
'rrset_values': ['2.2.3.6']
}),
call('POST', '/livedns/domains/unit.tests/records', data={
'rrset_name': 'www',
'rrset_ttl': 300,
'rrset_type': 'A',
'rrset_values': ['2.2.3.6']
}),
call('POST', '/livedns/domains/unit.tests/records', data={
'rrset_name': 'txt',
'rrset_ttl': 600,
'rrset_type': 'TXT',
'rrset_values': [
'Bah bah black sheep',
'have you any wool.',
'v=DKIM1;k=rsa;s=email;h=sha256;p=A/kinda+of/long/string'
'+with+numb3rs'
]
}),
call('POST', '/livedns/domains/unit.tests/records', data={
'rrset_name': 'spf',
'rrset_ttl': 600,
'rrset_type': 'SPF',
'rrset_values': ['v=spf1 ip4:192.168.0.1/16-all']
}),
call('POST', '/livedns/domains/unit.tests/records', data={
'rrset_name': 'ptr',
'rrset_ttl': 300,
'rrset_type': 'PTR',
'rrset_values': ['foo.bar.com.']
}),
call('POST', '/livedns/domains/unit.tests/records', data={
'rrset_name': 'mx',
'rrset_ttl': 300,
'rrset_type': 'MX',
'rrset_values': [
'10 smtp-4.unit.tests.',
'20 smtp-2.unit.tests.',
'30 smtp-3.unit.tests.',
'40 smtp-1.unit.tests.'
]
}),
call('POST', '/livedns/domains/unit.tests/records', data={
'rrset_name': 'excluded',
'rrset_ttl': 3600,
'rrset_type': 'CNAME',
'rrset_values': ['unit.tests.']
}),
call('POST', '/livedns/domains/unit.tests/records', data={
'rrset_name': 'dname',
'rrset_ttl': 300,
'rrset_type': 'DNAME',
'rrset_values': ['unit.tests.']
}),
call('POST', '/livedns/domains/unit.tests/records', data={
'rrset_name': 'cname',
'rrset_ttl': 300,
'rrset_type': 'CNAME',
'rrset_values': ['unit.tests.']
}),
call('POST', '/livedns/domains/unit.tests/records', data={
'rrset_name': 'aaaa',
'rrset_ttl': 600,
'rrset_type': 'AAAA',
'rrset_values': ['2601:644:500:e210:62f8:1dff:feb8:947a']
}),
call('POST', '/livedns/domains/unit.tests/records', data={
'rrset_name': '_srv._tcp',
'rrset_ttl': 600,
'rrset_type': 'SRV',
'rrset_values': [
'10 20 30 foo-1.unit.tests.',
'12 20 30 foo-2.unit.tests.'
]
}),
call('POST', '/livedns/domains/unit.tests/records', data={
'rrset_name': '_pop3._tcp',
'rrset_ttl': 600,
'rrset_type': 'SRV',
'rrset_values': [
'0 0 0 .',
]
}),
call('POST', '/livedns/domains/unit.tests/records', data={
'rrset_name': '_imap._tcp',
'rrset_ttl': 600,
'rrset_type': 'SRV',
'rrset_values': [
'0 0 0 .',
]
}),
call('POST', '/livedns/domains/unit.tests/records', data={
'rrset_name': '@',
'rrset_ttl': 3600,
'rrset_type': 'SSHFP',
'rrset_values': [
'1 1 7491973e5f8b39d5327cd4e08bc81b05f7710b49',
'1 1 bf6b6825d2977c511a475bbefb88aad54a92ac73'
]
}),
call('POST', '/livedns/domains/unit.tests/records', data={
'rrset_name': '@',
'rrset_ttl': 3600,
'rrset_type': 'CAA',
'rrset_values': ['0 issue "ca.unit.tests"']
}),
call('POST', '/livedns/domains/unit.tests/records', data={
'rrset_name': '@',
'rrset_ttl': 300,
'rrset_type': 'A',
'rrset_values': ['1.2.3.4', '1.2.3.5']
})
])
# expected number of total calls
self.assertEquals(19, provider._client._request.call_count)
provider._client._request.reset_mock()
# delete 1 and update 1
provider._client.zone_records = Mock(return_value=[
{
'rrset_name': 'www',
'rrset_ttl': 300,
'rrset_type': 'A',
'rrset_values': ['1.2.3.4']
},
{
'rrset_name': 'www',
'rrset_ttl': 300,
'rrset_type': 'A',
'rrset_values': ['2.2.3.4']
},
{
'rrset_name': 'ttl',
'rrset_ttl': 600,
'rrset_type': 'A',
'rrset_values': ['3.2.3.4']
}
])
# Domain exists, we don't care about return
resp.json.side_effect = ['{}']
wanted = Zone('unit.tests.', [])
wanted.add_record(Record.new(wanted, 'ttl', {
'ttl': 300,
'type': 'A',
'value': '3.2.3.4'
}))
plan = provider.plan(wanted)
self.assertTrue(plan.exists)
self.assertEquals(2, len(plan.changes))
self.assertEquals(2, provider.apply(plan))
# recreate for update, and deletes for the 2 parts of the other
provider._client._request.assert_has_calls([
call('DELETE', '/livedns/domains/unit.tests/records/www/A'),
call('DELETE', '/livedns/domains/unit.tests/records/ttl/A'),
call('POST', '/livedns/domains/unit.tests/records', data={
'rrset_name': 'ttl',
'rrset_ttl': 300,
'rrset_type': 'A',
'rrset_values': ['3.2.3.4']
})
], any_order=True)
def test_missing(self):
with self.assertRaises(ModuleNotFoundError):
from octodns.provider.gandi import GandiProvider
GandiProvider

669
tests/test_octodns_provider_gcore.py
View File

@ -2,670 +2,15 @@
#
#
from __future__ import (
absolute_import,
division,
print_function,
unicode_literals,
)
from __future__ import absolute_import, division, print_function, \
unicode_literals
from mock import Mock, call
from os.path import dirname, join
from requests_mock import ANY, mock as requests_mock
from unittest import TestCase
from octodns.record import Record, Update, Delete, Create
from octodns.provider.gcore import (
GCoreProvider,
GCoreClientBadRequest,
GCoreClientNotFound,
GCoreClientException,
)
from octodns.provider.yaml import YamlProvider
from octodns.zone import Zone
class TestGCoreShim(TestCase):
class TestGCoreProvider(TestCase):
expected = Zone("unit.tests.", [])
source = YamlProvider("test", join(dirname(__file__), "config"))
source.populate(expected)
default_filters = [
{"type": "geodns"},
{
"type": "default",
"limit": 1,
"strict": False,
},
{"type": "first_n", "limit": 1},
]
def test_populate(self):
provider = GCoreProvider("test_id", token="token")
# TC: 400 - Bad Request.
with requests_mock() as mock:
mock.get(ANY, status_code=400, text='{"error":"bad body"}')
with self.assertRaises(GCoreClientBadRequest) as ctx:
zone = Zone("unit.tests.", [])
provider.populate(zone)
self.assertIn('"error":"bad body"', str(ctx.exception))
# TC: 404 - Not Found.
with requests_mock() as mock:
mock.get(
ANY, status_code=404, text='{"error":"zone is not found"}'
)
with self.assertRaises(GCoreClientNotFound) as ctx:
zone = Zone("unit.tests.", [])
provider._client.zone(zone.name)
self.assertIn(
'"error":"zone is not found"', str(ctx.exception)
)
# TC: General error
with requests_mock() as mock:
mock.get(ANY, status_code=500, text="Things caught fire")
with self.assertRaises(GCoreClientException) as ctx:
zone = Zone("unit.tests.", [])
provider.populate(zone)
self.assertEqual("Things caught fire", str(ctx.exception))
# TC: No credentials or token error
with requests_mock() as mock:
with self.assertRaises(ValueError) as ctx:
GCoreProvider("test_id")
self.assertEqual(
"either token or login & password must be set",
str(ctx.exception),
)
# TC: Auth with login password
with requests_mock() as mock:
def match_body(request):
return {"username": "foo", "password": "bar"} == request.json()
auth_url = "http://api/auth/jwt/login"
mock.post(
auth_url,
additional_matcher=match_body,
status_code=200,
json={"access": "access"},
)
providerPassword = GCoreProvider(
"test_id",
url="http://dns",
auth_url="http://api",
login="foo",
password="bar",
)
assert mock.called
# make sure token passed in header
zone_rrset_url = "http://dns/zones/unit.tests/rrsets?all=true"
mock.get(
zone_rrset_url,
request_headers={"Authorization": "Bearer access"},
status_code=404,
)
zone = Zone("unit.tests.", [])
assert not providerPassword.populate(zone)
# TC: No diffs == no changes
with requests_mock() as mock:
base = "https://dnsapi.gcorelabs.com/v2/zones/unit.tests/rrsets"
with open("tests/fixtures/gcore-no-changes.json") as fh:
mock.get(base, text=fh.read())
zone = Zone("unit.tests.", [])
provider.populate(zone)
self.assertEqual(14, len(zone.records))
self.assertEqual(
{
"",
"_imap._tcp",
"_pop3._tcp",
"_srv._tcp",
"aaaa",
"cname",
"excluded",
"mx",
"ptr",
"sub",
"txt",
"www",
"www.sub",
},
{r.name for r in zone.records},
)
changes = self.expected.changes(zone, provider)
self.assertEqual(0, len(changes))
# TC: 4 create (dynamic) + 1 removed + 7 modified
with requests_mock() as mock:
base = "https://dnsapi.gcorelabs.com/v2/zones/unit.tests/rrsets"
with open("tests/fixtures/gcore-records.json") as fh:
mock.get(base, text=fh.read())
zone = Zone("unit.tests.", [])
provider.populate(zone)
self.assertEqual(16, len(zone.records))
changes = self.expected.changes(zone, provider)
self.assertEqual(11, len(changes))
self.assertEqual(
3, len([c for c in changes if isinstance(c, Create)])
)
self.assertEqual(
1, len([c for c in changes if isinstance(c, Delete)])
)
self.assertEqual(
7, len([c for c in changes if isinstance(c, Update)])
)
# TC: no pools can be built
with requests_mock() as mock:
base = "https://dnsapi.gcorelabs.com/v2/zones/unit.tests/rrsets"
mock.get(
base,
json={
"rrsets": [
{
"name": "unit.tests.",
"type": "A",
"ttl": 300,
"filters": self.default_filters,
"resource_records": [{"content": ["7.7.7.7"]}],
}
]
},
)
zone = Zone("unit.tests.", [])
with self.assertRaises(RuntimeError) as ctx:
provider.populate(zone)
self.assertTrue(
str(ctx.exception).startswith(
"filter is enabled, but no pools where built for"
),
f"{ctx.exception} - is not start from desired text",
)
def test_apply(self):
provider = GCoreProvider("test_id", url="http://api", token="token")
# TC: Zone does not exists but can be created.
with requests_mock() as mock:
mock.get(
ANY, status_code=404, text='{"error":"zone is not found"}'
)
mock.post(ANY, status_code=200, text='{"id":1234}')
plan = provider.plan(self.expected)
provider.apply(plan)
# TC: Zone does not exists and can't be created.
with requests_mock() as mock:
mock.get(
ANY, status_code=404, text='{"error":"zone is not found"}'
)
mock.post(
ANY,
status_code=400,
text='{"error":"parent zone is already'
' occupied by another client"}',
)
with self.assertRaises(
(GCoreClientNotFound, GCoreClientBadRequest)
) as ctx:
plan = provider.plan(self.expected)
provider.apply(plan)
self.assertIn(
"parent zone is already occupied by another client",
str(ctx.exception),
)
resp = Mock()
resp.json = Mock()
provider._client._request = Mock(return_value=resp)
with open("tests/fixtures/gcore-zone.json") as fh:
zone = fh.read()
# non-existent domain
resp.json.side_effect = [
GCoreClientNotFound(resp), # no zone in populate
GCoreClientNotFound(resp), # no domain during apply
zone,
]
plan = provider.plan(self.expected)
# TC: create all
self.assertEqual(13, len(plan.changes))
self.assertEqual(13, provider.apply(plan))
self.assertFalse(plan.exists)
provider._client._request.assert_has_calls(
[
call(
"GET",
"http://api/zones/unit.tests/rrsets",
params={"all": "true"},
),
call("GET", "http://api/zones/unit.tests"),
call("POST", "http://api/zones", data={"name": "unit.tests"}),
call(
"POST",
"http://api/zones/unit.tests/www.sub.unit.tests./A",
data={
"ttl": 300,
"resource_records": [{"content": ["2.2.3.6"]}],
},
),
call(
"POST",
"http://api/zones/unit.tests/www.unit.tests./A",
data={
"ttl": 300,
"resource_records": [{"content": ["2.2.3.6"]}],
},
),
call(
"POST",
"http://api/zones/unit.tests/txt.unit.tests./TXT",
data={
"ttl": 600,
"resource_records": [
{"content": ["Bah bah black sheep"]},
{"content": ["have you any wool."]},
{
"content": [
"v=DKIM1;k=rsa;s=email;h=sha256;p=A/kinda+"
"of/long/string+with+numb3rs"
]
},
],
},
),
call(
"POST",
"http://api/zones/unit.tests/sub.unit.tests./NS",
data={
"ttl": 3600,
"resource_records": [
{"content": ["6.2.3.4."]},
{"content": ["7.2.3.4."]},
],
},
),
call(
"POST",
"http://api/zones/unit.tests/ptr.unit.tests./PTR",
data={
"ttl": 300,
"resource_records": [
{"content": ["foo.bar.com."]},
],
},
),
call(
"POST",
"http://api/zones/unit.tests/mx.unit.tests./MX",
data={
"ttl": 300,
"resource_records": [
{"content": [10, "smtp-4.unit.tests."]},
{"content": [20, "smtp-2.unit.tests."]},
{"content": [30, "smtp-3.unit.tests."]},
{"content": [40, "smtp-1.unit.tests."]},
],
},
),
call(
"POST",
"http://api/zones/unit.tests/excluded.unit.tests./CNAME",
data={
"ttl": 3600,
"resource_records": [{"content": ["unit.tests."]}],
},
),
call(
"POST",
"http://api/zones/unit.tests/cname.unit.tests./CNAME",
data={
"ttl": 300,
"resource_records": [{"content": ["unit.tests."]}],
},
),
call(
"POST",
"http://api/zones/unit.tests/aaaa.unit.tests./AAAA",
data={
"ttl": 600,
"resource_records": [
{
"content": [
"2601:644:500:e210:62f8:1dff:feb8:947a"
]
}
],
},
),
call(
"POST",
"http://api/zones/unit.tests/_srv._tcp.unit.tests./SRV",
data={
"ttl": 600,
"resource_records": [
{"content": [10, 20, 30, "foo-1.unit.tests."]},
{"content": [12, 20, 30, "foo-2.unit.tests."]},
],
},
),
call(
"POST",
"http://api/zones/unit.tests/_pop3._tcp.unit.tests./SRV",
data={
"ttl": 600,
"resource_records": [{"content": [0, 0, 0, "."]}],
},
),
call(
"POST",
"http://api/zones/unit.tests/_imap._tcp.unit.tests./SRV",
data={
"ttl": 600,
"resource_records": [{"content": [0, 0, 0, "."]}],
},
),
call(
"POST",
"http://api/zones/unit.tests/unit.tests./A",
data={
"ttl": 300,
"resource_records": [
{"content": ["1.2.3.4"]},
{"content": ["1.2.3.5"]},
],
},
),
]
)
# expected number of total calls
self.assertEqual(16, provider._client._request.call_count)
# TC: delete 1 and update 1
provider._client._request.reset_mock()
provider._client.zone_records = Mock(
return_value=[
{
"name": "www",
"ttl": 300,
"type": "A",
"resource_records": [{"content": ["1.2.3.4"]}],
},
{
"name": "ttl",
"ttl": 600,
"type": "A",
"resource_records": [{"content": ["3.2.3.4"]}],
},
]
)
# Domain exists, we don't care about return
resp.json.side_effect = ["{}"]
wanted = Zone("unit.tests.", [])
wanted.add_record(
Record.new(
wanted, "ttl", {"ttl": 300, "type": "A", "value": "3.2.3.4"}
)
)
plan = provider.plan(wanted)
self.assertTrue(plan.exists)
self.assertEqual(2, len(plan.changes))
self.assertEqual(2, provider.apply(plan))
provider._client._request.assert_has_calls(
[
call(
"DELETE", "http://api/zones/unit.tests/www.unit.tests./A"
),
call(
"PUT",
"http://api/zones/unit.tests/ttl.unit.tests./A",
data={
"ttl": 300,
"resource_records": [{"content": ["3.2.3.4"]}],
},
),
]
)
# TC: create dynamics
provider._client._request.reset_mock()
provider._client.zone_records = Mock(return_value=[])
# Domain exists, we don't care about return
resp.json.side_effect = ["{}"]
wanted = Zone("unit.tests.", [])
wanted.add_record(
Record.new(
wanted,
"geo-simple",
{
"ttl": 300,
"type": "A",
"value": "3.3.3.3",
"dynamic": {
"pools": {
"pool-1": {
"fallback": "other",
"values": [
{"value": "1.1.1.1"},
{"value": "1.1.1.2"},
],
},
"pool-2": {
"fallback": "other",
"values": [
{"value": "2.2.2.1"},
],
},
"other": {"values": [{"value": "3.3.3.3"}]},
},
"rules": [
{"pool": "pool-1", "geos": ["EU-RU"]},
{"pool": "pool-2", "geos": ["EU"]},
{"pool": "other"},
],
},
},
),
)
wanted.add_record(
Record.new(
wanted,
"geo-defaults",
{
"ttl": 300,
"type": "A",
"value": "3.2.3.4",
"dynamic": {
"pools": {
"pool-1": {
"values": [
{"value": "2.2.2.1"},
],
},
},
"rules": [
{"pool": "pool-1", "geos": ["EU"]},
],
},
},
),
)
wanted.add_record(
Record.new(
wanted,
"cname-smpl",
{
"ttl": 300,
"type": "CNAME",
"value": "en.unit.tests.",
"dynamic": {
"pools": {
"pool-1": {
"fallback": "other",
"values": [
{"value": "ru-1.unit.tests."},
{"value": "ru-2.unit.tests."},
],
},
"pool-2": {
"fallback": "other",
"values": [
{"value": "eu.unit.tests."},
],
},
"other": {"values": [{"value": "en.unit.tests."}]},
},
"rules": [
{"pool": "pool-1", "geos": ["EU-RU"]},
{"pool": "pool-2", "geos": ["EU"]},
{"pool": "other"},
],
},
},
),
)
wanted.add_record(
Record.new(
wanted,
"cname-dflt",
{
"ttl": 300,
"type": "CNAME",
"value": "en.unit.tests.",
"dynamic": {
"pools": {
"pool-1": {
"values": [
{"value": "eu.unit.tests."},
],
},
},
"rules": [
{"pool": "pool-1", "geos": ["EU"]},
],
},
},
),
)
plan = provider.plan(wanted)
self.assertTrue(plan.exists)
self.assertEqual(4, len(plan.changes))
self.assertEqual(4, provider.apply(plan))
provider._client._request.assert_has_calls(
[
call(
"POST",
"http://api/zones/unit.tests/geo-simple.unit.tests./A",
data={
"ttl": 300,
"filters": self.default_filters,
"resource_records": [
{
"content": ["1.1.1.1"],
"meta": {"countries": ["RU"]},
},
{
"content": ["1.1.1.2"],
"meta": {"countries": ["RU"]},
},
{
"content": ["2.2.2.1"],
"meta": {"continents": ["EU"]},
},
{
"content": ["3.3.3.3"],
"meta": {"default": True},
},
],
},
),
call(
"POST",
"http://api/zones/unit.tests/geo-defaults.unit.tests./A",
data={
"ttl": 300,
"filters": self.default_filters,
"resource_records": [
{
"content": ["2.2.2.1"],
"meta": {"continents": ["EU"]},
},
{
"content": ["3.2.3.4"],
},
],
},
),
call(
"POST",
"http://api/zones/unit.tests/cname-smpl.unit.tests./CNAME",
data={
"ttl": 300,
"filters": self.default_filters,
"resource_records": [
{
"content": ["ru-1.unit.tests."],
"meta": {"countries": ["RU"]},
},
{
"content": ["ru-2.unit.tests."],
"meta": {"countries": ["RU"]},
},
{
"content": ["eu.unit.tests."],
"meta": {"continents": ["EU"]},
},
{
"content": ["en.unit.tests."],
"meta": {"default": True},
},
],
},
),
call(
"POST",
"http://api/zones/unit.tests/cname-dflt.unit.tests./CNAME",
data={
"ttl": 300,
"filters": self.default_filters,
"resource_records": [
{
"content": ["eu.unit.tests."],
"meta": {"continents": ["EU"]},
},
{
"content": ["en.unit.tests."],
},
],
},
),
]
)
def test_missing(self):
with self.assertRaises(ModuleNotFoundError):
from octodns.provider.gcore import GCoreProvider
GCoreProvider