# vim: tabstop=4 shiftwidth=4 softtabstop=4 # # Copyright 2013 VMware, Inc # # 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. # # @author: linb, VMware import mock from neutron.common import config as n_config from neutron import context from neutron.openstack.common import uuidutils from neutron.plugins.nicira.dbexts import vcns_db from neutron.plugins.nicira.vshield.common import exceptions as vcns_exc from neutron.plugins.nicira.vshield import vcns_driver from neutron.tests.unit.db.loadbalancer import test_db_loadbalancer from neutron.tests.unit.vmware import get_fake_conf from neutron.tests.unit.vmware import VCNS_NAME from neutron.tests.unit.vmware.vshield import fake_vcns _uuid = uuidutils.generate_uuid VSE_ID = 'edge-1' POOL_MAP_INFO = { 'pool_id': None, 'edge_id': VSE_ID, 'pool_vseid': 'pool-1'} VCNS_CONFIG_FILE = get_fake_conf("vcns.ini.test") class VcnsDriverTestCase(test_db_loadbalancer.LoadBalancerPluginDbTestCase): def vcns_loadbalancer_patch(self): instance = self.mock_vcns.start() instance.return_value.create_vip.side_effect = ( self.fc2.create_vip) instance.return_value.get_vip.side_effect = ( self.fc2.get_vip) instance.return_value.update_vip.side_effect = ( self.fc2.update_vip) instance.return_value.delete_vip.side_effect = ( self.fc2.delete_vip) instance.return_value.create_pool.side_effect = ( self.fc2.create_pool) instance.return_value.get_pool.side_effect = ( self.fc2.get_pool) instance.return_value.update_pool.side_effect = ( self.fc2.update_pool) instance.return_value.delete_pool.side_effect = ( self.fc2.delete_pool) instance.return_value.create_health_monitor.side_effect = ( self.fc2.create_health_monitor) instance.return_value.get_health_monitor.side_effect = ( self.fc2.get_health_monitor) instance.return_value.update_health_monitor.side_effect = ( self.fc2.update_health_monitor) instance.return_value.delete_health_monitor.side_effect = ( self.fc2.delete_health_monitor) instance.return_value.create_app_profile.side_effect = ( self.fc2.create_app_profile) instance.return_value.delete_app_profile.side_effect = ( self.fc2.delete_app_profile) self.pool_id = None self.vip_id = None def setUp(self): n_config.parse(['--config-file', VCNS_CONFIG_FILE]) # mock vcns self.fc2 = fake_vcns.FakeVcns(unique_router_name=False) self.mock_vcns = mock.patch(VCNS_NAME, autospec=True) self.vcns_loadbalancer_patch() self.driver = vcns_driver.VcnsDriver(mock.Mock()) super(VcnsDriverTestCase, self).setUp() self.addCleanup(self.fc2.reset_all) self.addCleanup(self.mock_vcns.stop) def tearDown(self): super(VcnsDriverTestCase, self).tearDown() class TestEdgeLbDriver(VcnsDriverTestCase): def test_create_and_get_vip(self): ctx = context.get_admin_context() with self.pool(no_delete=True) as pool: self.pool_id = pool['pool']['id'] POOL_MAP_INFO['pool_id'] = pool['pool']['id'] vcns_db.add_vcns_edge_pool_binding(ctx.session, POOL_MAP_INFO) with self.vip(pool=pool) as res: vip_create = res['vip'] self.driver.create_vip(ctx, VSE_ID, vip_create) vip_get = self.driver.get_vip(ctx, vip_create['id']) for k, v in vip_get.iteritems(): self.assertEqual(vip_create[k], v) def test_update_vip(self): ctx = context.get_admin_context() with self.pool(no_delete=True) as pool: self.pool_id = pool['pool']['id'] POOL_MAP_INFO['pool_id'] = pool['pool']['id'] vcns_db.add_vcns_edge_pool_binding(ctx.session, POOL_MAP_INFO) with self.vip(pool=pool) as res: vip_create = res['vip'] self.driver.create_vip(ctx, VSE_ID, vip_create) vip_update = {'id': vip_create['id'], 'pool_id': pool['pool']['id'], 'name': 'update_name', 'description': 'description', 'address': 'update_address', 'port_id': 'update_port_id', 'protocol_port': 'protocol_port', 'protocol': 'update_protocol'} self.driver.update_vip(ctx, vip_update) vip_get = self.driver.get_vip(ctx, vip_create['id']) for k, v in vip_get.iteritems(): if k in vip_update: self.assertEqual(vip_update[k], v) def test_delete_vip(self): ctx = context.get_admin_context() with self.pool(no_delete=True) as pool: self.pool_id = pool['pool']['id'] POOL_MAP_INFO['pool_id'] = pool['pool']['id'] vcns_db.add_vcns_edge_pool_binding(ctx.session, POOL_MAP_INFO) with self.vip(pool=pool) as res: vip_create = res['vip'] self.driver.create_vip(ctx, VSE_ID, vip_create) self.driver.delete_vip(ctx, vip_create['id']) self.assertRaises(vcns_exc.VcnsNotFound, self.driver.get_vip, ctx, vip_create['id']) #Test Pool Operation def test_create_and_get_pool(self): ctx = context.get_admin_context() with self.pool(no_delete=True) as p: self.pool_id = p['pool']['id'] pool_create = p['pool'] self.driver.create_pool(ctx, VSE_ID, pool_create, []) pool_get = self.driver.get_pool(ctx, pool_create['id'], VSE_ID) for k, v in pool_get.iteritems(): self.assertEqual(pool_create[k], v) def test_update_pool(self): ctx = context.get_admin_context() with self.pool(no_delete=True) as p: self.pool_id = p['pool']['id'] pool_create = p['pool'] self.driver.create_pool(ctx, VSE_ID, pool_create, []) pool_update = {'id': pool_create['id'], 'lb_method': 'lb_method', 'name': 'update_name', 'members': [], 'health_monitors': []} self.driver.update_pool(ctx, VSE_ID, pool_update, []) pool_get = self.driver.get_pool(ctx, pool_create['id'], VSE_ID) for k, v in pool_get.iteritems(): if k in pool_update: self.assertEqual(pool_update[k], v) def test_delete_pool(self): ctx = context.get_admin_context() with self.pool(no_delete=True) as p: self.pool_id = p['pool']['id'] pool_create = p['pool'] self.driver.create_pool(ctx, VSE_ID, pool_create, []) self.driver.delete_pool(ctx, pool_create['id'], VSE_ID) self.assertRaises(vcns_exc.VcnsNotFound, self.driver.get_pool, ctx, pool_create['id'], VSE_ID) def test_create_and_get_monitor(self): ctx = context.get_admin_context() with self.health_monitor(no_delete=True) as m: monitor_create = m['health_monitor'] self.driver.create_health_monitor(ctx, VSE_ID, monitor_create) monitor_get = self.driver.get_health_monitor( ctx, monitor_create['id'], VSE_ID) for k, v in monitor_get.iteritems(): self.assertEqual(monitor_create[k], v) def test_update_health_monitor(self): ctx = context.get_admin_context() with self.health_monitor(no_delete=True) as m: monitor_create = m['health_monitor'] self.driver.create_health_monitor( ctx, VSE_ID, monitor_create) monitor_update = {'id': monitor_create['id'], 'delay': 'new_delay', 'timeout': "new_timeout", 'type': 'type', 'max_retries': "max_retries"} self.driver.update_health_monitor( ctx, VSE_ID, monitor_create, monitor_update) monitor_get = self.driver.get_health_monitor( ctx, monitor_create['id'], VSE_ID) for k, v in monitor_get.iteritems(): if k in monitor_update: self.assertEqual(monitor_update[k], v) def test_delete_health_monitor(self): ctx = context.get_admin_context() with self.health_monitor(no_delete=True) as m: monitor_create = m['health_monitor'] self.driver.create_health_monitor(ctx, VSE_ID, monitor_create) self.driver.delete_health_monitor( ctx, monitor_create['id'], VSE_ID) self.assertRaises(vcns_exc.VcnsNotFound, self.driver.get_health_monitor, ctx, monitor_create['id'], VSE_ID)