x/vmware-nsx
Code Issues Proposed changes

NSXv: Use router Edge for LBaaSv2 unit tests

Browse Source 
Add unit tests for change in patch https://review.openstack.org/#/c/560895/

Change-Id: I3fb96747036b870046ca0777745f768c35009cbe
This commit is contained in:
Kobi Samoray
2018-04-23 14:22:45 +03:00
parent 71f92348c1
commit f8b189ee9d
1 changed files with 110 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

184
vmware_nsx/tests/unit/nsx_v/test_edge_loadbalancer_driver_v2.py
View File

@@ -20,6 +20,7 @@ from neutron.tests import base
from neutron_lbaas.services.loadbalancer import data_models as lb_models
from neutron_lib import context
from neutron_lib import exceptions as n_exc
from oslo_config import cfg
from vmware_nsx.db import nsxv_db
from vmware_nsx.plugins.nsx_v.vshield import vcns_driver
@@ -28,6 +29,7 @@ from vmware_nsx.services.lbaas.nsx_v import lbaas_common as lb_common
LB_VIP = '10.0.0.10'
LB_SUBNET = 'some-subnet'
LB_EDGE_ID = 'edge-x'
LB_ID = 'xxx-xxx'
LB_TENANT_ID = 'yyy-yyy'
@@ -111,7 +113,7 @@ class BaseTestEdgeLbaasV2(base.BaseTestCase):
self._patch_lb_plugin(self.lbv2_driver, self._tested_entity)
self.lb = lb_models.LoadBalancer(LB_ID, LB_TENANT_ID, 'lb-name', '',
'some-subnet', 'port-id', LB_VIP)
LB_SUBNET, 'port-id', LB_VIP)
self.listener = lb_models.Listener(LISTENER_ID, LB_TENANT_ID,
'l-name', '', None, LB_ID,
'HTTP', protocol_port=80,
@@ -168,17 +170,29 @@ class BaseTestEdgeLbaasV2(base.BaseTestCase):
setattr(lb_plugin, manager, self.real_manager)
class TestEdgeLbaasV2Loadbalancer(BaseTestEdgeLbaasV2):
class TestEdgeLbaasV2LoadbalancerOnRtr(BaseTestEdgeLbaasV2):
def setUp(self):
super(TestEdgeLbaasV2Loadbalancer, self).setUp()
super(TestEdgeLbaasV2LoadbalancerOnRtr, self).setUp()
cfg.CONF.set_override('use_routers_as_lbaas_platform',
self._deploy_on_router, group="nsxv")
@property
def _tested_entity(self):
return 'load_balancer'
@property
def _edge_getter(self):
return 'get_lbaas_edge_id_for_subnet'
@property
def _deploy_on_router(self):
return True
def test_create(self):
with mock.patch.object(lb_common, 'get_lbaas_edge_id'
with mock.patch.object(lb_common, self._edge_getter
) as mock_get_edge, \
mock.patch.object(lb_common, 'add_vip_as_secondary_ip'
) as mock_vip_sec_ip, \
mock.patch.object(lb_common, 'add_vip_fw_rule'
) as mock_add_vip_fwr, \
mock.patch.object(lb_common, 'set_lb_firewall_default_rule'
@@ -195,48 +209,18 @@ class TestEdgeLbaasV2Loadbalancer(BaseTestEdgeLbaasV2):
mock_get_lb_binding_by_edge.return_value = []
self.edge_driver.loadbalancer.create(self.context, self.lb)
mock_add_vip_fwr.assert_called_with(self.edge_driver.vcns,
LB_EDGE_ID,
LB_ID,
LB_VIP)
mock_db_binding.assert_called_with(self.context.session,
LB_ID,
LB_EDGE_ID,
LB_VIP_FWR_ID,
LB_VIP)
mock_set_fw_rule.assert_called_with(
self.edge_driver.vcns, LB_EDGE_ID, 'accept')
mock_get_edge.assert_called_with(mock.ANY, mock.ANY, LB_ID, LB_VIP,
mock.ANY, LB_TENANT_ID, 'compact')
mock_successful_completion = (
self.lbv2_driver.load_balancer.successful_completion)
mock_successful_completion.assert_called_with(self.context,
self.lb)
mock_enable_edge_acceleration.assert_called_with(
self.edge_driver.vcns, LB_EDGE_ID)
def test_create_with_flavor(self):
flavor_name = 'large'
with mock.patch.object(lb_common, 'get_lbaas_edge_id'
) as mock_get_edge, \
mock.patch.object(lb_common, 'add_vip_fw_rule'
) as mock_add_vip_fwr, \
mock.patch.object(lb_common, 'set_lb_firewall_default_rule'
) as mock_set_fw_rule, \
mock.patch.object(lb_common, 'enable_edge_acceleration'
) as mock_enable_edge_acceleration, \
mock.patch.object(nsxv_db,
'get_nsxv_lbaas_loadbalancer_binding_by_edge'
) as mock_get_lb_binding_by_edge, \
mock.patch.object(nsxv_db, 'add_nsxv_lbaas_loadbalancer_binding'
) as mock_db_binding,\
mock.patch('neutron.services.flavors.flavors_plugin.FlavorsPlugin.'
'get_flavor', return_value={'name': flavor_name}):
mock_get_edge.return_value = LB_EDGE_ID
mock_add_vip_fwr.return_value = LB_VIP_FWR_ID
mock_get_lb_binding_by_edge.return_value = []
self.lb.flavor_id = 'dummy'
self.edge_driver.loadbalancer.create(self.context, self.lb)
if self._deploy_on_router:
mock_vip_sec_ip.assert_called_with(self.edge_driver.vcns,
LB_EDGE_ID,
LB_VIP)
mock_get_edge.assert_called_with(mock.ANY, mock.ANY,
LB_SUBNET, LB_TENANT_ID)
else:
mock_set_fw_rule.assert_called_with(
self.edge_driver.vcns, LB_EDGE_ID, 'accept')
mock_get_edge.assert_called_with(mock.ANY, mock.ANY, LB_ID,
LB_VIP, mock.ANY,
LB_TENANT_ID, 'compact')
mock_add_vip_fwr.assert_called_with(self.edge_driver.vcns,
LB_EDGE_ID,
@@ -247,43 +231,16 @@ class TestEdgeLbaasV2Loadbalancer(BaseTestEdgeLbaasV2):
LB_EDGE_ID,
LB_VIP_FWR_ID,
LB_VIP)
mock_set_fw_rule.assert_called_with(
self.edge_driver.vcns, LB_EDGE_ID, 'accept')
mock_get_edge.assert_called_with(
mock.ANY, mock.ANY, LB_ID, LB_VIP,
mock.ANY, LB_TENANT_ID, flavor_name)
mock_successful_completion = (
self.lbv2_driver.load_balancer.successful_completion)
mock_successful_completion.assert_called_with(self.context,
self.lb)
mock_enable_edge_acceleration.assert_called_with(
self.edge_driver.vcns, LB_EDGE_ID)
self.lb.flavor_id = None
def test_create_with_illegal_flavor(self):
flavor_name = 'no_size'
with mock.patch.object(lb_common, 'get_lbaas_edge_id'
) as mock_get_edge, \
mock.patch.object(lb_common, 'add_vip_fw_rule'
) as mock_add_vip_fwr, \
mock.patch.object(nsxv_db,
'get_nsxv_lbaas_loadbalancer_binding_by_edge'
) as mock_get_lb_binding_by_edge, \
mock.patch('neutron.services.flavors.flavors_plugin.FlavorsPlugin.'
'get_flavor', return_value={'name': flavor_name}):
mock_get_edge.return_value = LB_EDGE_ID
mock_add_vip_fwr.return_value = LB_VIP_FWR_ID
mock_get_lb_binding_by_edge.return_value = []
self.lb.flavor_id = 'dummy'
self.assertRaises(
n_exc.InvalidInput,
self.edge_driver.loadbalancer.create,
self.context, self.lb)
self.lb.flavor_id = None
def test_update(self):
new_lb = lb_models.LoadBalancer(LB_ID, 'yyy-yyy', 'lb-name', 'heh-huh',
'some-subnet', 'port-id', LB_VIP)
LB_SUBNET, 'port-id', LB_VIP)
self.edge_driver.loadbalancer.update(self.context, self.lb, new_lb)
@@ -366,6 +323,85 @@ class TestEdgeLbaasV2Loadbalancer(BaseTestEdgeLbaasV2):
pass
class TestEdgeLbaasV2LoadbalancerOnEdge(TestEdgeLbaasV2LoadbalancerOnRtr):
@property
def _edge_getter(self):
return 'get_lbaas_edge_id'
@property
def _deploy_on_router(self):
return False
def setUp(self):
super(TestEdgeLbaasV2LoadbalancerOnEdge, self).setUp()
def test_create_with_flavor(self):
flavor_name = 'large'
with mock.patch.object(lb_common, 'get_lbaas_edge_id'
) as mock_get_edge, \
mock.patch.object(lb_common, 'add_vip_fw_rule'
) as mock_add_vip_fwr, \
mock.patch.object(lb_common, 'set_lb_firewall_default_rule'
) as mock_set_fw_rule, \
mock.patch.object(lb_common, 'enable_edge_acceleration'
) as mock_enable_edge_acceleration, \
mock.patch.object(nsxv_db,
'get_nsxv_lbaas_loadbalancer_binding_by_edge'
) as mock_get_lb_binding_by_edge, \
mock.patch.object(nsxv_db, 'add_nsxv_lbaas_loadbalancer_binding'
) as mock_db_binding,\
mock.patch('neutron.services.flavors.flavors_plugin.FlavorsPlugin.'
'get_flavor', return_value={'name': flavor_name}):
mock_get_edge.return_value = LB_EDGE_ID
mock_add_vip_fwr.return_value = LB_VIP_FWR_ID
mock_get_lb_binding_by_edge.return_value = []
self.lb.flavor_id = 'dummy'
self.edge_driver.loadbalancer.create(self.context, self.lb)
mock_add_vip_fwr.assert_called_with(self.edge_driver.vcns,
LB_EDGE_ID,
LB_ID,
LB_VIP)
mock_db_binding.assert_called_with(self.context.session,
LB_ID,
LB_EDGE_ID,
LB_VIP_FWR_ID,
LB_VIP)
mock_set_fw_rule.assert_called_with(
self.edge_driver.vcns, LB_EDGE_ID, 'accept')
mock_get_edge.assert_called_with(
mock.ANY, mock.ANY, LB_ID, LB_VIP,
mock.ANY, LB_TENANT_ID, flavor_name)
mock_successful_completion = (
self.lbv2_driver.load_balancer.successful_completion)
mock_successful_completion.assert_called_with(self.context,
self.lb)
mock_enable_edge_acceleration.assert_called_with(
self.edge_driver.vcns, LB_EDGE_ID)
self.lb.flavor_id = None
def test_create_with_illegal_flavor(self):
flavor_name = 'no_size'
with mock.patch.object(lb_common, 'get_lbaas_edge_id'
) as mock_get_edge, \
mock.patch.object(lb_common, 'add_vip_fw_rule'
) as mock_add_vip_fwr, \
mock.patch.object(nsxv_db,
'get_nsxv_lbaas_loadbalancer_binding_by_edge'
) as mock_get_lb_binding_by_edge, \
mock.patch('neutron.services.flavors.flavors_plugin.FlavorsPlugin.'
'get_flavor', return_value={'name': flavor_name}):
mock_get_edge.return_value = LB_EDGE_ID
mock_add_vip_fwr.return_value = LB_VIP_FWR_ID
mock_get_lb_binding_by_edge.return_value = []
self.lb.flavor_id = 'dummy'
self.assertRaises(
n_exc.InvalidInput,
self.edge_driver.loadbalancer.create,
self.context, self.lb)
self.lb.flavor_id = None
class TestEdgeLbaasV2Listener(BaseTestEdgeLbaasV2):
def setUp(self):
super(TestEdgeLbaasV2Listener, self).setUp()