x/vmware-nsx
Code Issues Proposed changes

NSX|P: Improve LB unit tests

Browse Source 
Change-Id: Icf03ea3b2d4a24ffe0c1c21e41660743784442fe
This commit is contained in:
Adit Sarfaty 2019-06-24 12:13:56 +03:00 committed by Kobi Samoray
parent 33c7ae15a6
commit 21dee81e7a
2 changed files with 78 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
vmware_nsx/tests/unit/nsx_p/test_plugin.py
View File

@ -36,6 +36,7 @@ from neutron_lib.api.definitions import portbindings
from neutron_lib.api.definitions import provider_net as pnet
from neutron_lib.api.definitions import vlantransparent as vlan_apidef
from neutron_lib.callbacks import events
from neutron_lib.callbacks import exceptions as nc_exc
from neutron_lib.callbacks import registry
from neutron_lib.callbacks import resources
from neutron_lib import constants
@ -48,6 +49,8 @@ from vmware_nsx.common import utils
from vmware_nsx.extensions import providersecuritygroup as provider_sg
from vmware_nsx.plugins.common import plugin as com_plugin
from vmware_nsx.plugins.nsx_p import plugin as nsx_plugin
from vmware_nsx.services.lbaas.nsx_p.implementation import loadbalancer_mgr
from vmware_nsx.services.lbaas.octavia import octavia_listener
from vmware_nsx.tests import unit as vmware
from vmware_nsx.tests.unit.common_plugin import common_v3
@ -2171,3 +2174,42 @@ class NsxPTestL3NatTestCase(NsxPTestL3NatTest,
self._update('floatingips', fip['floatingip'][
'id'], {'floatingip': {'port_id': port_id}},
expected_code=exc.HTTPBadRequest.code)
def test_router_delete_with_lb_service(self):
self.lb_mock1.stop()
self.lb_mock2.stop()
# Create the LB object - here the delete callback is registered
loadbalancer = loadbalancer_mgr.EdgeLoadBalancerManagerFromDict()
oct_listener = octavia_listener.NSXOctaviaListenerEndpoint(
loadbalancer=loadbalancer)
with self.router() as router:
with mock.patch.object(
self.plugin.nsxpolicy, 'search_by_tags',
return_value={'results': [{'id': 'dummy'}]}):
self.assertRaises(nc_exc.CallbackFailure,
self.plugin_instance.delete_router,
context.get_admin_context(),
router['router']['id'])
# Unregister callback
oct_listener._unsubscribe_router_delete_callback()
self.lb_mock1.start()
self.lb_mock2.start()
def test_router_delete_with_no_lb_service(self):
self.lb_mock1.stop()
self.lb_mock2.stop()
# Create the LB object - here the delete callback is registered
loadbalancer = loadbalancer_mgr.EdgeLoadBalancerManagerFromDict()
oct_listener = octavia_listener.NSXOctaviaListenerEndpoint(
loadbalancer=loadbalancer)
with self.router() as router:
with mock.patch.object(
self.plugin.nsxpolicy, 'search_by_tags',
return_value={'results': []}):
self.plugin_instance.delete_router(
context.get_admin_context(),
router['router']['id'])
# Unregister callback
oct_listener._unsubscribe_router_delete_callback()
self.lb_mock1.start()
self.lb_mock2.start()

38
vmware_nsx/tests/unit/services/lbaas/test_nsxp_driver.py
View File

@ -302,7 +302,7 @@ class TestEdgeLbaasV2Loadbalancer(BaseTestEdgeLbaasV2):
return_value=[]),\
mock.patch.object(self.core_plugin,
'service_router_has_loadbalancers',
return_value=False),\
return_value=False) as plugin_has_lb,\
mock.patch.object(self.service_client, 'get_router_lb_service',
return_value=None),\
mock.patch.object(self.service_client, 'create_or_overwrite'
@ -318,6 +318,31 @@ class TestEdgeLbaasV2Loadbalancer(BaseTestEdgeLbaasV2):
description=self.lb_dict['description'],
tags=mock.ANY, size='SMALL',
connectivity_path=mock.ANY)
plugin_has_lb.assert_called_once_with(mock.ANY, ROUTER_ID)
def test_create_same_router_fail(self):
neutron_router = {'id': ROUTER_ID, 'name': 'dummy',
'external_gateway_info': {'external_fixed_ips': []}}
with mock.patch.object(lb_utils, 'get_network_from_subnet',
return_value=LB_NETWORK), \
mock.patch.object(lb_utils, 'get_router_from_network',
return_value=ROUTER_ID),\
mock.patch.object(self.core_plugin, 'get_router',
return_value=neutron_router), \
mock.patch.object(self.core_plugin, '_find_router_gw_subnets',
return_value=[]),\
mock.patch.object(self.core_plugin,
'service_router_has_loadbalancers',
return_value=True) as plugin_has_lb,\
mock.patch.object(self.service_client, 'get_router_lb_service',
return_value=None):
self.assertRaises(
n_exc.BadRequest,
self.edge_driver.loadbalancer.create,
self.context, self.lb_dict, self.completor)
self.assertTrue(self.last_completor_called)
self.assertFalse(self.last_completor_succees)
plugin_has_lb.assert_called_once_with(mock.ANY, ROUTER_ID)
def test_create_external_vip(self):
with mock.patch.object(lb_utils, 'get_router_from_network',
@ -1176,15 +1201,23 @@ class TestEdgeLbaasV2Member(BaseTestEdgeLbaasV2):
) as mock_get_router, \
mock.patch.object(self.service_client, 'get_router_lb_service'
) as mock_get_lb_service, \
mock.patch.object(self.service_client, 'get',
return_value={}), \
mock.patch.object(self.core_plugin,
'service_router_has_loadbalancers',
return_value=False) as plugin_has_lb,\
mock.patch.object(self.pool_client, 'get'
) as mock_get_pool, \
mock.patch.object(self.core_plugin, '_find_router_gw_subnets',
return_value=[]),\
mock.patch.object(self.core_plugin, 'get_floatingips',
return_value=[{
'fixed_ip_address': MEMBER_ADDRESS}]),\
mock.patch.object(self.pool_client,
'create_pool_member_and_add_to_pool'
) as mock_update_pool_with_members:
mock_get_pool_members.return_value = [self.member]
mock_get_network.return_value = LB_NETWORK
mock_get_network.return_value = EXT_LB_NETWORK
mock_get_router.return_value = LB_ROUTER_ID
mock_get_lb_service.return_value = {'id': LB_SERVICE_ID}
mock_get_pool.return_value = LB_POOL
@ -1200,6 +1233,7 @@ class TestEdgeLbaasV2Member(BaseTestEdgeLbaasV2):
admin_state='ENABLED')
self.assertTrue(self.last_completor_called)
self.assertTrue(self.last_completor_succees)
plugin_has_lb.assert_called_once_with(mock.ANY, LB_ROUTER_ID)
def test_update(self):
new_member = lb_models.Member(MEMBER_ID, LB_TENANT_ID, POOL_ID,