openstack
/
heat
Code Issues Proposed changes

Clear implementations for neutron LBaaS v2 

LBaaS v2 was retired a while ago and now the related resources are all
hidden. Thus we can remove all implementation to reduce dependency on
python-neutronclient which is deprecated.

Story: 2010678
Task: 47761
Change-Id: I4de6cb353bc3699d124ea55666c87e4efd7e4350
This commit is contained in:
Takashi Kajinami 2023-03-31 13:52:27 +09:00 committed by Takashi Kajinami
parent 93700aa365
commit 87a8cbe4ac
16 changed files with 14 additions and 2977 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

204
heat/engine/resources/openstack/neutron/lbaas/health_monitor.py
View File

@ -15,14 +15,11 @@
# under the License.
from heat.common.i18n import _
from heat.engine import attributes
from heat.engine import constraints
from heat.engine import properties
from heat.engine.resources.openstack.neutron import neutron
from heat.engine.resources.openstack.heat import none_resource
from heat.engine import support
class HealthMonitor(neutron.NeutronResource):
class HealthMonitor(none_resource.NoneResource):
"""A resource to handle load balancer health monitors.
This resource creates and manages Neutron LBaaS v2 healthmonitors,
@ -36,203 +33,6 @@ class HealthMonitor(neutron.NeutronResource):
previous_status=support.SupportStatus(version='6.0.0')
)
required_service_extension = 'lbaasv2'
entity = 'lbaas_healthmonitor'
res_info_key = 'healthmonitor'
# Properties inputs for the resources create/update.
PROPERTIES = (
ADMIN_STATE_UP, DELAY, EXPECTED_CODES, HTTP_METHOD,
MAX_RETRIES, POOL, TIMEOUT, TYPE, URL_PATH, TENANT_ID
) = (
'admin_state_up', 'delay', 'expected_codes', 'http_method',
'max_retries', 'pool', 'timeout', 'type', 'url_path', 'tenant_id'
)
# Supported HTTP methods
HTTP_METHODS = (
GET, HEAT, POST, PUT, DELETE, TRACE, OPTIONS,
CONNECT, PATCH
) = (
'GET', 'HEAD', 'POST', 'PUT', 'DELETE', 'TRACE', 'OPTIONS',
'CONNECT', 'PATCH'
)
# Supported output attributes of the resources.
ATTRIBUTES = (POOLS_ATTR) = ('pools')
properties_schema = {
ADMIN_STATE_UP: properties.Schema(
properties.Schema.BOOLEAN,
_('The administrative state of the health monitor.'),
default=True,
update_allowed=True
),
DELAY: properties.Schema(
properties.Schema.INTEGER,
_('The minimum time in milliseconds between regular connections '
'of the member.'),
required=True,
update_allowed=True,
constraints=[constraints.Range(min=0)]
),
EXPECTED_CODES: properties.Schema(
properties.Schema.STRING,
_('The HTTP status codes expected in response from the '
'member to declare it healthy. Specify one of the following '
'values: a single value, such as 200. a list, such as 200, 202. '
'a range, such as 200-204.'),
update_allowed=True,
default='200'
),
HTTP_METHOD: properties.Schema(
properties.Schema.STRING,
_('The HTTP method used for requests by the monitor of type '
'HTTP.'),
update_allowed=True,
default=GET,
constraints=[constraints.AllowedValues(HTTP_METHODS)]
),
MAX_RETRIES: properties.Schema(
properties.Schema.INTEGER,
_('Number of permissible connection failures before changing the '
'member status to INACTIVE.'),
required=True,
update_allowed=True,
constraints=[constraints.Range(min=1, max=10)],
),
POOL: properties.Schema(
properties.Schema.STRING,
_('ID or name of the load balancing pool.'),
required=True,
constraints=[
constraints.CustomConstraint('neutron.lbaas.pool')
]
),
TIMEOUT: properties.Schema(
properties.Schema.INTEGER,
_('Maximum number of milliseconds for a monitor to wait for a '
'connection to be established before it times out.'),
required=True,
update_allowed=True,
constraints=[constraints.Range(min=0)]
),
TYPE: properties.Schema(
properties.Schema.STRING,
_('One of predefined health monitor types.'),
required=True,
constraints=[
constraints.AllowedValues(['PING', 'TCP', 'HTTP', 'HTTPS']),
]
),
URL_PATH: properties.Schema(
properties.Schema.STRING,
_('The HTTP path used in the HTTP request used by the monitor to '
'test a member health. A valid value is a string the begins '
'with a forward slash (/).'),
update_allowed=True,
default='/'
),
TENANT_ID: properties.Schema(
properties.Schema.STRING,
_('ID of the tenant who owns the health monitor.')
)
}
attributes_schema = {
POOLS_ATTR: attributes.Schema(
_('The list of Pools related to this monitor.'),
type=attributes.Schema.LIST
)
}
def __init__(self, name, definition, stack):
super(HealthMonitor, self).__init__(name, definition, stack)
self._lb_id = None
@property
def lb_id(self):
if self._lb_id is None:
client_plugin = self.client_plugin()
pool_id = client_plugin.find_resourceid_by_name_or_id(
client_plugin.RES_TYPE_LB_POOL,
self.properties[self.POOL])
pool = self.client().show_lbaas_pool(pool_id)['pool']
listener_id = pool['listeners'][0]['id']
listener = self.client().show_listener(listener_id)['listener']
self._lb_id = listener['loadbalancers'][0]['id']
return self._lb_id
def _check_lb_status(self):
return self.client_plugin().check_lb_status(self.lb_id)
def handle_create(self):
properties = self.prepare_properties(
self.properties,
self.physical_resource_name())
self.client_plugin().resolve_pool(
properties, self.POOL, 'pool_id')
return properties
def check_create_complete(self, properties):
if self.resource_id is None:
try:
healthmonitor = self.client().create_lbaas_healthmonitor(
{'healthmonitor': properties})['healthmonitor']
self.resource_id_set(healthmonitor['id'])
except Exception as ex:
if self.client_plugin().is_invalid(ex):
return False
raise
return self._check_lb_status()
def handle_update(self, json_snippet, tmpl_diff, prop_diff):
self._update_called = False
return prop_diff
def check_update_complete(self, prop_diff):
if not prop_diff:
return True
if not self._update_called:
try:
self.client().update_lbaas_healthmonitor(
self.resource_id, {'healthmonitor': prop_diff})
self._update_called = True
except Exception as ex:
if self.client_plugin().is_invalid(ex):
return False
raise
return self._check_lb_status()
def handle_delete(self):
self._delete_called = False
def check_delete_complete(self, data):
if self.resource_id is None:
return True
if not self._delete_called:
try:
self.client().delete_lbaas_healthmonitor(self.resource_id)
self._delete_called = True
except Exception as ex:
if self.client_plugin().is_invalid(ex):
return False
elif self.client_plugin().is_not_found(ex):
return True
raise
return self._check_lb_status()
def resource_mapping():
return {

237
heat/engine/resources/openstack/neutron/lbaas/l7policy.py
View File

@ -11,17 +11,12 @@
# License for the specific language governing permissions and limitations
# under the License.
from heat.common import exception
from heat.common.i18n import _
from heat.engine import attributes
from heat.engine import constraints
from heat.engine import properties
from heat.engine.resources.openstack.neutron import neutron
from heat.engine.resources.openstack.heat import none_resource
from heat.engine import support
from heat.engine import translation
class L7Policy(neutron.NeutronResource):
class L7Policy(none_resource.NoneResource):
"""A resource for managing LBaaS v2 L7Policies.
This resource manages Neutron-LBaaS v2 L7Policies, which represent
@ -42,234 +37,6 @@ class L7Policy(neutron.NeutronResource):
previous_status=support.SupportStatus(version='7.0.0')
)
required_service_extension = 'lbaasv2'
entity = 'lbaas_l7policy'
res_info_key = 'l7policy'
PROPERTIES = (
NAME, DESCRIPTION, ADMIN_STATE_UP, ACTION,
REDIRECT_POOL, REDIRECT_URL, POSITION, LISTENER
) = (
'name', 'description', 'admin_state_up', 'action',
'redirect_pool', 'redirect_url', 'position', 'listener'
)
L7ACTIONS = (
REJECT, REDIRECT_TO_POOL, REDIRECT_TO_URL
) = (
'REJECT', 'REDIRECT_TO_POOL', 'REDIRECT_TO_URL'
)
ATTRIBUTES = (RULES_ATTR) = ('rules')
properties_schema = {
NAME: properties.Schema(
properties.Schema.STRING,
_('Name of the policy.'),
update_allowed=True
),
DESCRIPTION: properties.Schema(
properties.Schema.STRING,
_('Description of the policy.'),
update_allowed=True
),
ADMIN_STATE_UP: properties.Schema(
properties.Schema.BOOLEAN,
_('The administrative state of the policy.'),
default=True,
update_allowed=True
),
ACTION: properties.Schema(
properties.Schema.STRING,
_('Action type of the policy.'),
required=True,
constraints=[constraints.AllowedValues(L7ACTIONS)],
update_allowed=True
),
REDIRECT_POOL: properties.Schema(
properties.Schema.STRING,
_('ID or name of the pool for REDIRECT_TO_POOL action type.'),
constraints=[
constraints.CustomConstraint('neutron.lbaas.pool')
],
update_allowed=True
),
REDIRECT_URL: properties.Schema(
properties.Schema.STRING,
_('URL for REDIRECT_TO_URL action type. '
'This should be a valid URL string.'),
update_allowed=True
),
POSITION: properties.Schema(
properties.Schema.NUMBER,
_('L7 policy position in ordered policies list. This must be '
'an integer starting from 1. If not specified, policy will be '
'placed at the tail of existing policies list.'),
constraints=[constraints.Range(min=1)],
update_allowed=True
),
LISTENER: properties.Schema(
properties.Schema.STRING,
_('ID or name of the listener this policy belongs to.'),
required=True,
constraints=[
constraints.CustomConstraint('neutron.lbaas.listener')
]
),
}
attributes_schema = {
RULES_ATTR: attributes.Schema(
_('L7Rules associated with this policy.'),
type=attributes.Schema.LIST
),
}
def translation_rules(self, props):
return [
translation.TranslationRule(
props,
translation.TranslationRule.RESOLVE,
[self.LISTENER],
client_plugin=self.client_plugin(),
finder='find_resourceid_by_name_or_id',
entity='listener'
),
]
def __init__(self, name, definition, stack):
super(L7Policy, self).__init__(name, definition, stack)
self._lb_id = None
@property
def lb_id(self):
if self._lb_id is None:
listener_id = self.properties[self.LISTENER]
listener = self.client().show_listener(listener_id)['listener']
self._lb_id = listener['loadbalancers'][0]['id']
return self._lb_id
def validate(self):
res = super(L7Policy, self).validate()
if res:
return res
if (self.properties[self.ACTION] == self.REJECT and
(self.properties[self.REDIRECT_POOL] is not None or
self.properties[self.REDIRECT_URL] is not None)):
msg = (_('Properties %(pool)s and %(url)s are not required when '
'%(action)s type is set to %(action_type)s.') %
{'pool': self.REDIRECT_POOL,
'url': self.REDIRECT_URL,
'action': self.ACTION,
'action_type': self.REJECT})
raise exception.StackValidationFailed(message=msg)
if self.properties[self.ACTION] == self.REDIRECT_TO_POOL:
if self.properties[self.REDIRECT_URL] is not None:
raise exception.ResourcePropertyValueDependency(
prop1=self.REDIRECT_URL,
prop2=self.ACTION,
value=self.REDIRECT_TO_URL)
if self.properties[self.REDIRECT_POOL] is None:
msg = (_('Property %(pool)s is required when %(action)s '
'type is set to %(action_type)s.') %
{'pool': self.REDIRECT_POOL,
'action': self.ACTION,
'action_type': self.REDIRECT_TO_POOL})
raise exception.StackValidationFailed(message=msg)
if self.properties[self.ACTION] == self.REDIRECT_TO_URL:
if self.properties[self.REDIRECT_POOL] is not None:
raise exception.ResourcePropertyValueDependency(
prop1=self.REDIRECT_POOL,
prop2=self.ACTION,
value=self.REDIRECT_TO_POOL)
if self.properties[self.REDIRECT_URL] is None:
msg = (_('Property %(url)s is required when %(action)s '
'type is set to %(action_type)s.') %
{'url': self.REDIRECT_URL,
'action': self.ACTION,
'action_type': self.REDIRECT_TO_URL})
raise exception.StackValidationFailed(message=msg)
def _check_lb_status(self):
return self.client_plugin().check_lb_status(self.lb_id)
def handle_create(self):
properties = self.prepare_properties(
self.properties,
self.physical_resource_name())
properties['listener_id'] = properties.pop(self.LISTENER)
if self.properties[self.REDIRECT_POOL] is not None:
self.client_plugin().resolve_pool(
properties, self.REDIRECT_POOL, 'redirect_pool_id')
return properties
def check_create_complete(self, properties):
if self.resource_id is None:
try:
l7policy = self.client().create_lbaas_l7policy(
{'l7policy': properties})['l7policy']
self.resource_id_set(l7policy['id'])
except Exception as ex:
if self.client_plugin().is_invalid(ex):
return False
raise
return self._check_lb_status()
def handle_update(self, json_snippet, tmpl_diff, prop_diff):
self._update_called = False
if self.REDIRECT_POOL in prop_diff:
if prop_diff[self.REDIRECT_POOL] is not None:
self.client_plugin().resolve_pool(
prop_diff, self.REDIRECT_POOL, 'redirect_pool_id')
else:
prop_diff.pop(self.REDIRECT_POOL)
prop_diff['redirect_pool_id'] = None
return prop_diff
def check_update_complete(self, prop_diff):
if not prop_diff:
return True
if not self._update_called:
try:
self.client().update_lbaas_l7policy(
self.resource_id, {'l7policy': prop_diff})
self._update_called = True
except Exception as ex:
if self.client_plugin().is_invalid(ex):
return False
raise
return self._check_lb_status()
def handle_delete(self):
self._delete_called = False
def check_delete_complete(self, data):
if self.resource_id is None:
return True
if not self._delete_called:
try:
self.client().delete_lbaas_l7policy(self.resource_id)
self._delete_called = True
except Exception as ex:
if self.client_plugin().is_invalid(ex):
return False
elif self.client_plugin().is_not_found(ex):
return True
raise
return self._check_lb_status()
def resource_mapping():
return {

190
heat/engine/resources/openstack/neutron/lbaas/l7rule.py
View File

@ -11,15 +11,12 @@
# License for the specific language governing permissions and limitations
# under the License.
from heat.common import exception
from heat.common.i18n import _
from heat.engine import constraints
from heat.engine import properties
from heat.engine.resources.openstack.neutron import neutron
from heat.engine.resources.openstack.heat import none_resource
from heat.engine import support
class L7Rule(neutron.NeutronResource):
class L7Rule(none_resource.NoneResource):
"""A resource for managing LBaaS v2 L7Rules.
This resource manages Neutron-LBaaS v2 L7Rules, which represent
@ -34,189 +31,6 @@ class L7Rule(neutron.NeutronResource):
previous_status=support.SupportStatus(version='7.0.0')
)
required_service_extension = 'lbaasv2'
entity = 'lbaas_l7rule'
res_info_key = 'rule'
PROPERTIES = (
ADMIN_STATE_UP, L7POLICY, TYPE, COMPARE_TYPE,
INVERT, KEY, VALUE
) = (
'admin_state_up', 'l7policy', 'type', 'compare_type',
'invert', 'key', 'value'
)
L7RULE_TYPES = (
HOST_NAME, PATH, FILE_TYPE, HEADER, COOKIE
) = (
'HOST_NAME', 'PATH', 'FILE_TYPE', 'HEADER', 'COOKIE'
)
L7COMPARE_TYPES = (
REGEX, STARTS_WITH, ENDS_WITH, CONTAINS, EQUAL_TO
) = (
'REGEX', 'STARTS_WITH', 'ENDS_WITH', 'CONTAINS', 'EQUAL_TO'
)
properties_schema = {
ADMIN_STATE_UP: properties.Schema(
properties.Schema.BOOLEAN,
_('The administrative state of the rule.'),
default=True,
update_allowed=True
),
L7POLICY: properties.Schema(
properties.Schema.STRING,
_('ID or name of L7 policy this rule belongs to.'),
required=True
),
TYPE: properties.Schema(
properties.Schema.STRING,
_('Rule type.'),
constraints=[constraints.AllowedValues(L7RULE_TYPES)],
update_allowed=True,
required=True
),
COMPARE_TYPE: properties.Schema(
properties.Schema.STRING,
_('Rule compare type.'),
constraints=[constraints.AllowedValues(L7COMPARE_TYPES)],
update_allowed=True,
required=True
),
INVERT: properties.Schema(
properties.Schema.BOOLEAN,
_('Invert the compare type.'),
default=False,
update_allowed=True
),
KEY: properties.Schema(
properties.Schema.STRING,
_('Key to compare. Relevant for HEADER and COOKIE types only.'),
update_allowed=True
),
VALUE: properties.Schema(
properties.Schema.STRING,
_('Value to compare.'),
update_allowed=True,
required=True
)
}
def __init__(self, name, definition, stack):
super(L7Rule, self).__init__(name, definition, stack)
self._l7p_id = None
self._lb_id = None
@property
def l7policy_id(self):
client_plugin = self.client_plugin()
if self._l7p_id is None:
self._l7p_id = client_plugin.find_resourceid_by_name_or_id(
client_plugin.RES_TYPE_LB_L7POLICY,
self.properties[self.L7POLICY])
return self._l7p_id
@property
def lb_id(self):
if self._lb_id is None:
policy = self.client().show_lbaas_l7policy(
self.l7policy_id)['l7policy']
listener_id = policy['listener_id']
listener = self.client().show_listener(listener_id)['listener']
self._lb_id = listener['loadbalancers'][0]['id']
return self._lb_id
def _check_lb_status(self):
return self.client_plugin().check_lb_status(self.lb_id)
def validate(self):
res = super(L7Rule, self).validate()
if res:
return res
if (self.properties[self.TYPE] in (self.HEADER, self.COOKIE) and
self.properties[self.KEY] is None):
msg = (_('Property %(key)s is missing. '
'This property should be specified for '
'rules of %(header)s and %(cookie)s types.') %
{'key': self.KEY,
'header': self.HEADER,
'cookie': self.COOKIE})
raise exception.StackValidationFailed(message=msg)
def handle_create(self):
rule_args = dict((k, v) for k, v in self.properties.items()
if k != self.L7POLICY)
return rule_args
def check_create_complete(self, rule_args):
if self.resource_id is None:
try:
l7rule = self.client().create_lbaas_l7rule(
self.l7policy_id,
{'rule': rule_args})['rule']
self.resource_id_set(l7rule['id'])
except Exception as ex:
if self.client_plugin().is_invalid(ex):
return False
raise
return self._check_lb_status()
def _res_get_args(self):
return [self.resource_id, self.l7policy_id]
def handle_update(self, json_snippet, tmpl_diff, prop_diff):
self._update_called = False
if (prop_diff.get(self.TYPE) in (self.COOKIE, self.HEADER) and
prop_diff.get(self.KEY) is None):
prop_diff[self.KEY] = tmpl_diff['Properties'].get(self.KEY)
return prop_diff
def check_update_complete(self, prop_diff):
if not prop_diff:
return True
if not self._update_called:
try:
self.client().update_lbaas_l7rule(
self.resource_id,
self.l7policy_id,
{'rule': prop_diff})
self._update_called = True
except Exception as ex:
if self.client_plugin().is_invalid(ex):
return False
raise
return self._check_lb_status()
def handle_delete(self):
self._delete_called = False
def check_delete_complete(self, data):
if self.resource_id is None:
return True
if not self._delete_called:
try:
self.client().delete_lbaas_l7rule(
self.resource_id,
self.l7policy_id)
self._delete_called = True
except Exception as ex:
if self.client_plugin().is_invalid(ex):
return False
elif self.client_plugin().is_not_found(ex):
return True
raise
return self._check_lb_status()
def resource_mapping():
return {

250
heat/engine/resources/openstack/neutron/lbaas/listener.py
View File

@ -14,17 +14,12 @@
# License for the specific language governing permissions and limitations
# under the License.
from heat.common import exception
from heat.common.i18n import _
from heat.engine import attributes
from heat.engine import constraints
from heat.engine import properties
from heat.engine.resources.openstack.neutron import neutron
from heat.engine.resources.openstack.heat import none_resource
from heat.engine import support
from heat.engine import translation
class Listener(neutron.NeutronResource):
class Listener(none_resource.NoneResource):
"""A resource for managing LBaaS v2 Listeners.
This resource creates and manages Neutron LBaaS v2 Listeners,
@ -38,247 +33,6 @@ class Listener(neutron.NeutronResource):
previous_status=support.SupportStatus(version='6.0.0')
)
required_service_extension = 'lbaasv2'
entity = 'listener'
PROPERTIES = (
PROTOCOL_PORT, PROTOCOL, LOADBALANCER, DEFAULT_POOL, NAME,
ADMIN_STATE_UP, DESCRIPTION, DEFAULT_TLS_CONTAINER_REF,
SNI_CONTAINER_REFS, CONNECTION_LIMIT, TENANT_ID
) = (
'protocol_port', 'protocol', 'loadbalancer', 'default_pool', 'name',
'admin_state_up', 'description', 'default_tls_container_ref',
'sni_container_refs', 'connection_limit', 'tenant_id'
)
PROTOCOLS = (
TCP, HTTP, HTTPS, TERMINATED_HTTPS,
) = (
'TCP', 'HTTP', 'HTTPS', 'TERMINATED_HTTPS',
)
ATTRIBUTES = (
LOADBALANCERS_ATTR, DEFAULT_POOL_ID_ATTR
) = (
'loadbalancers', 'default_pool_id'
)
properties_schema = {
PROTOCOL_PORT: properties.Schema(
properties.Schema.INTEGER,
_('TCP or UDP port on which to listen for client traffic.'),
required=True,
constraints=[
constraints.Range(1, 65535),
]
),
PROTOCOL: properties.Schema(
properties.Schema.STRING,
_('Protocol on which to listen for the client traffic.'),
required=True,
constraints=[
constraints.AllowedValues(PROTOCOLS),
]
),
LOADBALANCER: properties.Schema(
properties.Schema.STRING,
_('ID or name of the load balancer with which listener '
'is associated.'),
constraints=[
constraints.CustomConstraint('neutron.lbaas.loadbalancer')
]
),
DEFAULT_POOL: properties.Schema(
properties.Schema.STRING,
_('ID or name of the default pool for the listener. Requires '
'shared_pools service extension.'),
update_allowed=True,
constraints=[
constraints.CustomConstraint('neutron.lbaas.pool')
],
support_status=support.SupportStatus(version='9.0.0')
),
NAME: properties.Schema(
properties.Schema.STRING,
_('Name of this listener.'),
update_allowed=True
),
ADMIN_STATE_UP: properties.Schema(
properties.Schema.BOOLEAN,
_('The administrative state of this listener.'),
update_allowed=True,
default=True
),
DESCRIPTION: properties.Schema(
properties.Schema.STRING,
_('Description of this listener.'),
update_allowed=True,
default=''
),
DEFAULT_TLS_CONTAINER_REF: properties.Schema(
properties.Schema.STRING,
_('Default TLS container reference to retrieve TLS '
'information.'),
update_allowed=True
),
SNI_CONTAINER_REFS: properties.Schema(
properties.Schema.LIST,
_('List of TLS container references for SNI.'),
update_allowed=True
),
CONNECTION_LIMIT: properties.Schema(
properties.Schema.INTEGER,
_('The maximum number of connections permitted for this '
'load balancer. Defaults to -1, which is infinite.'),
update_allowed=True,
default=-1,
constraints=[
constraints.Range(min=-1),
]
),
TENANT_ID: properties.Schema(
properties.Schema.STRING,
_('The ID of the tenant who owns the listener.')
),
}
attributes_schema = {
LOADBALANCERS_ATTR: attributes.Schema(
_('ID of the load balancer this listener is associated to.'),
type=attributes.Schema.LIST
),
DEFAULT_POOL_ID_ATTR: attributes.Schema(
_('ID of the default pool this listener is associated to.'),
type=attributes.Schema.STRING
)
}
def __init__(self, name, definition, stack):
super(Listener, self).__init__(name, definition, stack)
self._lb_id = None
def translation_rules(self, props):
client_plugin = self.client_plugin()
return [
translation.TranslationRule(
props,
translation.TranslationRule.RESOLVE,
[self.LOADBALANCER],
client_plugin=client_plugin,
finder='find_resourceid_by_name_or_id',
entity=client_plugin.RES_TYPE_LOADBALANCER
),
translation.TranslationRule(
props,
translation.TranslationRule.RESOLVE,
[self.DEFAULT_POOL],
client_plugin=client_plugin,
finder='find_resourceid_by_name_or_id',
entity=client_plugin.RES_TYPE_LB_POOL
),
]
def validate(self):
res = super(Listener, self).validate()
if res:
return res
if (self.properties[self.LOADBALANCER] is None
and self.properties[self.DEFAULT_POOL] is None):
raise exception.PropertyUnspecifiedError(self.LOADBALANCER,
self.DEFAULT_POOL)
if self.properties[self.PROTOCOL] == self.TERMINATED_HTTPS:
if self.properties[self.DEFAULT_TLS_CONTAINER_REF] is None:
msg = (_('Property %(ref)s required when protocol is '
'%(term)s.') % {'ref': self.DEFAULT_TLS_CONTAINER_REF,
'term': self.TERMINATED_HTTPS})
raise exception.StackValidationFailed(message=msg)
@property
def lb_id(self):
if self._lb_id:
return self._lb_id
self._lb_id = self.properties[self.LOADBALANCER]
if self._lb_id is None:
pool_id = self.properties[self.DEFAULT_POOL]
pool = self.client().show_pool(pool_id)['pool']
self._lb_id = pool['loadbalancers'][0]['id']
return self._lb_id
def _check_lb_status(self):
return self.client_plugin().check_lb_status(self.lb_id)
def handle_create(self):
properties = self.prepare_properties(
self.properties,
self.physical_resource_name())
if self.LOADBALANCER in properties:
properties['loadbalancer_id'] = properties.pop(self.LOADBALANCER)
if self.DEFAULT_POOL in properties:
properties['default_pool_id'] = properties.pop(self.DEFAULT_POOL)
return properties
def check_create_complete(self, properties):
if self.resource_id is None:
try:
listener = self.client().create_listener(
{'listener': properties})['listener']
self.resource_id_set(listener['id'])
except Exception as ex:
if self.client_plugin().is_invalid(ex):
return False
raise
return self._check_lb_status()
def handle_update(self, json_snippet, tmpl_diff, prop_diff):
self._update_called = False
self.properties = json_snippet.properties(self.properties_schema,
self.context)
return prop_diff
def check_update_complete(self, prop_diff):
if not prop_diff:
return True
if self.DEFAULT_POOL in prop_diff:
prop_diff['default_pool_id'] = prop_diff.pop(self.DEFAULT_POOL)
if not self._update_called:
try:
self.client().update_listener(self.resource_id,
{'listener': prop_diff})
self._update_called = True
except Exception as ex:
if self.client_plugin().is_invalid(ex):
return False
raise
return self._check_lb_status()
def handle_delete(self):
self._delete_called = False
def check_delete_complete(self, data):
if self.resource_id is None:
return True
if not self._delete_called:
try:
self.client().delete_listener(self.resource_id)
self._delete_called = True
except Exception as ex:
if self.client_plugin().is_invalid(ex):
return False
elif self.client_plugin().is_not_found(ex):
return True
raise
return self._check_lb_status()
def resource_mapping():
return {

159
heat/engine/resources/openstack/neutron/lbaas/loadbalancer.py
View File

@ -14,19 +14,12 @@
# License for the specific language governing permissions and limitations
# under the License.
from neutronclient.common import exceptions
from heat.common import exception
from heat.common.i18n import _
from heat.engine import attributes
from heat.engine import constraints
from heat.engine import properties
from heat.engine.resources.openstack.neutron import neutron
from heat.engine.resources.openstack.heat import none_resource
from heat.engine import support
from heat.engine import translation
class LoadBalancer(neutron.NeutronResource):
class LoadBalancer(none_resource.NoneResource):
"""A resource for creating LBaaS v2 Load Balancers.
This resource creates and manages Neutron LBaaS v2 Load Balancers,
@ -40,154 +33,6 @@ class LoadBalancer(neutron.NeutronResource):
previous_status=support.SupportStatus(version='6.0.0')
)
required_service_extension = 'lbaasv2'
entity = 'loadbalancer'
PROPERTIES = (
DESCRIPTION, NAME, PROVIDER, VIP_ADDRESS, VIP_SUBNET,
ADMIN_STATE_UP, TENANT_ID
) = (
'description', 'name', 'provider', 'vip_address', 'vip_subnet',
'admin_state_up', 'tenant_id'
)
ATTRIBUTES = (
VIP_ADDRESS_ATTR, VIP_PORT_ATTR, VIP_SUBNET_ATTR, POOLS_ATTR
) = (
'vip_address', 'vip_port_id', 'vip_subnet_id', 'pools'
)
properties_schema = {
DESCRIPTION: properties.Schema(
properties.Schema.STRING,
_('Description of this Load Balancer.'),
update_allowed=True,
default=''
),
NAME: properties.Schema(
properties.Schema.STRING,
_('Name of this Load Balancer.'),
update_allowed=True
),
PROVIDER: properties.Schema(
properties.Schema.STRING,
_('Provider for this Load Balancer.'),
constraints=[
constraints.CustomConstraint('neutron.lbaas.provider')
],
),
VIP_ADDRESS: properties.Schema(
properties.Schema.STRING,
_('IP address for the VIP.'),
constraints=[
constraints.CustomConstraint('ip_addr')
],
),
VIP_SUBNET: properties.Schema(
properties.Schema.STRING,
_('The name or ID of the subnet on which to allocate the VIP '
'address.'),
constraints=[
constraints.CustomConstraint('neutron.subnet')
],
required=True
),
ADMIN_STATE_UP: properties.Schema(
properties.Schema.BOOLEAN,
_('The administrative state of this Load Balancer.'),
default=True,
update_allowed=True
),
TENANT_ID: properties.Schema(
properties.Schema.STRING,
_('The ID of the tenant who owns the Load Balancer. Only '
'administrative users can specify a tenant ID other than '
'their own.'),
constraints=[
constraints.CustomConstraint('keystone.project')
],
)
}
attributes_schema = {
VIP_ADDRESS_ATTR: attributes.Schema(
_('The VIP address of the LoadBalancer.'),
type=attributes.Schema.STRING
),
VIP_PORT_ATTR: attributes.Schema(
_('The VIP port of the LoadBalancer.'),
type=attributes.Schema.STRING
),
VIP_SUBNET_ATTR: attributes.Schema(
_('The VIP subnet of the LoadBalancer.'),
type=attributes.Schema.STRING
),
POOLS_ATTR: attributes.Schema(
_('Pools this LoadBalancer is associated with.'),
type=attributes.Schema.LIST,
support_status=support.SupportStatus(version='9.0.0')
),
}
def translation_rules(self, props):
client_plugin = self.client_plugin()
return [
translation.TranslationRule(
props,
translation.TranslationRule.RESOLVE,
[self.VIP_SUBNET],
client_plugin=client_plugin,
finder='find_resourceid_by_name_or_id',
entity=client_plugin.RES_TYPE_SUBNET
),
]
def handle_create(self):
properties = self.prepare_properties(
self.properties,
self.physical_resource_name()
)
properties['vip_subnet_id'] = properties.pop(self.VIP_SUBNET)
lb = self.client().create_loadbalancer(
{'loadbalancer': properties})['loadbalancer']
self.resource_id_set(lb['id'])
def check_create_complete(self, data):
return self.client_plugin().check_lb_status(self.resource_id)
def handle_update(self, json_snippet, tmpl_diff, prop_diff):
if prop_diff:
self.client().update_loadbalancer(
self.resource_id,
{'loadbalancer': prop_diff})
return prop_diff
def check_update_complete(self, prop_diff):
if prop_diff:
return self.client_plugin().check_lb_status(self.resource_id)
return True
def handle_delete(self):
pass
def check_delete_complete(self, data):
if self.resource_id is None:
return True
try:
try:
if self.client_plugin().check_lb_status(self.resource_id):
self.client().delete_loadbalancer(self.resource_id)
except exception.ResourceInError:
# Still try to delete loadbalancer in error state
self.client().delete_loadbalancer(self.resource_id)
except exceptions.NotFound:
# Resource is gone
return True
return False
def resource_mapping():
return {

268
heat/engine/resources/openstack/neutron/lbaas/pool.py
View File

@ -14,17 +14,12 @@
# License for the specific language governing permissions and limitations
# under the License.
from heat.common import exception
from heat.common.i18n import _
from heat.engine import attributes
from heat.engine import constraints
from heat.engine import properties
from heat.engine.resources.openstack.neutron import neutron
from heat.engine.resources.openstack.heat import none_resource
from heat.engine import support
from heat.engine import translation
class Pool(neutron.NeutronResource):
class Pool(none_resource.NoneResource):
"""A resource for managing LBaaS v2 Pools.
This resources manages Neutron-LBaaS v2 Pools, which represent a group
@ -39,265 +34,6 @@ class Pool(neutron.NeutronResource):
previous_status=support.SupportStatus(version='6.0.0')
)
required_service_extension = 'lbaasv2'
entity = 'lbaas_pool'
res_info_key = 'pool'
PROPERTIES = (
ADMIN_STATE_UP, DESCRIPTION, SESSION_PERSISTENCE, NAME,
LB_ALGORITHM, LISTENER, LOADBALANCER, PROTOCOL,
SESSION_PERSISTENCE_TYPE, SESSION_PERSISTENCE_COOKIE_NAME,
) = (
'admin_state_up', 'description', 'session_persistence', 'name',
'lb_algorithm', 'listener', 'loadbalancer', 'protocol',
'type', 'cookie_name'
)
SESSION_PERSISTENCE_TYPES = (
SOURCE_IP, HTTP_COOKIE, APP_COOKIE
) = (
'SOURCE_IP', 'HTTP_COOKIE', 'APP_COOKIE'
)
ATTRIBUTES = (
HEALTHMONITOR_ID_ATTR, LISTENERS_ATTR, MEMBERS_ATTR
) = (
'healthmonitor_id', 'listeners', 'members'
)
properties_schema = {
ADMIN_STATE_UP: properties.Schema(
properties.Schema.BOOLEAN,
_('The administrative state of this pool.'),
default=True,
update_allowed=True
),
DESCRIPTION: properties.Schema(
properties.Schema.STRING,
_('Description of this pool.'),
update_allowed=True,
default=''
),
SESSION_PERSISTENCE: properties.Schema(
properties.Schema.MAP,
_('Configuration of session persistence.'),
schema={
SESSION_PERSISTENCE_TYPE: properties.Schema(
properties.Schema.STRING,
_('Method of implementation of session '
'persistence feature.'),
required=True,
constraints=[constraints.AllowedValues(
SESSION_PERSISTENCE_TYPES
)]
),
SESSION_PERSISTENCE_COOKIE_NAME: properties.Schema(
properties.Schema.STRING,
_('Name of the cookie, '
'required if type is APP_COOKIE.')
)
},
),
NAME: properties.Schema(
properties.Schema.STRING,
_('Name of this pool.'),
update_allowed=True
),
LB_ALGORITHM: properties.Schema(
properties.Schema.STRING,
_('The algorithm used to distribute load between the members of '
'the pool.'),
required=True,
constraints=[
constraints.AllowedValues(['ROUND_ROBIN',
'LEAST_CONNECTIONS', 'SOURCE_IP']),
],
update_allowed=True,
),
LISTENER: properties.Schema(
properties.Schema.STRING,
_('Listener name or ID to be associated with this pool.'),
constraints=[
constraints.CustomConstraint('neutron.lbaas.listener')
]
),
LOADBALANCER: properties.Schema(
properties.Schema.STRING,
_('Loadbalancer name or ID to be associated with this pool. '
'Requires shared_pools service extension.'),
constraints=[
constraints.CustomConstraint('neutron.lbaas.loadbalancer')
],
support_status=support.SupportStatus(version='9.0.0')
),
PROTOCOL: properties.Schema(
properties.Schema.STRING,
_('Protocol of the pool.'),
required=True,
constraints=[
constraints.AllowedValues(['TCP', 'HTTP', 'HTTPS']),
]
),
}
attributes_schema = {
HEALTHMONITOR_ID_ATTR: attributes.Schema(
_('ID of the health monitor associated with this pool.'),
type=attributes.Schema.STRING
),
LISTENERS_ATTR: attributes.Schema(
_('Listener associated with this pool.'),
type=attributes.Schema.STRING
),
MEMBERS_ATTR: attributes.Schema(
_('Members associated with this pool.'),
cache_mode=attributes.Schema.CACHE_NONE,
type=attributes.Schema.LIST
),
}
def translation_rules(self, props):
client_plugin = self.client_plugin()
return [
translation.TranslationRule(
props,
translation.TranslationRule.RESOLVE,
[self.LISTENER],
client_plugin=client_plugin,
finder='find_resourceid_by_name_or_id',
entity=client_plugin.RES_TYPE_LB_LISTENER
),
translation.TranslationRule(
props,
translation.TranslationRule.RESOLVE,
[self.LOADBALANCER],
client_plugin=client_plugin,
finder='find_resourceid_by_name_or_id',
entity=client_plugin.RES_TYPE_LOADBALANCER
),
]
def __init__(self, name, definition, stack):
super(Pool, self).__init__(name, definition, stack)
self._lb_id = None
@property
def lb_id(self):
if self._lb_id:
return self._lb_id
self._lb_id = self.properties[self.LOADBALANCER]
if self._lb_id is None:
listener_id = self.properties[self.LISTENER]
listener = self.client().show_listener(listener_id)['listener']
self._lb_id = listener['loadbalancers'][0]['id']
return self._lb_id
def validate(self):
res = super(Pool, self).validate()
if res:
return res
if (self.properties[self.LISTENER] is None and
self.properties[self.LOADBALANCER] is None):
raise exception.PropertyUnspecifiedError(self.LISTENER,
self.LOADBALANCER)
if self.properties[self.SESSION_PERSISTENCE] is not None:
session_p = self.properties[self.SESSION_PERSISTENCE]
persistence_type = session_p[self.SESSION_PERSISTENCE_TYPE]
if persistence_type == self.APP_COOKIE:
if not session_p.get(self.SESSION_PERSISTENCE_COOKIE_NAME):
msg = (_('Property %(cookie)s is required when %(sp)s '
'type is set to %(app)s.') %
{'cookie': self.SESSION_PERSISTENCE_COOKIE_NAME,
'sp': self.SESSION_PERSISTENCE,
'app': self.APP_COOKIE})
raise exception.StackValidationFailed(message=msg)
elif persistence_type == self.SOURCE_IP:
if session_p.get(self.SESSION_PERSISTENCE_COOKIE_NAME):
msg = (_('Property %(cookie)s must NOT be specified when '
'%(sp)s type is set to %(ip)s.') %
{'cookie': self.SESSION_PERSISTENCE_COOKIE_NAME,
'sp': self.SESSION_PERSISTENCE,
'ip': self.SOURCE_IP})
raise exception.StackValidationFailed(message=msg)
def _check_lb_status(self):
return self.client_plugin().check_lb_status(self.lb_id)
def handle_create(self):
properties = self.prepare_properties(
self.properties,
self.physical_resource_name())
if self.LISTENER in properties:
properties['listener_id'] = properties.pop(self.LISTENER)
if self.LOADBALANCER in properties:
properties['loadbalancer_id'] = properties.pop(self.LOADBALANCER)
session_p = properties.get(self.SESSION_PERSISTENCE)
if session_p is not None:
session_props = self.prepare_properties(session_p, None)
properties[self.SESSION_PERSISTENCE] = session_props
return properties
def check_create_complete(self, properties):
if self.resource_id is None:
try:
pool = self.client().create_lbaas_pool(
{'pool': properties})['pool']
self.resource_id_set(pool['id'])
except Exception as ex:
if self.client_plugin().is_invalid(ex):
return False
raise
return self._check_lb_status()
def handle_update(self, json_snippet, tmpl_diff, prop_diff):
self._update_called = False
return prop_diff
def check_update_complete(self, prop_diff):
if not prop_diff:
return True
if not self._update_called:
try:
self.client().update_lbaas_pool(
self.resource_id, {'pool': prop_diff})
self._update_called = True
except Exception as ex:
if self.client_plugin().is_invalid(ex):
return False
raise
return self._check_lb_status()
def handle_delete(self):
self._delete_called = False
def check_delete_complete(self, data):
if self.resource_id is None:
return True
if not self._delete_called:
try:
self.client().delete_lbaas_pool(self.resource_id)
self._delete_called = True
except Exception as ex:
if self.client_plugin().is_invalid(ex):
return False
elif self.client_plugin().is_not_found(ex):
return True
raise
return self._check_lb_status()
def resource_mapping():
return {

200
heat/engine/resources/openstack/neutron/lbaas/pool_member.py
View File

@ -15,15 +15,11 @@
# under the License.
from heat.common.i18n import _
from heat.engine import attributes
from heat.engine import constraints
from heat.engine import properties
from heat.engine.resources.openstack.neutron import neutron
from heat.engine.resources.openstack.heat import none_resource
from heat.engine import support
from heat.engine import translation
class PoolMember(neutron.NeutronResource):
class PoolMember(none_resource.NoneResource):
"""A resource for managing LBaaS v2 Pool Members.
A pool member represents a single backend node.
@ -36,198 +32,6 @@ class PoolMember(neutron.NeutronResource):
previous_status=support.SupportStatus(version='6.0.0')
)
required_service_extension = 'lbaasv2'
entity = 'lbaas_member'
res_info_key = 'member'
PROPERTIES = (
POOL, ADDRESS, PROTOCOL_PORT, WEIGHT, ADMIN_STATE_UP,
SUBNET,
) = (
'pool', 'address', 'protocol_port', 'weight', 'admin_state_up',
'subnet'
)
ATTRIBUTES = (
ADDRESS_ATTR, POOL_ID_ATTR
) = (
'address', 'pool_id'
)
properties_schema = {
POOL: properties.Schema(
properties.Schema.STRING,
_('Name or ID of the load balancing pool.'),
required=True,
constraints=[
constraints.CustomConstraint('neutron.lbaas.pool')
]
),
ADDRESS: properties.Schema(
properties.Schema.STRING,
_('IP address of the pool member on the pool network.'),
required=True,
constraints=[
constraints.CustomConstraint('ip_addr')
]
),
PROTOCOL_PORT: properties.Schema(
properties.Schema.INTEGER,
_('Port on which the pool member listens for requests or '
'connections.'),
required=True,
constraints=[
constraints.Range(1, 65535),
]
),
WEIGHT: properties.Schema(
properties.Schema.INTEGER,
_('Weight of pool member in the pool (default to 1).'),
default=1,
constraints=[
constraints.Range(0, 256),
],
update_allowed=True
),
ADMIN_STATE_UP: properties.Schema(
properties.Schema.BOOLEAN,
_('The administrative state of the pool member.'),
default=True,
update_allowed=True
),
SUBNET: properties.Schema(
properties.Schema.STRING,
_('Subnet name or ID of this member.'),
constraints=[
constraints.CustomConstraint('neutron.subnet')
],
# Make this required untill bug #1585100 is resolved.
required=True
),
}
attributes_schema = {
ADDRESS_ATTR: attributes.Schema(
_('The IP address of the pool member.'),
type=attributes.Schema.STRING
),
POOL_ID_ATTR: attributes.Schema(
_('The ID of the pool to which the pool member belongs.'),
type=attributes.Schema.STRING
)
}
def translation_rules(self, props):
return [
translation.TranslationRule(
props,
translation.TranslationRule.RESOLVE,
[self.SUBNET],
client_plugin=self.client_plugin(),
finder='find_resourceid_by_name_or_id',
entity='subnet'
),
]
def __init__(self, name, definition, stack):
super(PoolMember, self).__init__(name, definition, stack)
self._pool_id = None
self._lb_id = None
@property
def pool_id(self):
if self._pool_id is None:
client_plugin = self.client_plugin()
self._pool_id = client_plugin.find_resourceid_by_name_or_id(
client_plugin.RES_TYPE_LB_POOL,
self.properties[self.POOL])
return self._pool_id
@property
def lb_id(self):
if self._lb_id is None:
pool = self.client().show_lbaas_pool(self.pool_id)['pool']
listener_id = pool['listeners'][0]['id']
listener = self.client().show_listener(listener_id)['listener']
self._lb_id = listener['loadbalancers'][0]['id']
return self._lb_id
def _check_lb_status(self):
return self.client_plugin().check_lb_status(self.lb_id)
def handle_create(self):
properties = self.prepare_properties(
self.properties,
self.physical_resource_name())
self.client_plugin().resolve_pool(
properties, self.POOL, 'pool_id')
properties.pop('pool_id')
properties['subnet_id'] = properties.pop(self.SUBNET)
return properties
def check_create_complete(self, properties):
if self.resource_id is None:
try:
member = self.client().create_lbaas_member(
self.pool_id, {'member': properties})['member']
self.resource_id_set(member['id'])
except Exception as ex:
if self.client_plugin().is_invalid(ex):
return False
raise
return self._check_lb_status()
def _res_get_args(self):
return [self.resource_id, self.pool_id]
def handle_update(self, json_snippet, tmpl_diff, prop_diff):
self._update_called = False
return prop_diff
def check_update_complete(self, prop_diff):
if not prop_diff:
return True
if not self._update_called:
try:
self.client().update_lbaas_member(self.resource_id,
self.pool_id,
{'member': prop_diff})
self._update_called = True
except Exception as ex:
if self.client_plugin().is_invalid(ex):
return False
raise
return self._check_lb_status()
def handle_delete(self):
self._delete_called = False
def check_delete_complete(self, data):
if self.resource_id is None:
return True
if not self._delete_called:
try:
self.client().delete_lbaas_member(self.resource_id,
self.pool_id)
self._delete_called = True
except Exception as ex:
if self.client_plugin().is_invalid(ex):
return False
elif self.client_plugin().is_not_found(ex):
return True
raise
return self._check_lb_status()
def resource_mapping():
return {

121
heat/tests/openstack/neutron/inline_templates.py
View File

@ -71,95 +71,6 @@ resources:
type: OS::Neutron::Net
'''
LB_TEMPLATE = '''
heat_template_version: 2016-04-08
description: Create a loadbalancer
resources:
lb:
type: OS::Neutron::LBaaS::LoadBalancer
properties:
name: my_lb
description: my loadbalancer
vip_address: 10.0.0.4
vip_subnet: sub123
provider: octavia
tenant_id: 1234
admin_state_up: True
'''
LISTENER_TEMPLATE = '''
heat_template_version: 2016-04-08
description: Create a listener
resources:
listener:
type: OS::Neutron::LBaaS::Listener
properties:
protocol_port: 80
protocol: TCP
loadbalancer: 123
default_pool: my_pool
name: my_listener
description: my listener
admin_state_up: True
default_tls_container_ref: ref
sni_container_refs:
- ref1
- ref2
connection_limit: -1
tenant_id: 1234
'''
POOL_TEMPLATE = '''
heat_template_version: 2016-04-08
description: Create a pool
resources:
pool:
type: OS::Neutron::LBaaS::Pool
properties:
name: my_pool
description: my pool
session_persistence:
type: HTTP_COOKIE
lb_algorithm: ROUND_ROBIN
loadbalancer: my_lb
listener: 123
protocol: HTTP
admin_state_up: True
'''
MEMBER_TEMPLATE = '''
heat_template_version: 2016-04-08
description: Create a pool member
resources:
member:
type: OS::Neutron::LBaaS::PoolMember
properties:
pool: 123
address: 1.2.3.4
protocol_port: 80
weight: 1
subnet: sub123
admin_state_up: True
'''
MONITOR_TEMPLATE = '''
heat_template_version: 2016-04-08
description: Create a health monitor
resources:
monitor:
type: OS::Neutron::LBaaS::HealthMonitor
properties:
admin_state_up: True
delay: 3
expected_codes: 200-202
http_method: HEAD
max_retries: 5
pool: 123
timeout: 10
type: HTTP
url_path: /health
'''
SECURITY_GROUP_RULE_TEMPLATE = '''
heat_template_version: 2016-10-14
resources:
@ -173,38 +84,6 @@ resources:
port_range_min: 100
'''
L7POLICY_TEMPLATE = '''
heat_template_version: 2016-04-08
description: Template to test L7Policy Neutron resource
resources:
l7policy:
type: OS::Neutron::LBaaS::L7Policy
properties:
admin_state_up: True
name: test_l7policy
description: test l7policy resource
action: REDIRECT_TO_URL
redirect_url: http://www.mirantis.com
listener: 123
position: 1
'''
L7RULE_TEMPLATE = '''
heat_template_version: 2016-04-08
description: Template to test L7Rule Neutron resource
resources:
l7rule:
type: OS::Neutron::LBaaS::L7Rule
properties:
admin_state_up: True
l7policy: 123
type: HEADER
compare_type: ENDS_WITH
key: test_key
value: test_value
invert: False
'''
SEGMENT_TEMPLATE = '''
heat_template_version: pike
description: Template to test Segment

0
heat/tests/openstack/neutron/lbaas/__init__.py
View File

169
heat/tests/openstack/neutron/lbaas/test_health_monitor.py
View File

@ -1,169 +0,0 @@
#
# Copyright 2015 IBM Corp.
#
# 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.
from unittest import mock
from neutronclient.common import exceptions
from heat.common import template_format
from heat.engine.resources.openstack.neutron.lbaas import health_monitor
from heat.tests import common
from heat.tests.openstack.neutron import inline_templates
from heat.tests import utils
class HealthMonitorTest(common.HeatTestCase):
@mock.patch('heat.engine.clients.os.neutron.'
'NeutronClientPlugin.has_extension', return_value=True)
def _create_stack(self, ext_func, tmpl=inline_templates.MONITOR_TEMPLATE):
self.t = template_format.parse(tmpl)
self.stack = utils.parse_stack(self.t)
self.healthmonitor = self.stack['monitor']
self.neutron_client = mock.MagicMock()
self.healthmonitor.client = mock.MagicMock(
return_value=self.neutron_client)
self.healthmonitor.client_plugin().find_resourceid_by_name_or_id = (
mock.MagicMock(return_value='123'))
self.healthmonitor.client_plugin().client = mock.MagicMock(
return_value=self.neutron_client)
def test_resource_mapping(self):
mapping = health_monitor.resource_mapping()
self.assertEqual(health_monitor.HealthMonitor,
mapping['OS::Neutron::LBaaS::HealthMonitor'])
def test_create(self):
self._create_stack()
self.neutron_client.show_loadbalancer.side_effect = [
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'ACTIVE'}},
]
self.neutron_client.create_lbaas_healthmonitor.side_effect = [
exceptions.StateInvalidClient,
{'healthmonitor': {'id': '1234'}}
]
expected = {
'healthmonitor': {
'admin_state_up': True,
'delay': 3,
'expected_codes': '200-202',
'http_method': 'HEAD',
'max_retries': 5,
'pool_id': '123',
'timeout': 10,
'type': 'HTTP',
'url_path': '/health'
}
}
props = self.healthmonitor.handle_create()
self.assertFalse(self.healthmonitor.check_create_complete(props))
self.neutron_client.create_lbaas_healthmonitor.assert_called_with(
expected)
self.assertFalse(self.healthmonitor.check_create_complete(props))
self.neutron_client.create_lbaas_healthmonitor.assert_called_with(
expected)
self.assertFalse(self.healthmonitor.check_create_complete(props))
self.assertTrue(self.healthmonitor.check_create_complete(props))
def test_show_resource(self):
self._create_stack()
self.healthmonitor.resource_id_set('1234')
self.assertTrue(self.healthmonitor._show_resource())
self.neutron_client.show_lbaas_healthmonitor.assert_called_with(
'1234')
def test_update(self):
self._create_stack()
self.healthmonitor.resource_id_set('1234')
self.neutron_client.show_loadbalancer.side_effect = [
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'ACTIVE'}},
]
self.neutron_client.update_lbaas_healthmonitor.side_effect = [
exceptions.StateInvalidClient, None]
prop_diff = {
'admin_state_up': False,
}
prop_diff = self.healthmonitor.handle_update(None, None, prop_diff)
self.assertFalse(self.healthmonitor.check_update_complete(prop_diff))
self.assertFalse(self.healthmonitor._update_called)
self.neutron_client.update_lbaas_healthmonitor.assert_called_with(
'1234', {'healthmonitor': prop_diff})
self.assertFalse(self.healthmonitor.check_update_complete(prop_diff))
self.assertTrue(self.healthmonitor._update_called)
self.neutron_client.update_lbaas_healthmonitor.assert_called_with(
'1234', {'healthmonitor': prop_diff})
self.assertFalse(self.healthmonitor.check_update_complete(prop_diff))
self.assertTrue(self.healthmonitor.check_update_complete(prop_diff))
def test_delete(self):
self._create_stack()
self.healthmonitor.resource_id_set('1234')
self.neutron_client.show_loadbalancer.side_effect = [
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'ACTIVE'}},
]
self.neutron_client.delete_lbaas_healthmonitor.side_effect = [
exceptions.StateInvalidClient, None]
self.healthmonitor.handle_delete()
self.assertFalse(self.healthmonitor.check_delete_complete(None))
self.assertFalse(self.healthmonitor._delete_called)
self.neutron_client.delete_lbaas_healthmonitor.assert_called_with(
'1234')
self.assertFalse(self.healthmonitor.check_delete_complete(None))
self.assertTrue(self.healthmonitor._delete_called)
self.neutron_client.delete_lbaas_healthmonitor.assert_called_with(
'1234')
self.assertFalse(self.healthmonitor.check_delete_complete(None))
self.assertTrue(self.healthmonitor.check_delete_complete(None))
def test_delete_already_gone(self):
self._create_stack()
self.healthmonitor.resource_id_set('1234')
self.neutron_client.delete_lbaas_healthmonitor.side_effect = (
exceptions.NotFound)
self.healthmonitor.handle_delete()
self.assertTrue(self.healthmonitor.check_delete_complete(None))
self.neutron_client.delete_lbaas_healthmonitor.assert_called_with(
'1234')
def test_delete_failed(self):
self._create_stack()
self.healthmonitor.resource_id_set('1234')
self.neutron_client.delete_lbaas_healthmonitor.side_effect = (
exceptions.Unauthorized)
self.healthmonitor.handle_delete()
self.assertRaises(exceptions.Unauthorized,
self.healthmonitor.check_delete_complete, None)
self.neutron_client.delete_lbaas_healthmonitor.assert_called_with(
'1234')

258
heat/tests/openstack/neutron/lbaas/test_l7policy.py
View File

@ -1,258 +0,0 @@
#
# 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.
from unittest import mock
import yaml
from neutronclient.common import exceptions
from heat.common import exception
from heat.common.i18n import _
from heat.common import template_format
from heat.engine.resources.openstack.neutron.lbaas import l7policy
from heat.tests import common
from heat.tests.openstack.neutron import inline_templates
from heat.tests import utils
class L7PolicyTest(common.HeatTestCase):
def test_resource_mapping(self):
mapping = l7policy.resource_mapping()
self.assertEqual(mapping['OS::Neutron::LBaaS::L7Policy'],
l7policy.L7Policy)
@mock.patch('heat.engine.clients.os.neutron.'
'NeutronClientPlugin.has_extension', return_value=True)
def _create_stack(self, ext_func, tmpl=inline_templates.L7POLICY_TEMPLATE):
self.t = template_format.parse(tmpl)
self.stack = utils.parse_stack(self.t)
self.l7policy = self.stack['l7policy']
self.neutron_client = mock.MagicMock()
self.l7policy.client = mock.MagicMock(return_value=self.neutron_client)
self.l7policy.client_plugin().find_resourceid_by_name_or_id = (
mock.MagicMock(return_value='123'))
self.l7policy.client_plugin().client = mock.MagicMock(
return_value=self.neutron_client)
self.neutron_client.show_loadbalancer.side_effect = [
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'ACTIVE'}},
]
def test_validate_reject_action_with_conflicting_props(self):
tmpl = yaml.safe_load(inline_templates.L7POLICY_TEMPLATE)
props = tmpl['resources']['l7policy']['properties']
props['action'] = 'REJECT'
self._create_stack(tmpl=yaml.dump(tmpl))
msg = _('Properties redirect_pool and redirect_url are not '
'required when action type is set to REJECT.')
with mock.patch('heat.engine.clients.os.neutron.NeutronClientPlugin.'
'has_extension', return_value=True):
self.assertRaisesRegex(exception.StackValidationFailed,
msg, self.l7policy.validate)
def test_validate_redirect_pool_action_with_url(self):
tmpl = yaml.safe_load(inline_templates.L7POLICY_TEMPLATE)
props = tmpl['resources']['l7policy']['properties']
props['action'] = 'REDIRECT_TO_POOL'
props['redirect_pool'] = '123'
self._create_stack(tmpl=yaml.dump(tmpl))
msg = _('redirect_url property should only be specified '
'for action with value REDIRECT_TO_URL.')
with mock.patch('heat.engine.clients.os.neutron.NeutronClientPlugin.'
'has_extension', return_value=True):
self.assertRaisesRegex(exception.ResourcePropertyValueDependency,
msg, self.l7policy.validate)
def test_validate_redirect_pool_action_without_pool(self):
tmpl = yaml.safe_load(inline_templates.L7POLICY_TEMPLATE)
props = tmpl['resources']['l7policy']['properties']
props['action'] = 'REDIRECT_TO_POOL'
del props['redirect_url']
self._create_stack(tmpl=yaml.dump(tmpl))
msg = _('Property redirect_pool is required when action type '
'is set to REDIRECT_TO_POOL.')
with mock.patch('heat.engine.clients.os.neutron.NeutronClientPlugin.'
'has_extension', return_value=True):
self.assertRaisesRegex(exception.StackValidationFailed,
msg, self.l7policy.validate)
def test_validate_redirect_url_action_with_pool(self):
tmpl = yaml.safe_load(inline_templates.L7POLICY_TEMPLATE)
props = tmpl['resources']['l7policy']['properties']
props['redirect_pool'] = '123'
self._create_stack(tmpl=yaml.dump(tmpl))
msg = _('redirect_pool property should only be specified '
'for action with value REDIRECT_TO_POOL.')
with mock.patch('heat.engine.clients.os.neutron.NeutronClientPlugin.'
'has_extension', return_value=True):
self.assertRaisesRegex(exception.ResourcePropertyValueDependency,
msg, self.l7policy.validate)
def test_validate_redirect_url_action_without_url(self):
tmpl = yaml.safe_load(inline_templates.L7POLICY_TEMPLATE)
props = tmpl['resources']['l7policy']['properties']
del props['redirect_url']
self._create_stack(tmpl=yaml.dump(tmpl))
msg = _('Property redirect_url is required when action type '
'is set to REDIRECT_TO_URL.')
with mock.patch('heat.engine.clients.os.neutron.NeutronClientPlugin.'
'has_extension', return_value=True):
self.assertRaisesRegex(exception.StackValidationFailed,
msg, self.l7policy.validate)
def test_create(self):
self._create_stack()
self.neutron_client.create_lbaas_l7policy.side_effect = [
exceptions.StateInvalidClient,
{'l7policy': {'id': '1234'}}
]
expected = {
'l7policy': {
'name': u'test_l7policy',
'description': u'test l7policy resource',
'action': u'REDIRECT_TO_URL',
'listener_id': u'123',
'redirect_url': u'http://www.mirantis.com',
'position': 1,
'admin_state_up': True
}
}
props = self.l7policy.handle_create()
self.assertFalse(self.l7policy.check_create_complete(props))
self.neutron_client.create_lbaas_l7policy.assert_called_with(expected)
self.assertFalse(self.l7policy.check_create_complete(props))
self.neutron_client.create_lbaas_l7policy.assert_called_with(expected)
self.assertFalse(self.l7policy.check_create_complete(props))
self.assertTrue(self.l7policy.check_create_complete(props))
def test_create_missing_properties(self):
self.patchobject(l7policy.L7Policy, 'is_service_available',
return_value=(True, None))
for prop in ('action', 'listener'):
tmpl = yaml.safe_load(inline_templates.L7POLICY_TEMPLATE)
del tmpl['resources']['l7policy']['properties'][prop]
self._create_stack(tmpl=yaml.dump(tmpl))
self.assertRaises(exception.StackValidationFailed,
self.l7policy.validate)
def test_show_resource(self):
self._create_stack()
self.l7policy.resource_id_set('1234')
self.neutron_client.show_lbaas_l7policy.return_value = {
'l7policy': {'id': '1234'}
}
self.assertEqual({'id': '1234'}, self.l7policy._show_resource())
self.neutron_client.show_lbaas_l7policy.assert_called_with('1234')
def test_update(self):
self._create_stack()
self.l7policy.resource_id_set('1234')
self.neutron_client.update_lbaas_l7policy.side_effect = [
exceptions.StateInvalidClient, None]
prop_diff = {
'admin_state_up': False,
'name': 'your_l7policy',
'redirect_url': 'http://www.google.com'
}
prop_diff = self.l7policy.handle_update(None, None, prop_diff)
self.assertFalse(self.l7policy.check_update_complete(prop_diff))
self.assertFalse(self.l7policy._update_called)
self.neutron_client.update_lbaas_l7policy.assert_called_with(
'1234', {'l7policy': prop_diff})
self.assertFalse(self.l7policy.check_update_complete(prop_diff))
self.assertTrue(self.l7policy._update_called)
self.neutron_client.update_lbaas_l7policy.assert_called_with(
'1234', {'l7policy': prop_diff})
self.assertFalse(self.l7policy.check_update_complete(prop_diff))
self.assertTrue(self.l7policy.check_update_complete(prop_diff))
def test_update_redirect_pool_prop_name(self):
self._create_stack()
self.l7policy.resource_id_set('1234')
self.neutron_client.update_lbaas_l7policy.side_effect = [
exceptions.StateInvalidClient, None]
unresolved_diff = {
'redirect_url': None,
'action': 'REDIRECT_TO_POOL',
'redirect_pool': 'UNRESOLVED_POOL'
}
resolved_diff = {
'redirect_url': None,
'action': 'REDIRECT_TO_POOL',
'redirect_pool_id': '123'
}
self.l7policy.handle_update(None, None, unresolved_diff)
self.assertFalse(self.l7policy.check_update_complete(resolved_diff))
self.assertFalse(self.l7policy._update_called)
self.neutron_client.update_lbaas_l7policy.assert_called_with(
'1234', {'l7policy': resolved_diff})
self.assertFalse(self.l7policy.check_update_complete(resolved_diff))
self.assertTrue(self.l7policy._update_called)
self.neutron_client.update_lbaas_l7policy.assert_called_with(
'1234', {'l7policy': resolved_diff})
self.assertFalse(self.l7policy.check_update_complete(resolved_diff))
self.assertTrue(self.l7policy.check_update_complete(resolved_diff))
def test_delete(self):
self._create_stack()
self.l7policy.resource_id_set('1234')
self.neutron_client.delete_lbaas_l7policy.side_effect = [
exceptions.StateInvalidClient, None]
self.l7policy.handle_delete()
self.assertFalse(self.l7policy.check_delete_complete(None))
self.assertFalse(self.l7policy._delete_called)
self.assertFalse(self.l7policy.check_delete_complete(None))
self.assertTrue(self.l7policy._delete_called)
self.neutron_client.delete_lbaas_l7policy.assert_called_with('1234')
self.assertFalse(self.l7policy.check_delete_complete(None))
self.assertTrue(self.l7policy.check_delete_complete(None))
def test_delete_already_gone(self):
self._create_stack()
self.l7policy.resource_id_set('1234')
self.neutron_client.delete_lbaas_l7policy.side_effect = (
exceptions.NotFound)
self.l7policy.handle_delete()
self.assertTrue(self.l7policy.check_delete_complete(None))
def test_delete_failed(self):
self._create_stack()
self.l7policy.resource_id_set('1234')
self.neutron_client.delete_lbaas_l7policy.side_effect = (
exceptions.Unauthorized)
self.l7policy.handle_delete()
self.assertRaises(exceptions.Unauthorized,
self.l7policy.check_delete_complete, None)

179
heat/tests/openstack/neutron/lbaas/test_l7rule.py
View File

@ -1,179 +0,0 @@
#
# 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.
from unittest import mock
import yaml
from neutronclient.common import exceptions
from heat.common import exception
from heat.common.i18n import _
from heat.common import template_format
from heat.engine.resources.openstack.neutron.lbaas import l7rule
from heat.tests import common
from heat.tests.openstack.neutron import inline_templates
from heat.tests import utils
class L7RuleTest(common.HeatTestCase):
def test_resource_mapping(self):
mapping = l7rule.resource_mapping()
self.assertEqual(mapping['OS::Neutron::LBaaS::L7Rule'],
l7rule.L7Rule)
@mock.patch('heat.engine.clients.os.neutron.'
'NeutronClientPlugin.has_extension', return_value=True)
def _create_stack(self, ext_func, tmpl=inline_templates.L7RULE_TEMPLATE):
self.t = template_format.parse(tmpl)
self.stack = utils.parse_stack(self.t)
self.l7rule = self.stack['l7rule']
self.neutron_client = mock.MagicMock()
self.l7rule.client = mock.MagicMock(return_value=self.neutron_client)
self.l7rule.client_plugin().find_resourceid_by_name_or_id = (
mock.MagicMock(return_value='123'))
self.l7rule.client_plugin().client = mock.MagicMock(
return_value=self.neutron_client)
self.neutron_client.show_loadbalancer.side_effect = [
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'ACTIVE'}},
]
def test_validate_when_key_required(self):
tmpl = yaml.safe_load(inline_templates.L7RULE_TEMPLATE)
props = tmpl['resources']['l7rule']['properties']
del props['key']
self._create_stack(tmpl=yaml.dump(tmpl))
msg = _('Property key is missing. This property should be '
'specified for rules of HEADER and COOKIE types.')
with mock.patch('heat.engine.clients.os.neutron.NeutronClientPlugin.'
'has_extension', return_value=True):
self.assertRaisesRegex(exception.StackValidationFailed,
msg, self.l7rule.validate)
def test_create(self):
self._create_stack()
self.neutron_client.create_lbaas_l7rule.side_effect = [
exceptions.StateInvalidClient,
{'rule': {'id': '1234'}}
]
expected = (
'123',
{
'rule': {
'admin_state_up': True,
'invert': False,
'type': u'HEADER',
'compare_type': u'ENDS_WITH',
'key': u'test_key',
'value': u'test_value',
'invert': False
}
}
)
props = self.l7rule.handle_create()
self.assertFalse(self.l7rule.check_create_complete(props))
self.neutron_client.create_lbaas_l7rule.assert_called_with(*expected)
self.assertFalse(self.l7rule.check_create_complete(props))
self.neutron_client.create_lbaas_l7rule.assert_called_with(*expected)
self.assertFalse(self.l7rule.check_create_complete(props))
self.assertTrue(self.l7rule.check_create_complete(props))
def test_create_missing_properties(self):
self.patchobject(l7rule.L7Rule, 'is_service_available',
return_value=(True, None))
for prop in ('l7policy', 'type', 'compare_type', 'value'):
tmpl = yaml.safe_load(inline_templates.L7RULE_TEMPLATE)
del tmpl['resources']['l7rule']['properties'][prop]
self._create_stack(tmpl=yaml.dump(tmpl))
self.assertRaises(exception.StackValidationFailed,
self.l7rule.validate)
def test_show_resource(self):
self._create_stack()
self.l7rule.resource_id_set('1234')
self.neutron_client.show_lbaas_l7rule.return_value = {
'rule': {'id': '1234'}
}
self.assertEqual({'id': '1234'}, self.l7rule._show_resource())
self.neutron_client.show_lbaas_l7rule.assert_called_with('1234', '123')
def test_update(self):
self._create_stack()
self.l7rule.resource_id_set('1234')
self.neutron_client.update_lbaas_l7rule.side_effect = [
exceptions.StateInvalidClient, None]
prop_diff = {
'admin_state_up': False,
'name': 'your_l7policy',
'redirect_url': 'http://www.google.com'
}
prop_diff = self.l7rule.handle_update(None, None, prop_diff)
self.assertFalse(self.l7rule.check_update_complete(prop_diff))
self.assertFalse(self.l7rule._update_called)
self.neutron_client.update_lbaas_l7rule.assert_called_with(
'1234', '123', {'rule': prop_diff})
self.assertFalse(self.l7rule.check_update_complete(prop_diff))
self.assertTrue(self.l7rule._update_called)
self.neutron_client.update_lbaas_l7rule.assert_called_with(
'1234', '123', {'rule': prop_diff})
self.assertFalse(self.l7rule.check_update_complete(prop_diff))
self.assertTrue(self.l7rule.check_update_complete(prop_diff))
def test_delete(self):
self._create_stack()
self.l7rule.resource_id_set('1234')
self.neutron_client.delete_lbaas_l7rule.side_effect = [
exceptions.StateInvalidClient, None]
self.l7rule.handle_delete()
self.assertFalse(self.l7rule.check_delete_complete(None))
self.assertFalse(self.l7rule._delete_called)
self.assertFalse(self.l7rule.check_delete_complete(None))
self.assertTrue(self.l7rule._delete_called)
self.neutron_client.delete_lbaas_l7rule.assert_called_with(
'1234', '123')
self.assertFalse(self.l7rule.check_delete_complete(None))
self.assertTrue(self.l7rule.check_delete_complete(None))
def test_delete_already_gone(self):
self._create_stack()
self.l7rule.resource_id_set('1234')
self.neutron_client.delete_lbaas_l7rule.side_effect = (
exceptions.NotFound)
self.l7rule.handle_delete()
self.assertTrue(self.l7rule.check_delete_complete(None))
def test_delete_failed(self):
self._create_stack()
self.l7rule.resource_id_set('1234')
self.neutron_client.delete_lbaas_l7rule.side_effect = (
exceptions.Unauthorized)
self.l7rule.handle_delete()
self.assertRaises(exceptions.Unauthorized,
self.l7rule.check_delete_complete, None)

195
heat/tests/openstack/neutron/lbaas/test_listener.py
View File

@ -1,195 +0,0 @@
#
# Copyright 2015 IBM Corp.
#
# All Rights Reserved.
# 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.
from unittest import mock
import yaml
from neutronclient.common import exceptions
from heat.common import exception
from heat.common import template_format
from heat.engine.resources.openstack.neutron.lbaas import listener
from heat.tests import common
from heat.tests.openstack.neutron import inline_templates
from heat.tests import utils
class ListenerTest(common.HeatTestCase):
def test_resource_mapping(self):
mapping = listener.resource_mapping()
self.assertEqual(listener.Listener,
mapping['OS::Neutron::LBaaS::Listener'])
@mock.patch('heat.engine.clients.os.neutron.'
'NeutronClientPlugin.has_extension', return_value=True)
def _create_stack(self, ext_func, tmpl=inline_templates.LISTENER_TEMPLATE):
self.t = template_format.parse(tmpl)
self.stack = utils.parse_stack(self.t)
self.listener = self.stack['listener']
self.neutron_client = mock.MagicMock()
self.listener.client = mock.MagicMock(return_value=self.neutron_client)
self.listener.client_plugin().find_resourceid_by_name_or_id = (
mock.MagicMock(return_value='123'))
self.listener.client_plugin().client = mock.MagicMock(
return_value=self.neutron_client)
def test_validate_terminated_https(self):
self.patchobject(listener.Listener, 'is_service_available',
return_value=(True, None))
tmpl = yaml.safe_load(inline_templates.LISTENER_TEMPLATE)
props = tmpl['resources']['listener']['properties']
props['protocol'] = 'TERMINATED_HTTPS'
del props['default_tls_container_ref']
self._create_stack(tmpl=yaml.dump(tmpl))
self.assertRaises(exception.StackValidationFailed,
self.listener.validate)
def test_create(self):
self._create_stack()
self.neutron_client.show_loadbalancer.side_effect = [
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'ACTIVE'}},
]
self.neutron_client.create_listener.side_effect = [
exceptions.StateInvalidClient,
{'listener': {'id': '1234'}}
]
expected = {
'listener': {
'protocol_port': 80,
'protocol': 'TCP',
'loadbalancer_id': '123',
'default_pool_id': 'my_pool',
'name': 'my_listener',
'description': 'my listener',
'admin_state_up': True,
'default_tls_container_ref': 'ref',
'sni_container_refs': ['ref1', 'ref2'],
'connection_limit': -1,
'tenant_id': '1234',
}
}
props = self.listener.handle_create()
self.assertFalse(self.listener.check_create_complete(props))
self.neutron_client.create_listener.assert_called_with(expected)
self.assertFalse(self.listener.check_create_complete(props))
self.neutron_client.create_listener.assert_called_with(expected)
self.assertFalse(self.listener.check_create_complete(props))
self.assertTrue(self.listener.check_create_complete(props))
@mock.patch('heat.engine.clients.os.neutron.'
'NeutronClientPlugin.has_extension', return_value=True)
def test_create_missing_properties(self, ext_func):
for prop in ('protocol', 'protocol_port', 'loadbalancer'):
tmpl = yaml.safe_load(inline_templates.LISTENER_TEMPLATE)
del tmpl['resources']['listener']['properties'][prop]
del tmpl['resources']['listener']['properties']['default_pool']
self._create_stack(tmpl=yaml.dump(tmpl))
if prop == 'loadbalancer':
self.assertRaises(exception.PropertyUnspecifiedError,
self.listener.validate)
else:
self.assertRaises(exception.StackValidationFailed,
self.listener.validate)
def test_show_resource(self):
self._create_stack()
self.listener.resource_id_set('1234')
self.neutron_client.show_listener.return_value = {
'listener': {'id': '1234'}
}
self.assertEqual({'id': '1234'}, self.listener._show_resource())
self.neutron_client.show_listener.assert_called_with('1234')
def test_update(self):
self._create_stack()
self.listener.resource_id_set('1234')
self.neutron_client.show_loadbalancer.side_effect = [
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'ACTIVE'}},
]
self.neutron_client.update_listener.side_effect = [
exceptions.StateInvalidClient, None]
prop_diff = {
'admin_state_up': False,
'name': 'your_listener',
}
prop_diff = self.listener.handle_update(self.listener.t,
None, prop_diff)
self.assertFalse(self.listener.check_update_complete(prop_diff))
self.assertFalse(self.listener._update_called)
self.neutron_client.update_listener.assert_called_with(
'1234', {'listener': prop_diff})
self.assertFalse(self.listener.check_update_complete(prop_diff))
self.assertTrue(self.listener._update_called)
self.neutron_client.update_listener.assert_called_with(
'1234', {'listener': prop_diff})
self.assertFalse(self.listener.check_update_complete(prop_diff))
self.assertTrue(self.listener.check_update_complete(prop_diff))
def test_delete(self):
self._create_stack()
self.listener.resource_id_set('1234')
self.neutron_client.show_loadbalancer.side_effect = [
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'ACTIVE'}},
]
self.neutron_client.delete_listener.side_effect = [
exceptions.StateInvalidClient, None]
self.listener.handle_delete()
self.assertFalse(self.listener.check_delete_complete(None))
self.assertFalse(self.listener._delete_called)
self.neutron_client.delete_listener.assert_called_with('1234')
self.assertFalse(self.listener.check_delete_complete(None))
self.assertTrue(self.listener._delete_called)
self.neutron_client.delete_listener.assert_called_with('1234')
self.assertFalse(self.listener.check_delete_complete(None))
self.assertTrue(self.listener.check_delete_complete(None))
def test_delete_already_gone(self):
self._create_stack()
self.listener.resource_id_set('1234')
self.neutron_client.delete_listener.side_effect = (
exceptions.NotFound)
self.listener.handle_delete()
self.assertTrue(self.listener.check_delete_complete(None))
def test_delete_failed(self):
self._create_stack()
self.listener.resource_id_set('1234')
self.neutron_client.delete_listener.side_effect = (
exceptions.Unauthorized)
self.listener.handle_delete()
self.assertRaises(exceptions.Unauthorized,
self.listener.check_delete_complete, None)

187
heat/tests/openstack/neutron/lbaas/test_loadbalancer.py
View File

@ -1,187 +0,0 @@
#
# Copyright 2015 IBM Corp.
#
# All Rights Reserved.
# 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.
from unittest import mock
from neutronclient.common import exceptions
from heat.common import exception
from heat.common import template_format
from heat.engine.resources.openstack.neutron.lbaas import loadbalancer
from heat.tests import common
from heat.tests.openstack.neutron import inline_templates
from heat.tests import utils
class LoadBalancerTest(common.HeatTestCase):
def test_resource_mapping(self):
mapping = loadbalancer.resource_mapping()
self.assertEqual(loadbalancer.LoadBalancer,
mapping['OS::Neutron::LBaaS::LoadBalancer'])
@mock.patch('heat.engine.clients.os.neutron.'
'NeutronClientPlugin.has_extension', return_value=True)
def _create_stack(self, ext_func, tmpl=inline_templates.LB_TEMPLATE):
self.t = template_format.parse(tmpl)
self.stack = utils.parse_stack(self.t)
self.lb = self.stack['lb']
self.neutron_client = mock.MagicMock()
self.lb.client = mock.MagicMock()
self.lb.client.return_value = self.neutron_client
self.lb.client_plugin().find_resourceid_by_name_or_id = mock.MagicMock(
return_value='123')
self.lb.client_plugin().client = mock.MagicMock(
return_value=self.neutron_client)
self.lb.translate_properties(self.lb.properties)
self.lb.resource_id_set('1234')
def test_create(self):
self._create_stack()
expected = {
'loadbalancer': {
'name': 'my_lb',
'description': 'my loadbalancer',
'vip_address': '10.0.0.4',
'vip_subnet_id': '123',
'provider': 'octavia',
'tenant_id': '1234',
'admin_state_up': True,
}
}
self.lb.handle_create()
self.neutron_client.create_loadbalancer.assert_called_with(expected)
def test_check_create_complete(self):
self._create_stack()
self.neutron_client.show_loadbalancer.side_effect = [
{'loadbalancer': {'provisioning_status': 'ACTIVE'}},
{'loadbalancer': {'provisioning_status': 'PENDING_CREATE'}},
{'loadbalancer': {'provisioning_status': 'ERROR'}},
]
self.assertTrue(self.lb.check_create_complete(None))
self.assertFalse(self.lb.check_create_complete(None))
self.assertRaises(exception.ResourceInError,
self.lb.check_create_complete, None)
def test_show_resource(self):
self._create_stack()
self.neutron_client.show_loadbalancer.return_value = {
'loadbalancer': {'id': '1234'}
}
self.assertEqual({'id': '1234'}, self.lb._show_resource())
self.neutron_client.show_loadbalancer.assert_called_with('1234')
def test_update(self):
self._create_stack()
prop_diff = {
'name': 'lb',
'description': 'a loadbalancer',
'admin_state_up': False,
}
prop_diff = self.lb.handle_update(None, None, prop_diff)
self.neutron_client.update_loadbalancer.assert_called_once_with(
'1234', {'loadbalancer': prop_diff})
def test_update_complete(self):
self._create_stack()
prop_diff = {
'name': 'lb',
'description': 'a loadbalancer',
'admin_state_up': False,
}
self.neutron_client.show_loadbalancer.side_effect = [
{'loadbalancer': {'provisioning_status': 'ACTIVE'}},
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
]
self.lb.handle_update(None, None, prop_diff)
self.assertTrue(self.lb.check_update_complete(prop_diff))
self.assertFalse(self.lb.check_update_complete(prop_diff))
self.assertTrue(self.lb.check_update_complete({}))
def test_delete_active(self):
self._create_stack()
self.neutron_client.show_loadbalancer.side_effect = [
{'loadbalancer': {'provisioning_status': 'ACTIVE'}},
exceptions.NotFound
]
self.lb.handle_delete()
self.assertFalse(self.lb.check_delete_complete(None))
self.assertTrue(self.lb.check_delete_complete(None))
self.neutron_client.delete_loadbalancer.assert_called_with('1234')
self.assertEqual(2, self.neutron_client.show_loadbalancer.call_count)
def test_delete_pending(self):
self._create_stack()
self.neutron_client.show_loadbalancer.side_effect = [
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'ACTIVE'}},
exceptions.NotFound
]
self.lb.handle_delete()
self.assertFalse(self.lb.check_delete_complete(None))
self.assertFalse(self.lb.check_delete_complete(None))
self.assertTrue(self.lb.check_delete_complete(None))
self.neutron_client.delete_loadbalancer.assert_called_with('1234')
self.assertEqual(3, self.neutron_client.show_loadbalancer.call_count)
def test_delete_error(self):
self._create_stack()
self.neutron_client.show_loadbalancer.side_effect = [
{'loadbalancer': {'provisioning_status': 'ERROR'}},
exceptions.NotFound
]
self.lb.handle_delete()
self.assertFalse(self.lb.check_delete_complete(None))
self.assertTrue(self.lb.check_delete_complete(None))
self.neutron_client.delete_loadbalancer.assert_called_with('1234')
self.assertEqual(2, self.neutron_client.show_loadbalancer.call_count)
def test_delete_already_gone(self):
self._create_stack()
self.neutron_client.show_loadbalancer.side_effect = (
exceptions.NotFound)
self.lb.handle_delete()
self.assertTrue(self.lb.check_delete_complete(None))
self.assertEqual(1, self.neutron_client.show_loadbalancer.call_count)
def test_delete_failed(self):
self._create_stack()
self.neutron_client.show_loadbalancer.return_value = {
'loadbalancer': {'provisioning_status': 'ACTIVE'}}
self.neutron_client.delete_loadbalancer.side_effect = (
exceptions.Unauthorized)
self.lb.handle_delete()
self.assertRaises(exceptions.Unauthorized,
self.lb.check_delete_complete, None)

209
heat/tests/openstack/neutron/lbaas/test_pool.py
View File

@ -1,209 +0,0 @@
#
# Copyright 2015 IBM Corp.
#
# All Rights Reserved.
# 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.
from unittest import mock
import yaml
from neutronclient.common import exceptions
from heat.common import exception
from heat.common.i18n import _
from heat.common import template_format
from heat.engine.resources.openstack.neutron.lbaas import pool
from heat.tests import common
from heat.tests.openstack.neutron import inline_templates
from heat.tests import utils
class PoolTest(common.HeatTestCase):
def test_resource_mapping(self):
mapping = pool.resource_mapping()
self.assertEqual(pool.Pool,
mapping['OS::Neutron::LBaaS::Pool'])
@mock.patch('heat.engine.clients.os.neutron.'
'NeutronClientPlugin.has_extension', return_value=True)
def _create_stack(self, ext_func, tmpl=inline_templates.POOL_TEMPLATE):
self.t = template_format.parse(tmpl)
self.stack = utils.parse_stack(self.t)
self.pool = self.stack['pool']
self.neutron_client = mock.MagicMock()
self.pool.client = mock.MagicMock(return_value=self.neutron_client)
self.pool.client_plugin().find_resourceid_by_name_or_id = (
mock.MagicMock(return_value='123'))
self.pool.client_plugin().client = mock.MagicMock(
return_value=self.neutron_client)
def test_validate_no_cookie_name(self):
tmpl = yaml.safe_load(inline_templates.POOL_TEMPLATE)
sp = tmpl['resources']['pool']['properties']['session_persistence']
sp['type'] = 'APP_COOKIE'
self._create_stack(tmpl=yaml.dump(tmpl))
msg = _('Property cookie_name is required when '
'session_persistence type is set to APP_COOKIE.')
with mock.patch('heat.engine.clients.os.neutron.NeutronClientPlugin.'
'has_extension', return_value=True):
self.assertRaisesRegex(exception.StackValidationFailed,
msg, self.pool.validate)
def test_validate_source_ip_cookie_name(self):
tmpl = yaml.safe_load(inline_templates.POOL_TEMPLATE)
sp = tmpl['resources']['pool']['properties']['session_persistence']
sp['type'] = 'SOURCE_IP'
sp['cookie_name'] = 'cookie'
self._create_stack(tmpl=yaml.dump(tmpl))
msg = _('Property cookie_name must NOT be specified when '
'session_persistence type is set to SOURCE_IP.')
with mock.patch('heat.engine.clients.os.neutron.NeutronClientPlugin.'
'has_extension', return_value=True):
self.assertRaisesRegex(exception.StackValidationFailed,
msg, self.pool.validate)
def test_create(self):
self._create_stack()
self.neutron_client.show_loadbalancer.side_effect = [
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'ACTIVE'}},
]
self.neutron_client.create_lbaas_pool.side_effect = [
exceptions.StateInvalidClient,
{'pool': {'id': '1234'}}
]
expected = {
'pool': {
'name': 'my_pool',
'description': 'my pool',
'session_persistence': {
'type': 'HTTP_COOKIE'
},
'lb_algorithm': 'ROUND_ROBIN',
'listener_id': '123',
'loadbalancer_id': 'my_lb',
'protocol': 'HTTP',
'admin_state_up': True
}
}
props = self.pool.handle_create()
self.assertFalse(self.pool.check_create_complete(props))
self.neutron_client.create_lbaas_pool.assert_called_with(expected)
self.assertFalse(self.pool.check_create_complete(props))
self.neutron_client.create_lbaas_pool.assert_called_with(expected)
self.assertFalse(self.pool.check_create_complete(props))
self.assertTrue(self.pool.check_create_complete(props))
def test_create_missing_properties(self):
self.patchobject(pool.Pool, 'is_service_available',
return_value=(True, None))
for prop in ('lb_algorithm', 'listener', 'protocol'):
tmpl = yaml.safe_load(inline_templates.POOL_TEMPLATE)
del tmpl['resources']['pool']['properties']['loadbalancer']
del tmpl['resources']['pool']['properties'][prop]
self._create_stack(tmpl=yaml.dump(tmpl))
if prop == 'listener':
self.assertRaises(exception.PropertyUnspecifiedError,
self.pool.validate)
else:
self.assertRaises(exception.StackValidationFailed,
self.pool.validate)
def test_show_resource(self):
self._create_stack()
self.pool.resource_id_set('1234')
self.neutron_client.show_lbaas_pool.return_value = {
'pool': {'id': '1234'}
}
self.assertEqual(self.pool._show_resource(), {'id': '1234'})
self.neutron_client.show_lbaas_pool.assert_called_with('1234')
def test_update(self):
self._create_stack()
self.pool.resource_id_set('1234')
self.neutron_client.show_loadbalancer.side_effect = [
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'ACTIVE'}},
]
self.neutron_client.update_lbaas_pool.side_effect = [
exceptions.StateInvalidClient, None]
prop_diff = {
'admin_state_up': False,
'name': 'your_pool',
'lb_algorithm': 'SOURCE_IP'
}
prop_diff = self.pool.handle_update(None, None, prop_diff)
self.assertFalse(self.pool.check_update_complete(prop_diff))
self.assertFalse(self.pool._update_called)
self.neutron_client.update_lbaas_pool.assert_called_with(
'1234', {'pool': prop_diff})
self.assertFalse(self.pool.check_update_complete(prop_diff))
self.assertTrue(self.pool._update_called)
self.neutron_client.update_lbaas_pool.assert_called_with(
'1234', {'pool': prop_diff})
self.assertFalse(self.pool.check_update_complete(prop_diff))
self.assertTrue(self.pool.check_update_complete(prop_diff))
def test_delete(self):
self._create_stack()
self.pool.resource_id_set('1234')
self.neutron_client.show_loadbalancer.side_effect = [
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'ACTIVE'}},
]
self.neutron_client.delete_lbaas_pool.side_effect = [
exceptions.StateInvalidClient, None]
self.pool.handle_delete()
self.assertFalse(self.pool.check_delete_complete(None))
self.assertFalse(self.pool._delete_called)
self.assertFalse(self.pool.check_delete_complete(None))
self.assertTrue(self.pool._delete_called)
self.neutron_client.delete_lbaas_pool.assert_called_with('1234')
self.assertFalse(self.pool.check_delete_complete(None))
self.assertTrue(self.pool.check_delete_complete(None))
def test_delete_already_gone(self):
self._create_stack()
self.pool.resource_id_set('1234')
self.neutron_client.delete_lbaas_pool.side_effect = (
exceptions.NotFound)
self.pool.handle_delete()
self.assertTrue(self.pool.check_delete_complete(None))
def test_delete_failed(self):
self._create_stack()
self.pool.resource_id_set('1234')
self.neutron_client.delete_lbaas_pool.side_effect = (
exceptions.Unauthorized)
self.pool.handle_delete()
self.assertRaises(exceptions.Unauthorized,
self.pool.check_delete_complete, None)

165
heat/tests/openstack/neutron/lbaas/test_pool_member.py
View File

@ -1,165 +0,0 @@
#
# Copyright 2015 IBM Corp.
#
# All Rights Reserved.
# 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.
from unittest import mock
from neutronclient.common import exceptions
from heat.common import template_format
from heat.engine.resources.openstack.neutron.lbaas import pool_member
from heat.tests import common
from heat.tests.openstack.neutron import inline_templates
from heat.tests import utils
class PoolMemberTest(common.HeatTestCase):
def test_resource_mapping(self):
mapping = pool_member.resource_mapping()
self.assertEqual(pool_member.PoolMember,
mapping['OS::Neutron::LBaaS::PoolMember'])
@mock.patch('heat.engine.clients.os.neutron.'
'NeutronClientPlugin.has_extension', return_value=True)
def _create_stack(self, ext_func, tmpl=inline_templates.MEMBER_TEMPLATE):
self.t = template_format.parse(tmpl)
self.stack = utils.parse_stack(self.t)
self.member = self.stack['member']
self.neutron_client = mock.MagicMock()
self.member.client = mock.MagicMock(return_value=self.neutron_client)
self.member.client_plugin().find_resourceid_by_name_or_id = (
mock.MagicMock(return_value='123'))
self.member.client_plugin().client = mock.MagicMock(
return_value=self.neutron_client)
self.member.translate_properties(self.member.properties)
def test_create(self):
self._create_stack()
self.neutron_client.show_loadbalancer.side_effect = [
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'ACTIVE'}},
]
self.neutron_client.create_lbaas_member.side_effect = [
exceptions.StateInvalidClient,
{'member': {'id': '1234'}}
]
expected = {
'member': {
'address': '1.2.3.4',
'protocol_port': 80,
'weight': 1,
'subnet_id': '123',
'admin_state_up': True,
}
}
props = self.member.handle_create()
self.assertFalse(self.member.check_create_complete(props))
self.neutron_client.create_lbaas_member.assert_called_with('123',
expected)
self.assertFalse(self.member.check_create_complete(props))
self.neutron_client.create_lbaas_member.assert_called_with('123',
expected)
self.assertFalse(self.member.check_create_complete(props))
self.assertTrue(self.member.check_create_complete(props))
def test_show_resource(self):
self._create_stack()
self.member.resource_id_set('1234')
self.neutron_client.show_lbaas_member.return_value = {
'member': {'id': '1234'}
}
self.assertEqual(self.member._show_resource(), {'id': '1234'})
self.neutron_client.show_lbaas_member.assert_called_with('1234', '123')
def test_update(self):
self._create_stack()
self.member.resource_id_set('1234')
self.neutron_client.show_loadbalancer.side_effect = [
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'ACTIVE'}},
]
self.neutron_client.update_lbaas_member.side_effect = [
exceptions.StateInvalidClient, None]
prop_diff = {
'admin_state_up': False,
'weight': 2,
}
prop_diff = self.member.handle_update(None, None, prop_diff)
self.assertFalse(self.member.check_update_complete(prop_diff))
self.assertFalse(self.member._update_called)
self.neutron_client.update_lbaas_member.assert_called_with(
'1234', '123', {'member': prop_diff})
self.assertFalse(self.member.check_update_complete(prop_diff))
self.assertTrue(self.member._update_called)
self.neutron_client.update_lbaas_member.assert_called_with(
'1234', '123', {'member': prop_diff})
self.assertFalse(self.member.check_update_complete(prop_diff))
self.assertTrue(self.member.check_update_complete(prop_diff))
def test_delete(self):
self._create_stack()
self.member.resource_id_set('1234')
self.neutron_client.show_loadbalancer.side_effect = [
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'PENDING_UPDATE'}},
{'loadbalancer': {'provisioning_status': 'ACTIVE'}},
]
self.neutron_client.delete_lbaas_member.side_effect = [
exceptions.StateInvalidClient, None]
self.member.handle_delete()
self.assertFalse(self.member.check_delete_complete(None))
self.assertFalse(self.member._delete_called)
self.neutron_client.delete_lbaas_member.assert_called_with('1234',
'123')
self.assertFalse(self.member.check_delete_complete(None))
self.assertTrue(self.member._delete_called)
self.neutron_client.delete_lbaas_member.assert_called_with('1234',
'123')
self.assertFalse(self.member.check_delete_complete(None))
self.assertTrue(self.member.check_delete_complete(None))
def test_delete_already_gone(self):
self._create_stack()
self.member.resource_id_set('1234')
self.neutron_client.delete_lbaas_member.side_effect = (
exceptions.NotFound)
self.member.handle_delete()
self.assertTrue(self.member.check_delete_complete(None))
def test_delete_failed(self):
self._create_stack()
self.member.resource_id_set('1234')
self.neutron_client.delete_lbaas_member.side_effect = (
exceptions.Unauthorized)
self.member.handle_delete()
self.assertRaises(exceptions.Unauthorized,
self.member.check_delete_complete, None)