Browse Source

Merge "Adds designate.domain constraint"

changes/09/193409/3
Jenkins 7 years ago committed by Gerrit Code Review
parent
commit
2eafc71af8
  1. 35
      heat/engine/clients/os/designate.py
  2. 156
      heat/tests/clients/test_designate_client.py
  3. 0
      heat/tests/openstack/designate/__init__.py
  4. 2
      setup.cfg

35
heat/engine/clients/os/designate.py

@ -11,21 +11,50 @@
# License for the specific language governing permissions and limitations
# under the License.
from designateclient import client
from designateclient import exceptions
from designateclient import v1 as client
from heat.common import exception as heat_exception
from heat.engine.clients import client_plugin
from heat.engine import constraints
class DesignateClientPlugin(client_plugin.ClientPlugin):
exceptions_module = [exceptions]
service_types = ['dns']
def _create(self):
args = self._get_client_args(service_name='designate',
service_type='dns')
service_type=self.service_types[0])
return client.client('1', **args)
return client.Client(auth_url=args['auth_url'],
project_id=args['project_id'],
token=args['token'](),
endpoint=args['os_endpoint'],
cacert=args['cacert'],
insecure=args['insecure'])
def is_not_found(self, ex):
return isinstance(ex, exceptions.NotFound)
def get_domain_id(self, domain_id_or_name):
try:
domain_obj = self.client().domains.get(domain_id_or_name)
return domain_obj.id
except exceptions.NotFound:
for domain in self.client().domains.list():
if domain.name == domain_id_or_name:
return domain.id
raise heat_exception.EntityNotFound(entity='Designate Domain',
name=domain_id_or_name)
class DesignateDomainConstraint(constraints.BaseCustomConstraint):
expected_exceptions = (heat_exception.EntityNotFound,)
def validate_with_client(self, client, domain):
client.client_plugin('designate').get_domain_id(domain)

156
heat/tests/clients/test_designate_client.py

@ -0,0 +1,156 @@
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import mock
import six
from designateclient import exceptions as designate_exceptions
from designateclient import v1 as designate_client
from heat.common import exception as heat_exception
from heat.engine.clients.os import designate as client
from heat.tests import common
class DesignateDomainConstraintTest(common.HeatTestCase):
def test_expected_exceptions(self):
self.assertEqual((heat_exception.EntityNotFound,),
client.DesignateDomainConstraint.expected_exceptions,
"DesignateDomainConstraint expected exceptions error")
def test_constrain(self):
constrain = client.DesignateDomainConstraint()
client_mock = mock.MagicMock()
client_plugin_mock = mock.MagicMock()
client_plugin_mock.get_domain_id.return_value = None
client_mock.client_plugin.return_value = client_plugin_mock
self.assertIsNone(constrain.validate_with_client(client_mock,
'domain_1'))
client_plugin_mock.get_domain_id.assert_called_once_with('domain_1')
class DesignateClientPluginTest(common.HeatTestCase):
@mock.patch.object(designate_client, 'Client')
@mock.patch.object(client.DesignateClientPlugin, '_get_client_args')
def test_client(self,
get_client_args,
client_designate):
args = dict(
auth_url='auth_url',
project_id='project_id',
token=lambda: '',
os_endpoint='os_endpoint',
cacert='cacert',
insecure='insecure'
)
get_client_args.return_value = args
client_plugin = client.DesignateClientPlugin(
context=mock.MagicMock()
)
client_plugin.client()
# Make sure the right args are created
get_client_args.assert_called_once_with(
service_name='designate',
service_type='dns'
)
# Make sure proper client is created with expected args
client_designate.assert_called_once_with(
auth_url='auth_url',
project_id='project_id',
token='',
endpoint='os_endpoint',
cacert='cacert',
insecure='insecure'
)
class DesignateClientPluginDomainTest(common.HeatTestCase):
sample_uuid = '477e8273-60a7-4c41-b683-fdb0bc7cd152'
sample_name = 'test-domain.com'
def _get_mock_domain(self):
domain = mock.MagicMock()
domain.id = self.sample_uuid
domain.name = self.sample_name
return domain
def setUp(self):
super(DesignateClientPluginDomainTest, self).setUp()
self._client = mock.MagicMock()
self.client_plugin = client.DesignateClientPlugin(
context=mock.MagicMock()
)
@mock.patch.object(client.DesignateClientPlugin, 'client')
def test_get_domain_id(self, client_designate):
self._client.domains.get.return_value = self._get_mock_domain()
client_designate.return_value = self._client
self.assertEqual(self.sample_uuid,
self.client_plugin.get_domain_id(self.sample_uuid))
self._client.domains.get.assert_called_once_with(
self.sample_uuid)
@mock.patch.object(client.DesignateClientPlugin, 'client')
def test_get_domain_id_not_found(self, client_designate):
self._client.domains.get.side_effect = (designate_exceptions
.NotFound)
client_designate.return_value = self._client
ex = self.assertRaises(heat_exception.EntityNotFound,
self.client_plugin.get_domain_id,
self.sample_uuid)
msg = ("The Designate Domain (%(name)s) could not be found." %
{'name': self.sample_uuid})
self.assertEqual(msg, six.text_type(ex))
self._client.domains.get.assert_called_once_with(
self.sample_uuid)
@mock.patch.object(client.DesignateClientPlugin, 'client')
def test_get_domain_id_by_name(self, client_designate):
self._client.domains.get.side_effect = (designate_exceptions
.NotFound)
self._client.domains.list.return_value = [self._get_mock_domain()]
client_designate.return_value = self._client
self.assertEqual(self.sample_uuid,
self.client_plugin.get_domain_id(self.sample_name))
self._client.domains.get.assert_called_once_with(
self.sample_name)
self._client.domains.list.assert_called_once_with()
@mock.patch.object(client.DesignateClientPlugin, 'client')
def test_get_domain_id_by_name_not_found(self, client_designate):
self._client.domains.get.side_effect = (designate_exceptions
.NotFound)
self._client.domains.list.return_value = []
client_designate.return_value = self._client
ex = self.assertRaises(heat_exception.EntityNotFound,
self.client_plugin.get_domain_id,
self.sample_name)
msg = ("The Designate Domain (%(name)s) could not be found." %
{'name': self.sample_name})
self.assertEqual(msg, six.text_type(ex))
self._client.domains.get.assert_called_once_with(
self.sample_name)
self._client.domains.list.assert_called_once_with()

0
heat/tests/openstack/designate/__init__.py

2
setup.cfg

@ -50,6 +50,7 @@ heat.clients =
barbican = heat.engine.clients.os.barbican:BarbicanClientPlugin
ceilometer = heat.engine.clients.os.ceilometer:CeilometerClientPlugin
cinder = heat.engine.clients.os.cinder:CinderClientPlugin
designate = heat.engine.clients.os.designate:DesignateClientPlugin
glance = heat.engine.clients.os.glance:GlanceClientPlugin
heat = heat.engine.clients.os.heat_plugin:HeatClientPlugin
keystone = heat.engine.clients.os.keystone:KeystoneClientPlugin
@ -90,6 +91,7 @@ heat.constraints =
manila.share_snapshot = heat.engine.clients.os.manila:ManilaShareSnapshotConstraint
manila.share_network = heat.engine.clients.os.manila:ManilaShareNetworkConstraint
manila.share_type = heat.engine.clients.os.manila:ManilaShareTypeConstraint
designate.domain = heat.engine.clients.os.designate:DesignateDomainConstraint
heat.stack_lifecycle_plugins =

Loading…
Cancel
Save