1ba129e6a3
Old names are kept for backward compatibility. Last references to the old NVP plugins are going to be dropped in Juno. Completes blueprint nicira-plugin-renaming Change-Id: Ia1aa4f658891e7ce9469748432c338353195edfd
229 lines
9.7 KiB
Python
229 lines
9.7 KiB
Python
# 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.vmware.dbexts import vcns_db
|
|
from neutron.plugins.vmware.vshield.common import exceptions as vcns_exc
|
|
from neutron.plugins.vmware.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)
|