openstack/shade
Code Issues Proposed changes

Merge "Remove designateclient from commands related to recordsets"

Browse Source
This commit is contained in:
Jenkins 2017-06-02 12:21:29 +00:00 committed by Gerrit Code Review
commit b77c9a7481
3 changed files with 53 additions and 204 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
shade/_tasks.py
View File

@ -407,31 +407,6 @@ class RolesForUser(task_manager.Task):
return client.keystone_client.roles.roles_for_user(**self.args)
class RecordSetList(task_manager.Task):
def main(self, client):
return client.designate_client.recordsets.list(**self.args)
class RecordSetGet(task_manager.Task):
def main(self, client):
return client.designate_client.recordsets.get(**self.args)
class RecordSetCreate(task_manager.Task):
def main(self, client):
return client.designate_client.recordsets.create(**self.args)
class RecordSetUpdate(task_manager.Task):
def main(self, client):
return client.designate_client.recordsets.update(**self.args)
class RecordSetDelete(task_manager.Task):
def main(self, client):
return client.designate_client.recordsets.delete(**self.args)
class NovaQuotasSet(task_manager.Task):
def main(self, client):
return client.nova_client.quotas.update(**self.args)

50
shade/openstackcloud.py
View File

@ -7050,8 +7050,9 @@ class OpenStackCloud(
:returns: A list of recordsets.
"""
with _utils.shade_exceptions("Error fetching recordsets list"):
return self.manager.submit_task(_tasks.RecordSetList(zone=zone))
return self._dns_client.get(
"/v2/zones/{zone_id}/recordsets".format(zone_id=zone),
error_message="Error fetching recordsets list")
def get_recordset(self, zone, name_or_id):
"""Get a recordset by name or ID.
@ -7064,9 +7065,10 @@ class OpenStackCloud(
"""
try:
return self.manager.submit_task(_tasks.RecordSetGet(
zone=zone,
recordset=name_or_id))
return self._dns_client.get(
"/v2/zones/{zone_id}/recordsets/{recordset_id}".format(
zone_id=zone, recordset_id=name_or_id),
error_message="Error fetching recordset")
except Exception:
return None
@ -7097,11 +7099,22 @@ class OpenStackCloud(
# We capitalize the type in case the user sends in lowercase
recordset_type = recordset_type.upper()
with _utils.shade_exceptions(
"Unable to create recordset {name}".format(name=name)):
return self.manager.submit_task(_tasks.RecordSetCreate(
zone=zone, name=name, type_=recordset_type, records=records,
description=description, ttl=ttl))
body = {
'name': name,
'type': recordset_type,
'records': records
}
if description:
body['description'] = description
if ttl:
body['ttl'] = ttl
return self._dns_client.post(
"/v2/zones/{zone_id}/recordsets".format(zone_id=zone),
json=body,
error_message="Error creating recordset {name}".format(name=name))
@_utils.valid_kwargs('description', 'ttl', 'records')
def update_recordset(self, zone, name_or_id, **kwargs):
@ -7127,11 +7140,10 @@ class OpenStackCloud(
raise OpenStackCloudException(
"Recordset %s not found." % name_or_id)
with _utils.shade_exceptions(
"Error updating recordset {0}".format(name_or_id)):
new_recordset = self.manager.submit_task(
_tasks.RecordSetUpdate(
zone=zone, recordset=name_or_id, values=kwargs))
new_recordset = self._dns_client.put(
"/v2/zones/{zone_id}/recordsets/{recordset_id}".format(
zone_id=zone_obj['id'], recordset_id=name_or_id), json=kwargs,
error_message="Error updating recordset {0}".format(name_or_id))
return new_recordset
@ -7156,10 +7168,10 @@ class OpenStackCloud(
self.log.debug("Recordset %s not found for deleting", name_or_id)
return False
with _utils.shade_exceptions(
"Error deleting recordset {0}".format(name_or_id)):
self.manager.submit_task(
_tasks.RecordSetDelete(zone=zone['id'], recordset=name_or_id))
self._dns_client.delete(
"/v2/zones/{zone_id}/recordsets/{recordset_id}".format(
zone_id=zone['id'], recordset_id=name_or_id),
error_message="Error deleting recordset {0}".format(name_or_id))
return True

182
shade/tests/unit/test_recordset.py
View File

@ -53,22 +53,10 @@ class TestRecordset(base.RequestsMockTestCase):
"links": {},
"metadata": {
'total_count': 1}}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public', append=['zones'],
qs_elements=['name={0}'.format(zone['id'])]),
json={"zones": [zone]}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public', append=['zones', zone['id']]),
json=zone),
self.get_designate_discovery_mock_dict(),
dict(method='POST',
uri=self.get_mock_url(
'dns', 'public',
append=['zones', zone['id'], 'recordsets']),
append=['v2', 'zones', zone['id'], 'recordsets']),
json=new_recordset,
validate=dict(json=recordset)),
])
@ -92,22 +80,10 @@ class TestRecordset(base.RequestsMockTestCase):
"links": {},
"metadata": {
'total_count': 1}}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public', append=['zones'],
qs_elements=['name={0}'.format(zone['id'])]),
json={"zones": [zone]}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public', append=['zones', zone['id']]),
json=zone),
self.get_designate_discovery_mock_dict(),
dict(method='POST',
uri=self.get_mock_url(
'dns', 'public',
append=['zones', zone['id'], 'recordsets']),
append=['v2', 'zones', zone['id'], 'recordsets']),
status_code=500,
validate=dict(json={
'name': 'www2.example.net.',
@ -115,8 +91,8 @@ class TestRecordset(base.RequestsMockTestCase):
'type': 'A'})),
])
with testtools.ExpectedException(
shade.OpenStackCloudException,
"Unable to create recordset www2.example.net."
shade.exc.OpenStackCloudHTTPError,
"Error creating recordset www2.example.net."
):
self.cloud.create_recordset('1', 'www2.example.net.',
'a', ['192.168.1.2'])
@ -138,58 +114,16 @@ class TestRecordset(base.RequestsMockTestCase):
"links": {},
"metadata": {
'total_count': 1}}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public', append=['zones'],
qs_elements=['name={0}'.format(zone['id'])]),
json={"zones": [zone]}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public', append=['zones'],
qs_elements=['name={0}'.format(zone['id'])]),
json={"zones": [zone]}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public',
append=['zones', zone['id'], 'recordsets'],
qs_elements=['name={0}'.format(
new_recordset['id'])]),
json={"recordsets": [new_recordset]}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public',
append=['zones', zone['id'],
append=['v2', 'zones', zone['id'],
'recordsets', new_recordset['id']]),
json={"recordsets": [new_recordset]}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public', append=['zones'],
qs_elements=['name={0}'.format(zone['id'])]),
json={"zones": [zone]}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public', append=['zones'],
qs_elements=['name={0}'.format(zone['id'])]),
json={"zones": [zone]}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public',
append=['zones', zone['id'], 'recordsets'],
qs_elements=['name={0}'.format(
new_recordset['id'])]),
json={"recordsets": [new_recordset]}),
self.get_designate_discovery_mock_dict(),
json=new_recordset),
dict(method='PUT',
uri=self.get_mock_url(
'dns', 'public',
append=['zones', zone['id'],
append=['v2', 'zones', zone['id'],
'recordsets', new_recordset['id']]),
json=expected_recordset,
validate=dict(json={'ttl': new_ttl}))
@ -208,102 +142,43 @@ class TestRecordset(base.RequestsMockTestCase):
"links": {},
"metadata": {
'total_count': 1}}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public', append=['zones'],
qs_elements=['name={0}'.format(zone['id'])]),
json={"zones": [zone]}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public', append=['zones'],
qs_elements=['name={0}'.format(zone['id'])]),
json={"zones": [zone]}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public',
append=['zones', zone['id'], 'recordsets'],
qs_elements=['name={0}'.format(
new_recordset['id'])]),
json={"recordsets": [new_recordset]}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public',
append=['zones', zone['id'],
append=['v2', 'zones', zone['id'],
'recordsets', new_recordset['id']]),
json={"recordsets": [new_recordset]}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public', append=['zones'],
qs_elements=['name={0}'.format(zone['id'])]),
json={"zones": [zone]}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public', append=['zones'],
qs_elements=['name={0}'.format(zone['id'])]),
json={"zones": [zone]}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public',
append=['zones', zone['id'], 'recordsets'],
qs_elements=['name={0}'.format(
new_recordset['id'])]),
json={"recordsets": [new_recordset]}),
self.get_designate_discovery_mock_dict(),
json=new_recordset),
dict(method='DELETE',
uri=self.get_mock_url(
'dns', 'public',
append=['zones', zone['id'],
append=['v2', 'zones', zone['id'],
'recordsets', new_recordset['id']]),
json={})
])
self.assertTrue(self.cloud.delete_recordset('1', '1'))
self.assert_calls()
def _prepare_get_recordset_calls(self, zone_id, name_or_id):
def test_get_recordset_by_id(self):
self.register_uris([
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public', append=['zones'],
qs_elements=['name={0}'.format(zone['id'])]),
json={"zones": [zone]}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public', append=['zones'],
qs_elements=['name={0}'.format(zone['id'])]),
json={"zones": [zone]}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public',
append=['zones', zone['id'], 'recordsets'],
qs_elements=['name={0}'.format(name_or_id)]),
json={"recordsets": [new_recordset]}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public',
append=['zones', zone['id'],
'recordsets', new_recordset['id']]),
append=['v2', 'zones', '1', 'recordsets', '1']),
json=new_recordset),
])
def test_get_recordset_by_id(self):
recordset = self._prepare_get_recordset_calls('1', '1')
recordset = self.cloud.get_recordset('1', '1')
self.assertEqual(recordset['id'], '1')
self.assert_calls()
def test_get_recordset_by_name(self):
self._prepare_get_recordset_calls('1', new_recordset['name'])
self.register_uris([
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public',
append=['v2', 'zones', '1', 'recordsets',
new_recordset['name']]),
json=new_recordset),
])
recordset = self.cloud.get_recordset('1', new_recordset['name'])
self.assertEqual(new_recordset['name'], recordset['name'])
self.assert_calls()
@ -311,24 +186,11 @@ class TestRecordset(base.RequestsMockTestCase):
def test_get_recordset_not_found_returns_false(self):
recordset_name = "www.nonexistingrecord.net."
self.register_uris([
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public', append=['zones'],
qs_elements=['name={0}'.format(zone['id'])]),
json={"zones": [zone]}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public', append=['zones'],
qs_elements=['name={0}'.format(zone['id'])]),
json={"zones": [zone]}),
self.get_designate_discovery_mock_dict(),
dict(method='GET',
uri=self.get_mock_url(
'dns', 'public',
append=['zones', zone['id'], 'recordsets'],
qs_elements=['name={0}'.format(recordset_name)]),
append=['v2', 'zones', '1', 'recordsets',
recordset_name]),
json=[])
])
recordset = self.cloud.get_recordset('1', recordset_name)