diff --git a/CHANGELOG.md b/CHANGELOG.md index 80e1850..7a5da59 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,7 @@ * [DynProvider](https://github.com/octodns/octodns-dynprovider/) * [EasyDnsProvider](https://github.com/octodns/octodns-easydns/) * [EtcHostsProvider](https://github.com/octodns/octodns-etchosts/) + * [GoogleCloudProvider](https://github.com/octodns/octodns-googlecloud/) * [Ns1Provider](https://github.com/octodns/octodns-ns1/) * [PowerDnsProvider](https://github.com/octodns/octodns-powerdns/) * [Route53Provider](https://github.com/octodns/octodns-route53/) also diff --git a/README.md b/README.md index a7aeaad..edccd70 100644 --- a/README.md +++ b/README.md @@ -205,7 +205,7 @@ The table below lists the providers octoDNS supports. We're currently in the pro | [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 | | -| [GoogleCloudProvider](/octodns/provider/googlecloud.py) | | google-cloud-dns | A, AAAA, CAA, CNAME, MX, NAPTR, NS, PTR, SPF, SRV, TXT | No | | +| [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 | | | [Ns1Provider](https://github.com/octodns/octodns-ns1/) | [octodns_ns1](https://github.com/octodns/octodns-ns1/) | | | | | diff --git a/octodns/provider/googlecloud.py b/octodns/provider/googlecloud.py index 9d424f0..c75e010 100644 --- a/octodns/provider/googlecloud.py +++ b/octodns/provider/googlecloud.py @@ -5,334 +5,18 @@ from __future__ import absolute_import, division, print_function, \ unicode_literals -import shlex -import time from logging import getLogger -from uuid import uuid4 -import re -from google.cloud import dns - -from .base import BaseProvider -from ..record import Record - - -class GoogleCloudProvider(BaseProvider): - """ - Google Cloud DNS provider - - google_cloud: - class: octodns.provider.googlecloud.GoogleCloudProvider - # Credentials file for a service_account or other account can be - # specified with the GOOGLE_APPLICATION_CREDENTIALS environment - # variable. (https://console.cloud.google.com/apis/credentials) - # - # The project to work on (not required) - # project: foobar - # - # The File with the google credentials (not required). If used, the - # "project" parameter needs to be set, else it will fall back to the - # "default credentials" - # credentials_file: ~/google_cloud_credentials_file.json - # - """ - - SUPPORTS = set(('A', 'AAAA', 'CAA', 'CNAME', 'MX', 'NAPTR', - 'NS', 'PTR', 'SPF', 'SRV', 'TXT')) - SUPPORTS_GEO = False - SUPPORTS_DYNAMIC = False - - CHANGE_LOOP_WAIT = 5 - - def __init__(self, id, project=None, credentials_file=None, - *args, **kwargs): - - if credentials_file: - self.gcloud_client = dns.Client.from_service_account_json( - credentials_file, project=project) - else: - self.gcloud_client = dns.Client(project=project) - - # Logger - self.log = getLogger(f'GoogleCloudProvider[{id}]') - self.id = id - - self._gcloud_zones = {} - - super(GoogleCloudProvider, self).__init__(id, *args, **kwargs) - - def _apply(self, plan): - """Required function of manager.py to actually apply a record change. - - :param plan: Contains the zones and changes to be made - :type plan: octodns.provider.base.Plan - - :type return: void - """ - desired = plan.desired - changes = plan.changes - - self.log.debug('_apply: zone=%s, len(changes)=%d', desired.name, - len(changes)) - - # Get gcloud zone, or create one if none existed before. - if desired.name not in self.gcloud_zones: - gcloud_zone = self._create_gcloud_zone(desired.name) - else: - gcloud_zone = self.gcloud_zones.get(desired.name) - - gcloud_changes = gcloud_zone.changes() - - for change in changes: - class_name = change.__class__.__name__ - _rrset_func = getattr(self, f'_rrset_for_{change.record._type}') - - if class_name == 'Create': - gcloud_changes.add_record_set( - _rrset_func(gcloud_zone, change.record)) - elif class_name == 'Delete': - gcloud_changes.delete_record_set( - _rrset_func(gcloud_zone, change.record)) - elif class_name == 'Update': - gcloud_changes.delete_record_set( - _rrset_func(gcloud_zone, change.existing)) - gcloud_changes.add_record_set( - _rrset_func(gcloud_zone, change.new)) - else: - msg = f'Change type "{class_name}" for change ' \ - f'"{str(change)}" is none of "Create", "Delete" or "Update' - raise RuntimeError(msg) - - gcloud_changes.create() - - for i in range(120): - gcloud_changes.reload() - # https://cloud.google.com/dns/api/v1/changes#resource - # status can be one of either "pending" or "done" - if gcloud_changes.status != 'pending': - break - self.log.debug("Waiting for changes to complete") - time.sleep(self.CHANGE_LOOP_WAIT) - - if gcloud_changes.status != 'done': - timeout = i * self.CHANGE_LOOP_WAIT - raise RuntimeError(f"Timeout reached after {timeout} seconds") - - def _create_gcloud_zone(self, dns_name): - """Creates a google cloud ManagedZone with dns_name, and zone named - derived from it. calls .create() method and returns it. - - :param dns_name: fqdn of zone to create - :type dns_name: str - - :type return: new google.cloud.dns.ManagedZone - """ - # Zone name must begin with a letter, end with a letter or digit, - # and only contain lowercase letters, digits or dashes, - # and be 63 characters or less - zone_name = f'zone-{dns_name.replace(".", "-")}-{uuid4().hex}'[:63] - - gcloud_zone = self.gcloud_client.zone( - name=zone_name, - dns_name=dns_name - ) - gcloud_zone.create(client=self.gcloud_client) - - # add this new zone to the list of zones. - self._gcloud_zones[gcloud_zone.dns_name] = gcloud_zone - - self.log.info(f"Created zone {zone_name}. Fqdn {dns_name}.") - - return gcloud_zone - - def _get_gcloud_records(self, gcloud_zone, page_token=None): - """ Generator function which yields ResourceRecordSet for the managed - gcloud zone, until there are no more records to pull. - - :param gcloud_zone: zone to pull records from - :type gcloud_zone: google.cloud.dns.ManagedZone - :param page_token: page token for the page to get - - :return: a resource record set - :type return: google.cloud.dns.ResourceRecordSet - """ - gcloud_iterator = gcloud_zone.list_resource_record_sets( - page_token=page_token) - for gcloud_record in gcloud_iterator: - yield gcloud_record - # This is to get results which may be on a "paged" page. - # (if more than max_results) entries. - if gcloud_iterator.next_page_token: - for gcloud_record in self._get_gcloud_records( - gcloud_zone, gcloud_iterator.next_page_token): - # yield from is in python 3 only. - yield gcloud_record - - def _get_cloud_zones(self, page_token=None): - """Load all ManagedZones into the self._gcloud_zones dict which is - mapped with the dns_name as key. - - :return: void - """ - - gcloud_zones = self.gcloud_client.list_zones(page_token=page_token) - for gcloud_zone in gcloud_zones: - self._gcloud_zones[gcloud_zone.dns_name] = gcloud_zone - - if gcloud_zones.next_page_token: - self._get_cloud_zones(gcloud_zones.next_page_token) - - @property - def gcloud_zones(self): - if not self._gcloud_zones: - self._get_cloud_zones() - return self._gcloud_zones - - def populate(self, zone, target=False, lenient=False): - """Required function of manager.py to collect records from zone. - - :param zone: A dns zone - :type zone: octodns.zone.Zone - :param target: Unused. - :type target: bool - :param lenient: Unused. Check octodns.manager for usage. - :type lenient: bool - - :type return: void - """ - - self.log.debug('populate: name=%s, target=%s, lenient=%s', zone.name, - target, lenient) - - exists = False - before = len(zone.records) - - gcloud_zone = self.gcloud_zones.get(zone.name) - - if gcloud_zone: - exists = True - for gcloud_record in self._get_gcloud_records(gcloud_zone): - if gcloud_record.record_type.upper() not in self.SUPPORTS: - continue - - record_name = gcloud_record.name - if record_name.endswith(zone.name): - # google cloud always return fqdn. Make relative record - # here. "root" records will then get the '' record_name, - # which is also the way octodns likes it. - record_name = record_name[:-(len(zone.name) + 1)] - typ = gcloud_record.record_type.upper() - data = getattr(self, f'_data_for_{typ}') - data = data(gcloud_record) - data['type'] = typ - data['ttl'] = gcloud_record.ttl - self.log.debug('populate: adding record %s records: %s', - record_name, data) - record = Record.new(zone, record_name, data, source=self) - zone.add_record(record, lenient=lenient) - - self.log.info('populate: found %s records, exists=%s', - len(zone.records) - before, exists) - return exists - - def _data_for_A(self, gcloud_record): - return { - 'values': gcloud_record.rrdatas - } - - _data_for_AAAA = _data_for_A - - def _data_for_CAA(self, gcloud_record): - return { - 'values': [{ - 'flags': v[0], - 'tag': v[1], - 'value': v[2]} - for v in [shlex.split(g) for g in gcloud_record.rrdatas]]} - - def _data_for_CNAME(self, gcloud_record): - return { - 'value': gcloud_record.rrdatas[0] - } - - def _data_for_MX(self, gcloud_record): - return {'values': [{ - "preference": v[0], - "exchange": v[1]} - for v in [shlex.split(g) for g in gcloud_record.rrdatas]]} - - def _data_for_NAPTR(self, gcloud_record): - return {'values': [{ - 'order': v[0], - 'preference': v[1], - 'flags': v[2], - 'service': v[3], - 'regexp': v[4], - 'replacement': v[5]} - for v in [shlex.split(g) for g in gcloud_record.rrdatas]]} - - _data_for_NS = _data_for_A - - _data_for_PTR = _data_for_CNAME - - _fix_semicolons = re.compile(r'(? 1: - return { - 'values': [self._fix_semicolons.sub('\\;', rr) - for rr in gcloud_record.rrdatas]} - return { - 'value': self._fix_semicolons.sub('\\;', gcloud_record.rrdatas[0])} - - def _data_for_SRV(self, gcloud_record): - return {'values': [{ - 'priority': v[0], - 'weight': v[1], - 'port': v[2], - 'target': v[3]} - for v in [shlex.split(g) for g in gcloud_record.rrdatas]]} - - _data_for_TXT = _data_for_SPF - - def _rrset_for_A(self, gcloud_zone, record): - return gcloud_zone.resource_record_set( - record.fqdn, record._type, record.ttl, record.values) - - _rrset_for_AAAA = _rrset_for_A - - def _rrset_for_CAA(self, gcloud_zone, record): - return gcloud_zone.resource_record_set( - record.fqdn, record._type, record.ttl, [ - f'{v.flags} {v.tag} {v.value}' for v in record.values]) - - def _rrset_for_CNAME(self, gcloud_zone, record): - return gcloud_zone.resource_record_set( - record.fqdn, record._type, record.ttl, [record.value]) - - def _rrset_for_MX(self, gcloud_zone, record): - return gcloud_zone.resource_record_set( - record.fqdn, record._type, record.ttl, [ - f'{v.preference} {v.exchange}' for v in record.values]) - - def _rrset_for_NAPTR(self, gcloud_zone, record): - return gcloud_zone.resource_record_set( - record.fqdn, record._type, record.ttl, [ - f'{v.order} {v.preference} "{v.flags}" "{v.service}" ' - f'"{v.regexp}" {v.replacement}' for v in record.values]) - - _rrset_for_NS = _rrset_for_A - - _rrset_for_PTR = _rrset_for_CNAME - - def _rrset_for_SPF(self, gcloud_zone, record): - return gcloud_zone.resource_record_set( - record.fqdn, record._type, record.ttl, record.chunked_values) - - def _rrset_for_SRV(self, gcloud_zone, record): - return gcloud_zone.resource_record_set( - record.fqdn, record._type, record.ttl, [ - f'{v.priority} {v.weight} {v.port} {v.target}' - for v in record.values]) - - _rrset_for_TXT = _rrset_for_SPF +logger = getLogger('GoogleCloud') +try: + logger.warn('octodns_googlecloud shimmed. Update your provider class to ' + 'octodns_googlecloud.GoogleCloudProvider. ' + 'Shim will be removed in 1.0') + from octodns_googlecloud import GoogleCloudProvider + GoogleCloudProvider # pragma: no cover +except ModuleNotFoundError: + logger.exception('GoogleCloudProvider has been moved into a seperate ' + 'module, octodns_googlecloud is now required. Provider ' + 'class should be updated to ' + 'octodns_googlecloud.GoogleCloudProvider') + raise diff --git a/requirements.txt b/requirements.txt index d4e6353..5ba50c7 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,8 +6,6 @@ azure-mgmt-trafficmanager==0.51.0 dnspython==1.16.0 docutils==0.16 fqdn==1.5.0 -google-cloud-core==1.4.1 -google-cloud-dns==0.32.0 jmespath==0.10.0 msrestazure==0.6.4 natsort==6.2.1 diff --git a/tests/test_octodns_provider_googlecloud.py b/tests/test_octodns_provider_googlecloud.py index 9348ea1..4a8fdf5 100644 --- a/tests/test_octodns_provider_googlecloud.py +++ b/tests/test_octodns_provider_googlecloud.py @@ -5,458 +5,12 @@ from __future__ import absolute_import, division, print_function, \ unicode_literals -from octodns.record import Create, Delete, Update, Record -from octodns.provider.googlecloud import GoogleCloudProvider - -from octodns.zone import Zone -from octodns.provider.base import Plan, BaseProvider - from unittest import TestCase -from mock import Mock, patch, PropertyMock - -zone = Zone(name='unit.tests.', sub_zones=[]) -octo_records = [] -octo_records.append(Record.new(zone, '', { - 'ttl': 0, - 'type': 'A', - 'values': ['1.2.3.4', '10.10.10.10']})) -octo_records.append(Record.new(zone, 'a', { - 'ttl': 1, - 'type': 'A', - 'values': ['1.2.3.4', '1.1.1.1']})) -octo_records.append(Record.new(zone, 'aa', { - 'ttl': 9001, - 'type': 'A', - 'values': ['1.2.4.3']})) -octo_records.append(Record.new(zone, 'aaa', { - 'ttl': 2, - 'type': 'A', - 'values': ['1.1.1.3']})) -octo_records.append(Record.new(zone, 'cname', { - 'ttl': 3, - 'type': 'CNAME', - 'value': 'a.unit.tests.'})) -octo_records.append(Record.new(zone, 'mx1', { - 'ttl': 3, - 'type': 'MX', - 'values': [{ - 'priority': 10, - 'value': 'mx1.unit.tests.', - }, { - 'priority': 20, - 'value': 'mx2.unit.tests.', - }]})) -octo_records.append(Record.new(zone, 'mx2', { - 'ttl': 3, - 'type': 'MX', - 'values': [{ - 'priority': 10, - 'value': 'mx1.unit.tests.', - }]})) -octo_records.append(Record.new(zone, '', { - 'ttl': 4, - 'type': 'NS', - 'values': ['ns1.unit.tests.', 'ns2.unit.tests.']})) -octo_records.append(Record.new(zone, 'foo', { - 'ttl': 5, - 'type': 'NS', - 'value': 'ns1.unit.tests.'})) -octo_records.append(Record.new(zone, '_srv._tcp', { - 'ttl': 6, - 'type': 'SRV', - 'values': [{ - 'priority': 10, - 'weight': 20, - 'port': 30, - 'target': 'foo-1.unit.tests.', - }, { - 'priority': 12, - 'weight': 30, - 'port': 30, - 'target': 'foo-2.unit.tests.', - }]})) -octo_records.append(Record.new(zone, '_srv2._tcp', { - 'ttl': 7, - 'type': 'SRV', - 'values': [{ - 'priority': 12, - 'weight': 17, - 'port': 1, - 'target': 'srvfoo.unit.tests.', - }]})) -octo_records.append(Record.new(zone, 'txt1', { - 'ttl': 8, - 'type': 'TXT', - 'value': 'txt singleton test'})) -octo_records.append(Record.new(zone, 'txt2', { - 'ttl': 9, - 'type': 'TXT', - 'values': ['txt multiple test', 'txt multiple test 2']})) -octo_records.append(Record.new(zone, 'naptr', { - 'ttl': 9, - 'type': 'NAPTR', - 'values': [{ - 'order': 100, - 'preference': 10, - 'flags': 'S', - 'service': 'SIP+D2U', - 'regexp': "!^.*$!sip:customer-service@unit.tests!", - 'replacement': '_sip._udp.unit.tests.' - }]})) -octo_records.append(Record.new(zone, 'caa', { - 'ttl': 9, - 'type': 'CAA', - 'value': { - 'flags': 0, - 'tag': 'issue', - 'value': 'ca.unit.tests', - }})) -for record in octo_records: - zone.add_record(record) - -# This is the format which the google API likes. -resource_record_sets = [ - ('unit.tests.', u'A', 0, [u'1.2.3.4', u'10.10.10.10']), - (u'a.unit.tests.', u'A', 1, [u'1.1.1.1', u'1.2.3.4']), - (u'aa.unit.tests.', u'A', 9001, [u'1.2.4.3']), - (u'aaa.unit.tests.', u'A', 2, [u'1.1.1.3']), - (u'cname.unit.tests.', u'CNAME', 3, [u'a.unit.tests.']), - (u'mx1.unit.tests.', u'MX', 3, - [u'10 mx1.unit.tests.', u'20 mx2.unit.tests.']), - (u'mx2.unit.tests.', u'MX', 3, [u'10 mx1.unit.tests.']), - ('unit.tests.', u'NS', 4, [u'ns1.unit.tests.', u'ns2.unit.tests.']), - (u'foo.unit.tests.', u'NS', 5, [u'ns1.unit.tests.']), - (u'_srv._tcp.unit.tests.', u'SRV', 6, - [u'10 20 30 foo-1.unit.tests.', u'12 30 30 foo-2.unit.tests.']), - (u'_srv2._tcp.unit.tests.', u'SRV', 7, [u'12 17 1 srvfoo.unit.tests.']), - (u'txt1.unit.tests.', u'TXT', 8, [u'txt singleton test']), - (u'txt2.unit.tests.', u'TXT', 9, - [u'txt multiple test', u'txt multiple test 2']), - (u'naptr.unit.tests.', u'NAPTR', 9, [ - u'100 10 "S" "SIP+D2U" "!^.*$!sip:customer-service@unit.tests!"' - u' _sip._udp.unit.tests.']), - (u'caa.unit.tests.', u'CAA', 9, [u'0 issue ca.unit.tests']) -] - - -class DummyResourceRecordSet: - def __init__(self, record_name, record_type, ttl, rrdatas): - self.name = record_name - self.record_type = record_type - self.ttl = ttl - self.rrdatas = rrdatas - - def __eq__(self, other): - try: - return self.name == other.name \ - and self.record_type == other.record_type \ - and self.ttl == other.ttl \ - and sorted(self.rrdatas) == sorted(other.rrdatas) - except: - return False - - def __repr__(self): - return f"{self.name} {self.record_type} {self.ttl} {self.rrdatas}" - - def __hash__(self): - return hash(repr(self)) - - -class DummyGoogleCloudZone: - def __init__(self, dns_name, name=""): - self.dns_name = dns_name - self.name = name - - def resource_record_set(self, *args): - return DummyResourceRecordSet(*args) - - def list_resource_record_sets(self, *args): - pass - - def create(self, *args, **kwargs): - pass - - -class DummyIterator: - """Returns a mock DummyIterator object to use in testing. - This is because API calls for google cloud DNS, if paged, contains a - "next_page_token", which can be used to grab a subsequent - iterator with more results. - - :type return: DummyIterator - """ - def __init__(self, list_of_stuff, page_token=None): - self.iterable = iter(list_of_stuff) - self.next_page_token = page_token - - def __iter__(self): - return self - - # python2 - def next(self): - return next(self.iterable) - - # python3 - def __next__(self): - return next(self.iterable) - - -class TestGoogleCloudProvider(TestCase): - @patch('octodns.provider.googlecloud.dns') - def _get_provider(*args): - '''Returns a mock GoogleCloudProvider object to use in testing. - - :type return: GoogleCloudProvider - ''' - return GoogleCloudProvider(id=1, project="mock") - - @patch('octodns.provider.googlecloud.dns') - def test___init__(self, *_): - self.assertIsInstance(GoogleCloudProvider(id=1, - credentials_file="test", - project="unit test"), - BaseProvider) - - self.assertIsInstance(GoogleCloudProvider(id=1), - BaseProvider) - - @patch('octodns.provider.googlecloud.time.sleep') - @patch('octodns.provider.googlecloud.dns') - def test__apply(self, *_): - class DummyDesired: - def __init__(self, name, changes): - self.name = name - self.changes = changes - - apply_z = Zone("unit.tests.", []) - create_r = Record.new(apply_z, '', { - 'ttl': 0, - 'type': 'A', - 'values': ['1.2.3.4', '10.10.10.10']}) - delete_r = Record.new(apply_z, 'a', { - 'ttl': 1, - 'type': 'A', - 'values': ['1.2.3.4', '1.1.1.1']}) - update_existing_r = Record.new(apply_z, 'aa', { - 'ttl': 9001, - 'type': 'A', - 'values': ['1.2.4.3']}) - update_new_r = Record.new(apply_z, 'aa', { - 'ttl': 666, - 'type': 'A', - 'values': ['1.4.3.2']}) - - gcloud_zone_mock = DummyGoogleCloudZone("unit.tests.", "unit-tests") - status_mock = Mock() - return_values_for_status = iter( - ["pending"] * 11 + ['done', 'done']) - type(status_mock).status = PropertyMock( - side_effect=lambda: next(return_values_for_status)) - gcloud_zone_mock.changes = Mock(return_value=status_mock) - - provider = self._get_provider() - provider.gcloud_client = Mock() - provider._gcloud_zones = {"unit.tests.": gcloud_zone_mock} - desired = Mock() - desired.name = "unit.tests." - changes = [] - changes.append(Create(create_r)) - changes.append(Delete(delete_r)) - changes.append(Update(existing=update_existing_r, new=update_new_r)) - - provider.apply(Plan( - existing=[update_existing_r, delete_r], - desired=desired, - changes=changes, - exists=True - )) - - calls_mock = gcloud_zone_mock.changes.return_value - mocked_calls = [] - for mock_call in calls_mock.add_record_set.mock_calls: - mocked_calls.append(mock_call[1][0]) - - self.assertEqual(mocked_calls, [ - DummyResourceRecordSet( - 'unit.tests.', 'A', 0, ['1.2.3.4', '10.10.10.10']), - DummyResourceRecordSet( - 'aa.unit.tests.', 'A', 666, ['1.4.3.2']) - ]) - - mocked_calls2 = [] - for mock_call in calls_mock.delete_record_set.mock_calls: - mocked_calls2.append(mock_call[1][0]) - - self.assertEqual(mocked_calls2, [ - DummyResourceRecordSet( - 'a.unit.tests.', 'A', 1, ['1.2.3.4', '1.1.1.1']), - DummyResourceRecordSet( - 'aa.unit.tests.', 'A', 9001, ['1.2.4.3']) - ]) - - type(status_mock).status = "pending" - - with self.assertRaises(RuntimeError): - provider.apply(Plan( - existing=[update_existing_r, delete_r], - desired=desired, - changes=changes, - exists=True - )) - - unsupported_change = Mock() - unsupported_change.__len__ = Mock(return_value=1) - type_mock = Mock() - type_mock._type = "A" - unsupported_change.record = type_mock - - mock_plan = Mock() - type(mock_plan).desired = PropertyMock(return_value=DummyDesired( - "dummy name", [])) - type(mock_plan).changes = [unsupported_change] - - with self.assertRaises(RuntimeError): - provider.apply(mock_plan) - - def test__get_gcloud_client(self): - provider = self._get_provider() - - self.assertIsInstance(provider, GoogleCloudProvider) - - @patch('octodns.provider.googlecloud.dns') - def test_populate(self, _): - def _get_mock_zones(page_token=None): - if not page_token: - return DummyIterator([ - DummyGoogleCloudZone('example.com.'), - ], page_token="MOCK_PAGE_TOKEN") - elif page_token == "MOCK_PAGE_TOKEN": - return DummyIterator([ - DummyGoogleCloudZone('example2.com.'), - ], page_token="MOCK_PAGE_TOKEN2") - - return DummyIterator([ - google_cloud_zone - ]) - - def _get_mock_record_sets(page_token=None): - if not page_token: - return DummyIterator( - [DummyResourceRecordSet(*v) for v in - resource_record_sets[:3]], page_token="MOCK_PAGE_TOKEN") - elif page_token == "MOCK_PAGE_TOKEN": - - return DummyIterator( - [DummyResourceRecordSet(*v) for v in - resource_record_sets[3:5]], page_token="MOCK_PAGE_TOKEN2") - return DummyIterator( - [DummyResourceRecordSet(*v) for v in resource_record_sets[5:]]) - - google_cloud_zone = DummyGoogleCloudZone('unit.tests.') - - provider = self._get_provider() - provider.gcloud_client.list_zones = Mock(side_effect=_get_mock_zones) - google_cloud_zone.list_resource_record_sets = Mock( - side_effect=_get_mock_record_sets) - - self.assertEqual(provider.gcloud_zones.get("unit.tests.").dns_name, - "unit.tests.") - - test_zone = Zone('unit.tests.', []) - exists = provider.populate(test_zone) - self.assertTrue(exists) - - # test_zone gets fed the same records as zone does, except it's in - # the format returned by google API, so after populate they should look - # exactly the same. - self.assertEqual(test_zone.records, zone.records) - - test_zone2 = Zone('nonexistent.zone.', []) - exists = provider.populate(test_zone2, False, False) - self.assertFalse(exists) - - self.assertEqual(len(test_zone2.records), 0, - msg="Zone should not get records from wrong domain") - - provider.SUPPORTS = set() - test_zone3 = Zone('unit.tests.', []) - provider.populate(test_zone3) - self.assertEqual(len(test_zone3.records), 0) - - @patch('octodns.provider.googlecloud.dns') - def test_populate_corner_cases(self, _): - provider = self._get_provider() - test_zone = Zone('unit.tests.', []) - not_same_fqdn = DummyResourceRecordSet( - 'unit.tests.gr', u'A', 0, [u'1.2.3.4']), - - provider._get_gcloud_records = Mock( - side_effect=[not_same_fqdn]) - provider._gcloud_zones = { - "unit.tests.": DummyGoogleCloudZone("unit.tests.", "unit-tests")} - - provider.populate(test_zone) - - self.assertEqual(len(test_zone.records), 1) - - self.assertEqual(test_zone.records.pop().fqdn, - u'unit.tests.gr.unit.tests.') - - def test__get_gcloud_zone(self): - provider = self._get_provider() - - provider.gcloud_client = Mock() - provider.gcloud_client.list_zones = Mock( - return_value=DummyIterator([])) - - self.assertIsNone(provider.gcloud_zones.get("nonexistent.zone"), - msg="Check that nonexistent zones return None when" - "there's no create=True flag") - - def test__get_rrsets(self): - provider = self._get_provider() - dummy_gcloud_zone = DummyGoogleCloudZone("unit.tests") - for octo_record in octo_records: - _rrset_func = getattr( - provider, f'_rrset_for_{octo_record._type}') - self.assertEqual( - _rrset_func(dummy_gcloud_zone, octo_record).record_type, - octo_record._type - ) - - def test__create_zone(self): - provider = self._get_provider() - - provider.gcloud_client = Mock() - provider.gcloud_client.list_zones = Mock( - return_value=DummyIterator([])) - - mock_zone = provider._create_gcloud_zone("nonexistent.zone.mock") - - mock_zone.create.assert_called() - provider.gcloud_client.zone.assert_called() - - def test__create_zone_ip6_arpa(self): - def _create_dummy_zone(name, dns_name): - return DummyGoogleCloudZone(name=name, dns_name=dns_name) - - provider = self._get_provider() - - provider.gcloud_client = Mock() - provider.gcloud_client.zone = Mock(side_effect=_create_dummy_zone) - - mock_zone = \ - provider._create_gcloud_zone('0.0.0.0.8.b.d.0.1.0.0.2.ip6.arpa') - self.assertRegexpMatches(mock_zone.name, '^[a-z][a-z0-9-]*[a-z0-9]$') - self.assertEqual(len(mock_zone.name), 63) - def test_semicolon_fixup(self): - provider = self._get_provider() +class TestGoogleCloudShim(TestCase): - self.assertEquals({ - 'values': ['abcd\\; ef\\;g', 'hij\\; klm\\;n'] - }, provider._data_for_TXT( - DummyResourceRecordSet( - 'unit.tests.', 'TXT', 0, ['abcd; ef;g', 'hij\\; klm\\;n']) - )) + def test_missing(self): + with self.assertRaises(ModuleNotFoundError): + from octodns.provider.googlecloud import GoogleCloudProvider + GoogleCloudProvider