Browse Source

NSX|P: Support Octavia allowed-cidrs

Change-Id: I8bcc082f01a8d1a4b1816f12f1c1be366e7daeaa
changes/06/763206/2
asarfaty 3 months ago
parent
commit
789a9830da
8 changed files with 336 additions and 16 deletions
  1. +14
    -0
      vmware_nsx/plugins/nsx_p/plugin.py
  2. +192
    -0
      vmware_nsx/services/lbaas/nsx_p/implementation/lb_utils.py
  3. +42
    -10
      vmware_nsx/services/lbaas/nsx_p/implementation/listener_mgr.py
  4. +2
    -2
      vmware_nsx/services/lbaas/nsx_p/implementation/loadbalancer_mgr.py
  5. +4
    -0
      vmware_nsx/services/lbaas/nsx_p/implementation/member_mgr.py
  6. +13
    -0
      vmware_nsx/services/lbaas/octavia/octavia_driver.py
  7. +5
    -3
      vmware_nsx/tests/unit/services/lbaas/lb_data_models.py
  8. +64
    -1
      vmware_nsx/tests/unit/services/lbaas/test_nsxp_driver.py

+ 14
- 0
vmware_nsx/plugins/nsx_p/plugin.py View File

@ -3275,6 +3275,20 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
for vs in vs_list:
vs_client.update(vs['id'], ip_address=vip_address)
# Update the vip address group for allowed cidr rules
vip_group_id = lb_utils.VIP_GRP_ID % device_id
try:
self.nsxpolicy.group.get(policy_constants.DEFAULT_DOMAIN,
vip_group_id)
except nsx_lib_exc.ResourceNotFound:
pass
else:
expr = self.nsxpolicy.group.build_ip_address_expression(
[vip_address])
self.nsxpolicy.group.update_with_conditions(
policy_constants.DEFAULT_DOMAIN, vip_group_id,
conditions=[expr])
def create_floatingip(self, context, floatingip):
# First do some validations
fip_data = floatingip['floatingip']


+ 192
- 0
vmware_nsx/services/lbaas/nsx_p/implementation/lb_utils.py View File

@ -14,6 +14,7 @@
# under the License.
import functools
import netaddr
from neutron_lib import exceptions as n_exc
from oslo_log import log as logging
@ -25,6 +26,7 @@ from vmware_nsx.services.lbaas.nsx_p.implementation import lb_const as p_const
from vmware_nsx.services.lbaas.nsx_v3.implementation import lb_utils
from vmware_nsxlib.v3 import exceptions as nsxlib_exc
from vmware_nsxlib.v3 import load_balancer as nsxlib_lb
from vmware_nsxlib.v3 import nsx_constants
from vmware_nsxlib.v3.policy import constants as p_constants
from vmware_nsxlib.v3.policy import utils as p_utils
from vmware_nsxlib.v3 import utils
@ -36,6 +38,9 @@ SERVICE_LB_TAG_SCOPE = 'loadbalancer_id'
# ids in the same tag
SERVICE_LB_TAG_MAX = 20
VIP_GRP_ID = '%s-vip'
MAX_SOURCES_IN_RULE = 128
def get_rule_match_conditions(policy):
match_conditions = []
@ -163,6 +168,10 @@ def get_tags(plugin, resource_id, resource_type, project_id, project_name):
project_id, project_name)
def get_router_from_network(context, plugin, subnet_id):
return lb_utils.get_router_from_network(context, plugin, subnet_id)
def build_persistence_profile_tags(pool_tags, listener):
tags = pool_tags[:]
# With octavia loadbalancer name might not be among data passed
@ -365,3 +374,186 @@ def remove_service_tag_callback(lb_id):
def get_lb_rtr_lock(router_id):
return locking.LockManager.get_lock('lb-router-%s' % str(router_id))
def _get_negated_allowed_cidrs(allowed_cidrs, is_ipv4=True):
allowed_set = netaddr.IPSet(allowed_cidrs)
all_cidr = '0.0.0.0/0' if is_ipv4 else '::/0'
all_set = netaddr.IPSet([all_cidr])
negate_set = all_set - allowed_set
# Translate to cidr, ignoring unsupported cidrs.
negate_cidrs = [str(cidr) for cidr in negate_set.iter_cidrs()
if (not str(cidr).startswith('0.0.0.0/') and
not str(cidr).startswith('::/'))]
# split into max len (128) lists.(%s)
negated_list = [negate_cidrs[i:i + MAX_SOURCES_IN_RULE]
for i in range(0, len(negate_cidrs), MAX_SOURCES_IN_RULE)]
return negated_list
def get_lb_vip_address(core_plugin, context, loadbalancer):
# If loadbalancer vip_port already has floating ip, use floating
# IP as the virtual server VIP address. Else, use the loadbalancer
# vip_address directly on virtual server.
filters = {'port_id': [loadbalancer['vip_port_id']]}
floating_ips = core_plugin.get_floatingips(context,
filters=filters)
if floating_ips:
return floating_ips[0]['floating_ip_address']
return loadbalancer['vip_address']
def get_lb_router_id(core_plugin, context, loadbalancer):
# First try to get it from the vip subnet
router_id = get_router_from_network(
context, core_plugin, loadbalancer['vip_subnet_id'])
if router_id:
return router_id
# Try from the LB service
service = get_lb_nsx_lb_service(
core_plugin.nsxpolicy, loadbalancer['id'], try_old=True)
if service and service.get('connectivity_path'):
return p_utils.path_to_id(service['connectivity_path'])
def set_allowed_cidrs_fw(core_plugin, context, loadbalancer, listeners):
nsxpolicy = core_plugin.nsxpolicy
lb_vip_address = get_lb_vip_address(
core_plugin, context, loadbalancer)
vip_group_id = VIP_GRP_ID % loadbalancer['id']
is_ipv4 = bool(':' not in lb_vip_address)
# Find out if the GW policy exists or not
try:
nsxpolicy.gateway_policy.get(
p_constants.DEFAULT_DOMAIN,
map_id=loadbalancer['id'], silent=True)
except nsxlib_exc.ResourceNotFound:
gw_exist = False
else:
gw_exist = True
# list all the relevant listeners
fw_listeners = []
for listener in listeners:
if listener.get('allowed_cidrs'):
fw_listeners.append({
'id': listener.get('listener_id', listener.get('id')),
'port': listener['protocol_port'],
'allowed_cidrs': listener['allowed_cidrs'],
'negate_cidrs': _get_negated_allowed_cidrs(
listener['allowed_cidrs'],
is_ipv4=is_ipv4)})
if not fw_listeners:
# Delete the GW policy if it exists
if gw_exist:
nsxpolicy.gateway_policy.delete(
p_constants.DEFAULT_DOMAIN,
map_id=loadbalancer['id'])
# Delete related services
tags_to_search = [{'scope': lb_const.LB_LB_TYPE,
'tag': loadbalancer['id']}]
services = nsxpolicy.search_by_tags(
tags_to_search,
nsxpolicy.service.parent_entry_def.resource_type()
)['results']
for srv in services:
if not srv.get('marked_for_delete'):
nsxpolicy.service.delete(srv['id'])
# Delete the vip group
nsxpolicy.group.delete(p_constants.DEFAULT_DOMAIN,
vip_group_id)
return
# Find the router to apply the rules on from the LB service
router_id = get_lb_router_id(core_plugin, context, loadbalancer)
if not router_id:
LOG.info("No router found for LB %s allowed cidrs",
loadbalancer['id'])
return
# Create a group for the vip address (to make it easier to update)
expr = nsxpolicy.group.build_ip_address_expression(
[lb_vip_address])
tags = nsxpolicy.build_v3_tags_payload(
loadbalancer, resource_type=lb_const.LB_LB_TYPE,
project_name=context.tenant_name)
nsxpolicy.group.create_or_overwrite_with_conditions(
"LB_%s_vip" % loadbalancer['id'],
p_constants.DEFAULT_DOMAIN, group_id=vip_group_id,
conditions=[expr], tags=tags)
vip_group_path = nsxpolicy.group.entry_def(
domain_id=p_constants.DEFAULT_DOMAIN,
group_id=vip_group_id).get_resource_full_path()
# Create the list of rules
rules = []
for listener in fw_listeners:
ip_version = netaddr.IPAddress(lb_vip_address).version
# Create the service for this listener
srv_tags = nsxpolicy.build_v3_tags_payload(
loadbalancer, resource_type=lb_const.LB_LB_TYPE,
project_name=context.tenant_name)
srv_tags.append({
'scope': lb_const.LB_LISTENER_TYPE,
'tag': listener['id']})
srv_name = "LB Listener %s" % listener['id']
nsxpolicy.service.create_or_overwrite(
srv_name,
service_id=listener['id'],
description="Service for listener %s" % listener['id'],
protocol=nsx_constants.TCP,
dest_ports=[listener['port']],
tags=srv_tags)
# Build the rules for this listener (128 sources each)
rule_index = 0
for cidr_list in listener['negate_cidrs']:
rule_name = "Allowed cidrs for listener %s" % listener['id']
rule_id = listener['id']
if len(listener['negate_cidrs']) > 1:
rule_name = rule_name + " (part %s of %s)" % (
rule_index, len(listener['negate_cidrs']))
rule_id = rule_id + "-%s" % rule_index
description = "Allow only %s" % listener['allowed_cidrs']
rules.append(nsxpolicy.gateway_policy.build_entry(
rule_name,
p_constants.DEFAULT_DOMAIN, loadbalancer['id'],
rule_id,
description=description,
action=nsx_constants.FW_ACTION_DROP,
ip_protocol=(nsx_constants.IPV4 if ip_version == 4
else nsx_constants.IPV6),
dest_groups=[vip_group_path],
source_groups=cidr_list,
service_ids=[listener['id']],
scope=[nsxpolicy.tier1.get_path(router_id)],
direction=nsx_constants.IN,
plain_groups=True))
rule_index = rule_index + 1
# Create / update the GW policy
if gw_exist:
# only update the rules of this policy
nsxpolicy.gateway_policy.update_entries(
p_constants.DEFAULT_DOMAIN,
loadbalancer['id'], rules,
category=p_constants.CATEGORY_LOCAL_GW)
else:
policy_name = "LB %s allowed cidrs" % loadbalancer['id']
description = ("Allowed CIDRs rules for loadbalancer %s" %
loadbalancer['id'])
tags = nsxpolicy.build_v3_tags_payload(
loadbalancer, resource_type=lb_const.LB_LB_TYPE,
project_name=context.tenant_name)
nsxpolicy.gateway_policy.create_with_entries(
policy_name, p_constants.DEFAULT_DOMAIN,
map_id=loadbalancer['id'],
description=description,
tags=tags,
entries=rules,
category=p_constants.CATEGORY_LOCAL_GW)

+ 42
- 10
vmware_nsx/services/lbaas/nsx_p/implementation/listener_mgr.py View File

@ -20,6 +20,7 @@ from oslo_log import log as logging
from oslo_utils import excutils
from vmware_nsx._i18n import _
from vmware_nsx.common import utils as com_utils
from vmware_nsx.services.lbaas import base_mgr
from vmware_nsx.services.lbaas import lb_common
from vmware_nsx.services.lbaas import lb_const
@ -81,16 +82,8 @@ class EdgeListenerManagerFromDict(base_mgr.NsxpLoadbalancerBaseManager):
def _get_virtual_server_kwargs(self, context, listener, vs_name, tags,
lb_service_id, certificate=None):
# If loadbalancer vip_port already has floating ip, use floating
# IP as the virtual server VIP address. Else, use the loadbalancer
# vip_address directly on virtual server.
filters = {'port_id': [listener['loadbalancer']['vip_port_id']]}
floating_ips = self.core_plugin.get_floatingips(context,
filters=filters)
if floating_ips:
lb_vip_address = floating_ips[0]['floating_ip_address']
else:
lb_vip_address = listener['loadbalancer']['vip_address']
lb_vip_address = lb_utils.get_lb_vip_address(
self.core_plugin, context, listener['loadbalancer'])
kwargs = {'virtual_server_id': listener['id'],
'ip_address': lb_vip_address,
@ -165,6 +158,7 @@ class EdgeListenerManagerFromDict(base_mgr.NsxpLoadbalancerBaseManager):
listener.get('default_pool'), listener, completor)
def create(self, context, listener, completor, certificate=None):
self._validate_allowed_cidrs(listener, completor)
nsxlib_lb = self.core_plugin.nsxpolicy.load_balancer
vs_client = nsxlib_lb.virtual_server
vs_name = utils.get_name_and_uuid(listener['name'] or 'listener',
@ -196,8 +190,23 @@ class EdgeListenerManagerFromDict(base_mgr.NsxpLoadbalancerBaseManager):
self._update_default_pool(context, listener, completor)
# Update the allowed cidrs fw
listeners = copy.copy(listener['loadbalancer']['listeners'])
listeners.append(listener)
lb_utils.set_allowed_cidrs_fw(self.core_plugin,
context, listener['loadbalancer'],
listeners)
completor(success=True)
def _validate_allowed_cidrs(self, listener, completor):
if (listener.get('allowed_cidrs') and
not com_utils.is_nsx_version_3_0_0(self.core_plugin._nsx_version)):
completor(success=False)
msg = (_('Allowed cidrs are not supported with NSX %s') %
self.core_plugin._nsx_version)
raise n_exc.BadRequest(resource='lbaas-listener', msg=msg)
def _get_pool_tags(self, context, pool, listener_tenant_id):
return lb_utils.get_tags(self.core_plugin, pool['id'],
lb_const.LB_POOL_TYPE,
@ -251,6 +260,7 @@ class EdgeListenerManagerFromDict(base_mgr.NsxpLoadbalancerBaseManager):
def update(self, context, old_listener, new_listener, completor,
certificate=None):
self._validate_allowed_cidrs(new_listener, completor)
nsxlib_lb = self.core_plugin.nsxpolicy.load_balancer
vs_client = nsxlib_lb.virtual_server
app_client = self._get_nsxlib_app_profile(nsxlib_lb, old_listener)
@ -294,6 +304,18 @@ class EdgeListenerManagerFromDict(base_mgr.NsxpLoadbalancerBaseManager):
new_listener.get('default_pool_id')):
self._update_default_pool(context, new_listener,
completor, old_listener)
# Update the allowed cidrs fw (with an updated list of listeners)
listeners = []
for elem in new_listener['loadbalancer']['listeners']:
if elem['listener_id'] == new_listener['id']:
listeners.append(new_listener)
else:
listeners.append(elem)
lb_utils.set_allowed_cidrs_fw(self.core_plugin,
context, new_listener['loadbalancer'],
listeners)
completor(success=True)
def delete(self, context, listener, completor):
@ -348,6 +370,16 @@ class EdgeListenerManagerFromDict(base_mgr.NsxpLoadbalancerBaseManager):
{'crt': res_obj['id'], 'list': listener['id']})
LOG.error(msg)
# Update the allowed cidrs fw
listeners = copy.copy(listener['loadbalancer']['listeners'])
for elem in listeners:
if elem['listener_id'] == listener['id']:
listeners.remove(elem)
break
lb_utils.set_allowed_cidrs_fw(self.core_plugin,
context, listener['loadbalancer'],
listeners)
completor(success=True)
def delete_cascade(self, context, listener, completor):


+ 2
- 2
vmware_nsx/services/lbaas/nsx_p/implementation/loadbalancer_mgr.py View File

@ -32,7 +32,7 @@ LOG = logging.getLogger(__name__)
class EdgeLoadBalancerManagerFromDict(base_mgr.NsxpLoadbalancerBaseManager):
def _get_lb_router(self, context, lb):
router_id = lb_utils.get_router_from_network(
router_id = p_utils.get_router_from_network(
context, self.core_plugin, lb['vip_subnet_id'])
return router_id
@ -143,7 +143,7 @@ class EdgeLoadBalancerManagerFromDict(base_mgr.NsxpLoadbalancerBaseManager):
def delete(self, context, lb, completor):
router_id = None
try:
router_id = lb_utils.get_router_from_network(
router_id = p_utils.get_router_from_network(
context, self.core_plugin, lb['vip_subnet_id'])
except n_exc.SubnetNotFound:
LOG.warning("VIP subnet %s not found while deleting "


+ 4
- 0
vmware_nsx/services/lbaas/nsx_p/implementation/member_mgr.py View File

@ -135,6 +135,10 @@ class EdgeMemberManagerFromDict(base_mgr.NsxpLoadbalancerBaseManager):
connectivity_path=connectivity_path)
p_utils.update_router_lb_vip_advertisement(
context, self.core_plugin, router_id)
# Update the LB gateway policy now that we have a router
p_utils.set_allowed_cidrs_fw(self.core_plugin,
context, lb, lb['listeners'])
except Exception as e:
with excutils.save_and_reraise_exception():
completor(success=False)


+ 13
- 0
vmware_nsx/services/lbaas/octavia/octavia_driver.py View File

@ -139,6 +139,19 @@ class NSXOctaviaDriver(driver_base.ProviderDriver):
lb_dict['vip_port_id'] = db_lb.vip.port_id
lb_dict['vip_network_id'] = db_lb.vip.network_id
lb_dict['vip_subnet_id'] = db_lb.vip.subnet_id
# Add the listeners to the dictionary
listeners = []
for listener in db_lb.listeners:
db_listener = self.repositories.listener.get(
db_apis.get_session(), id=listener.id)
listener_obj = oct_utils.db_listener_to_provider_listener(
db_listener)
listener_dict = listener_obj.to_dict(
recurse=False, render_unsets=True)
# Add allowed cidrs too
listener_dict['allowed_cidrs'] = listener_obj.allowed_cidrs
listeners.append(listener_dict)
lb_dict['listeners'] = listeners
return lb_dict
def _get_listener_in_pool_dict(self, pool_dict, is_update):


+ 5
- 3
vmware_nsx/tests/unit/services/lbaas/lb_data_models.py View File

@ -57,7 +57,7 @@ class BaseDataModel(object):
if isinstance(item, BaseDataModel):
ret[attr].append(item.to_dict())
else:
ret[attr] = item
ret[attr].append(item)
elif isinstance(getattr(self, attr), BaseDataModel):
ret[attr] = value.to_dict()
elif six.PY2 and isinstance(value, six.text_type):
@ -654,7 +654,7 @@ class Listener(BaseDataModel):
'loadbalancer_id', 'protocol', 'default_tls_container_id',
'sni_containers', 'protocol_port', 'connection_limit',
'admin_state_up', 'provisioning_status', 'operating_status',
'default_pool', 'loadbalancer', 'l7_policies']
'default_pool', 'loadbalancer', 'l7_policies', 'allowed_cidrs']
def __init__(self, id=None, tenant_id=None, name=None, description=None,
default_pool_id=None, loadbalancer_id=None, protocol=None,
@ -662,7 +662,7 @@ class Listener(BaseDataModel):
protocol_port=None, connection_limit=None,
admin_state_up=None, provisioning_status=None,
operating_status=None, default_pool=None, loadbalancer=None,
l7_policies=None):
l7_policies=None, allowed_cidrs=None):
self.id = id
self.tenant_id = tenant_id
self.name = name
@ -680,6 +680,7 @@ class Listener(BaseDataModel):
self.default_pool = default_pool
self.loadbalancer = loadbalancer
self.l7_policies = l7_policies or []
self.allowed_cidrs = allowed_cidrs or []
def attached_to_loadbalancer(self):
return bool(self.loadbalancer)
@ -700,6 +701,7 @@ class Listener(BaseDataModel):
del ret_dict['l7_policies']
ret_dict['l7policies'] = [{'id': l7_policy.id}
for l7_policy in self.l7_policies]
ret_dict['allowed_cidrs'] = self.allowed_cidrs
return ret_dict
@classmethod


+ 64
- 1
vmware_nsx/tests/unit/services/lbaas/test_nsxp_driver.py View File

@ -34,6 +34,8 @@ from vmware_nsx.services.lbaas.nsx_v3.implementation import lb_utils
from vmware_nsx.services.lbaas.octavia import octavia_listener
from vmware_nsx.tests.unit.services.lbaas import lb_data_models as lb_models
from vmware_nsx.tests.unit.services.lbaas import lb_translators
from vmware_nsxlib.v3 import exceptions as nsxlib_exc
from vmware_nsxlib.v3.policy import constants as policy_constants
# TODO(asarfaty): Use octavia models for those tests
@ -162,6 +164,7 @@ class BaseTestEdgeLbaasV2(base.BaseTestCase):
self.lbv2_driver = mock.Mock()
self.core_plugin = mock.Mock()
self.core_plugin._nsx_version = '2.5.0'
base_mgr.LoadbalancerBaseManager._lbv2_driver = self.lbv2_driver
base_mgr.LoadbalancerBaseManager._core_plugin = self.core_plugin
self._patch_lb_plugin(self.lbv2_driver, self._tested_entity)
@ -179,6 +182,10 @@ class BaseTestEdgeLbaasV2(base.BaseTestCase):
self.terminated_https_listener = lb_models.Listener(
HTTPS_LISTENER_ID, LB_TENANT_ID, 'listener3', '', None, LB_ID,
'TERMINATED_HTTPS', protocol_port=443, loadbalancer=self.lb)
self.allowed_cidr_listener = lb_models.Listener(
LISTENER_ID, LB_TENANT_ID, 'listener4', '', None, LB_ID,
'HTTP', protocol_port=80, allowed_cidrs=['1.1.1.0/24'],
loadbalancer=self.lb)
self.pool = lb_models.Pool(POOL_ID, LB_TENANT_ID, 'pool1', '',
None, 'HTTP', 'ROUND_ROBIN',
loadbalancer_id=LB_ID,
@ -226,6 +233,8 @@ class BaseTestEdgeLbaasV2(base.BaseTestCase):
self.lb)
self.listener_dict = lb_translators.lb_listener_obj_to_dict(
self.listener)
self.cidr_list_dict = lb_translators.lb_listener_obj_to_dict(
self.allowed_cidr_listener)
self.https_listener_dict = lb_translators.lb_listener_obj_to_dict(
self.https_listener)
self.terminated_https_listener_dict = lb_translators.\
@ -712,7 +721,7 @@ class TestEdgeLbaasV2Listener(BaseTestEdgeLbaasV2):
def _tested_entity(self):
return 'listener'
def _create_listener(self, protocol='HTTP'):
def _create_listener(self, protocol='HTTP', allowed_cidr=False):
with mock.patch.object(self.core_plugin, 'get_floatingips'
) as mock_get_floatingips, \
mock.patch.object(self.core_plugin,
@ -721,6 +730,11 @@ class TestEdgeLbaasV2Listener(BaseTestEdgeLbaasV2):
mock.patch.object(self.core_plugin.nsxpolicy, 'search_by_tags',
return_value={'results': [
{'id': LB_SERVICE_ID}]}),\
mock.patch.object(self.core_plugin.nsxpolicy.gateway_policy,
'get',
side_effect=nsxlib_exc.ResourceNotFound), \
mock.patch.object(self.core_plugin.nsxpolicy.gateway_policy,
'create_with_entries') as create_gw_pol, \
mock.patch.object(self.vs_client, 'create_or_overwrite'
) as mock_add_virtual_server:
mock_get_floatingips.return_value = []
@ -729,6 +743,8 @@ class TestEdgeLbaasV2Listener(BaseTestEdgeLbaasV2):
if protocol == 'HTTPS':
listener = self.https_listener_dict
listener_id = HTTP_LISTENER_ID
if allowed_cidr:
listener = self.cidr_list_dict
self.edge_driver.listener.create(self.context, listener,
self.completor)
@ -748,9 +764,29 @@ class TestEdgeLbaasV2Listener(BaseTestEdgeLbaasV2):
self.assertTrue(self.last_completor_called)
self.assertTrue(self.last_completor_succees)
if not allowed_cidr:
create_gw_pol.assert_not_called()
else:
create_gw_pol.assert_called_once_with(
'LB %s allowed cidrs' % LB_ID,
policy_constants.DEFAULT_DOMAIN,
map_id=LB_ID,
category=policy_constants.CATEGORY_LOCAL_GW,
description=mock.ANY,
entries=[mock.ANY],
tags=mock.ANY)
def test_create_http_listener(self):
self._create_listener()
def test_create_allowed_cidr_listener(self):
orig_nsx_ver = self.core_plugin._nsx_version
self.core_plugin._nsx_version = '3.1.0'
with mock.patch.object(lb_utils, 'get_router_from_network',
return_value=ROUTER_ID):
self._create_listener(allowed_cidr=True)
self.core_plugin._nsx_version = orig_nsx_ver
def test_create_https_listener(self):
self._create_listener(protocol='HTTPS')
@ -1079,6 +1115,8 @@ class TestEdgeLbaasV2Listener(BaseTestEdgeLbaasV2):
def test_delete(self):
with mock.patch.object(self.service_client, 'get'
) as mock_get_lb_service, \
mock.patch.object(self.core_plugin, 'get_floatingips',
return_value=[]), \
mock.patch.object(self.app_client, 'delete'
) as mock_delete_app_profile, \
mock.patch.object(self.vs_client, 'delete'
@ -1096,6 +1134,30 @@ class TestEdgeLbaasV2Listener(BaseTestEdgeLbaasV2):
self.assertTrue(self.last_completor_called)
self.assertTrue(self.last_completor_succees)
def test_delete_cascade(self):
self.reset_completor()
with mock.patch.object(self.service_client, 'get'
) as mock_get_lb_service, \
mock.patch.object(self.core_plugin, 'get_floatingips',
return_value=[]), \
mock.patch.object(self.app_client, 'delete'
) as mock_delete_app_profile, \
mock.patch.object(self.vs_client, 'delete'
) as mock_delete_virtual_server:
mock_get_lb_service.return_value = {
'id': LB_SERVICE_ID,
'virtual_server_ids': [LB_VS_ID]}
self.edge_driver.listener.delete_cascade(
self.context, self.listener_dict,
self.completor)
mock_delete_virtual_server.assert_called_with(LB_VS_ID)
mock_delete_app_profile.assert_called_with(LISTENER_ID)
self.assertTrue(self.last_completor_called)
self.assertTrue(self.last_completor_succees)
class TestEdgeLbaasV2Pool(BaseTestEdgeLbaasV2):
def setUp(self):
@ -1578,6 +1640,7 @@ class TestEdgeLbaasV2Member(BaseTestEdgeLbaasV2):
mock.patch.object(self.core_plugin, 'get_floatingips',
return_value=[{
'fixed_ip_address': MEMBER_ADDRESS,
'floating_ip_address': '1.1.1.1',
'router_id': LB_ROUTER_ID}]),\
mock.patch.object(self.pool_client,
'create_pool_member_and_add_to_pool'


Loading…
Cancel
Save