x/group-based-policy
Code Issues Proposed changes

Merge "AIM Policy Driver - Part 2 - Extension Driver" into stable/mitaka

Browse Source
This commit is contained in:
Jenkins
2016-08-01 03:09:08 +00:00
committed by Gerrit Code Review
parent d68ab07d6c 23d8e27ce4
commit 3f73868297
8 changed files with 336 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
devstack/lib/apic_aim
View File

@@ -11,6 +11,8 @@ function configure_apic_aim {
# devstack/lib/neutron_plugins/ml2 does not allow overriding
# Q_PLUGIN_CLASS in override_defaults, so do it here instread
# Neutron Configuration for AIM
iniset $NEUTRON_CONF DEFAULT core_plugin ml2plus
iniset /$Q_PLUGIN_CONF_FILE apic_aim_auth auth_plugin v3password
@@ -21,6 +23,13 @@ function configure_apic_aim {
iniset /$Q_PLUGIN_CONF_FILE apic_aim_auth project_domain_name default
iniset /$Q_PLUGIN_CONF_FILE apic_aim_auth project_name admin
# GBP Configuration for AIM
# Policy drivers (REVISIT: chain_mapping might needed to be added later)
iniset $NEUTRON_CONF group_policy policy_drivers "aim_mapping"
# Extension drivers (REVISIT: proxy_group might needed to be added later)
iniset $NEUTRON_CONF group_policy extension_drivers "aim_extension"
# Service Chain (REVISIT: not overriding any defaults yet)
init_aim
}

10
gbpservice/neutron/db/grouppolicy/group_policy_mapping_db.py
View File

@@ -352,8 +352,7 @@ class GroupPolicyMappingDbPlugin(gpdb.GroupPolicyDbPlugin):
return self._get_collection_count(context, PolicyTargetGroupMapping,
filters=filters)
@log.log_method_call
def get_policy_target_groups(self, context, filters=None, fields=None,
def _get_policy_target_groups(self, context, filters=None, fields=None,
sorts=None, limit=None, marker=None,
page_reverse=False):
marker_obj = self._get_marker_obj(context, 'policy_target_group',
@@ -365,6 +364,13 @@ class GroupPolicyMappingDbPlugin(gpdb.GroupPolicyDbPlugin):
marker_obj=marker_obj,
page_reverse=page_reverse)
@log.log_method_call
def get_policy_target_groups(self, context, filters=None, fields=None,
sorts=None, limit=None, marker=None,
page_reverse=False):
return self._get_policy_target_groups(
context, filters, fields, sorts, limit, marker, page_reverse)
@log.log_method_call
def create_l2_policy(self, context, l2_policy):
l2p = l2_policy['l2_policy']

56
gbpservice/neutron/extensions/aim_driver_ext.py Normal file
View File

@@ -0,0 +1,56 @@
# 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.
from neutron.api import extensions
from gbpservice.neutron.extensions import group_policy as gp
AIM_DRIVER_EXT = 'aim-driver-extensions'
DIST_NAMES = 'apic:distinguished_names'
EXTENDED_ATTRIBUTES_2_0 = {
gp.POLICY_TARGET_GROUPS: {
DIST_NAMES: {
'allow_post': False, 'allow_put': False, 'is_visible': True},
},
}
class Aim_driver_ext(extensions.ExtensionDescriptor):
@classmethod
def get_name(cls):
return "Extensions for AIM driver"
@classmethod
def get_alias(cls):
return AIM_DRIVER_EXT
@classmethod
def get_description(cls):
return _("Adds AIM driver specific attributes to GBP resources.")
@classmethod
def get_namespace(cls):
return ("http://docs.openstack.org/ext/neutron/grouppolicy/"
"aim_driver_ext/api/v1.0")
@classmethod
def get_updated(cls):
return "2016-07-11T10:00:00-00:00"
def get_extended_resources(self, version):
if version == "2.0":
return EXTENDED_ATTRIBUTES_2_0
else:
return {}

244
gbpservice/neutron/services/grouppolicy/drivers/cisco/apic/aim_mapping.py
View File

@@ -10,11 +10,9 @@
# License for the specific language governing permissions and limitations
# under the License.
from aim import aim_manager
from aim.api import resource as aim_resource
from aim import context as aim_context
from neutron._i18n import _LI
from neutron import context as nctx
from neutron import manager
from oslo_concurrency import lockutils
from oslo_log import helpers as log
@@ -23,10 +21,15 @@ from oslo_log import log as logging
from gbpservice.neutron.extensions import group_policy as gpolicy
from gbpservice.neutron.plugins.ml2plus.drivers.apic_aim import (
mechanism_driver as aim_md)
from gbpservice.neutron.plugins.ml2plus.drivers.apic_aim.extensions import (
cisco_apic)
from gbpservice.neutron.plugins.ml2plus.drivers.apic_aim import model
from gbpservice.neutron.services.grouppolicy.common import (
constants as gp_const)
from gbpservice.neutron.services.grouppolicy.common import exceptions as gpexc
from gbpservice.neutron.services.grouppolicy.drivers import (
neutron_resources as nrd)
from gbpservice.neutron.services.grouppolicy import plugin as gbp_plugin
LOG = logging.getLogger(__name__)
@@ -47,7 +50,6 @@ class AIMMappingDriver(nrd.CommonNeutronBase):
def initialize(self):
LOG.info(_LI("APIC AIM Policy Driver initializing"))
self.db = model.DbModel()
self.aim = aim_manager.AimManager()
super(AIMMappingDriver, self).initialize()
self._apic_aim_mech_driver = None
@@ -59,97 +61,14 @@ class AIMMappingDriver(nrd.CommonNeutronBase):
ml2plus_plugin.mechanism_manager.mech_drivers['apic_aim'].obj)
return self._apic_aim_mech_driver
@property
def aim(self):
return self.aim_mech_driver.aim
@property
def name_mapper(self):
return self.aim_mech_driver.name_mapper
def _aim_tenant_name(self, context):
session = context._plugin_context.session
tenant_id = context.current['tenant_id']
tenant_name = self.name_mapper.tenant(session, tenant_id)
LOG.info(_LI("Mapped tenant_id %(id)s to %(apic_name)s"),
{'id': tenant_id, 'apic_name': tenant_name})
return tenant_name
def _aim_endpoint_group(self, context, bd_name=None, bd_tenant_name=None):
session = context._plugin_context.session
tenant_name = self._aim_tenant_name(context)
id = context.current['id']
name = context.current['name']
epg_name = self.name_mapper.policy_target_group(session, id, name)
LOG.info(_LI("Mapped ptg_id %(id)s with name %(name)s to "
"%(apic_name)s"),
{'id': id, 'name': name, 'apic_name': epg_name})
epg = aim_resource.EndpointGroup(tenant_name=str(tenant_name),
name=str(epg_name),
app_profile_name=aim_md.AP_NAME,
bd_name=bd_name,
bd_tenant_name=bd_tenant_name)
return epg
def _aim_bridge_domain(self, context, network_id, network_name):
session = context._plugin_context.session
tenant_name = self._aim_tenant_name(context)
bd_name = self.name_mapper.network(session, network_id, network_name)
LOG.info(_LI("Mapped network_id %(id)s with name %(name)s to "
"%(apic_name)s"),
{'id': network_id, 'name': network_name,
'apic_name': bd_name})
bd = aim_resource.BridgeDomain(tenant_name=str(tenant_name),
name=str(bd_name))
return bd
def _get_l2p_subnets(self, context, l2p_id, clean_session=False):
plugin_context = context._plugin_context
l2p = context._plugin.get_l2_policy(plugin_context, l2p_id)
# REVISIT: The following should be a get_subnets call via local API
return self._core_plugin.get_subnets_by_network(
plugin_context, l2p['network_id'])
def _sync_ptg_subnets(self, context, l2p):
l2p_subnets = [x['id'] for x in
self._get_l2p_subnets(context, l2p['id'])]
ptgs = context._plugin.get_policy_target_groups(
nctx.get_admin_context(), {'l2_policy_id': [l2p['id']]})
for sub in l2p_subnets:
# Add to PTG
for ptg in ptgs:
if sub not in ptg['subnets']:
try:
(context._plugin.
_add_subnet_to_policy_target_group(
nctx.get_admin_context(), ptg['id'], sub))
except gpolicy.PolicyTargetGroupNotFound as e:
LOG.warning(e)
def _use_implicit_subnet(self, context, force_add=False,
clean_session=False):
"""Implicit subnet for AIM.
The first PTG in a L2P will allocate a new subnet from the L3P.
Any subsequent PTG in the same L2P will use the same subnet.
Additional subnets will be allocated as and when the currently used
subnet runs out of IP addresses.
"""
l2p_id = context.current['l2_policy_id']
with lockutils.lock(l2p_id, external=True):
subs = self._get_l2p_subnets(context, l2p_id)
subs = set([x['id'] for x in subs])
added = []
if not subs or force_add:
l2p = context._plugin.get_l2_policy(context._plugin_context,
l2p_id)
name = APIC_OWNED + l2p['name']
added = super(
AIMMappingDriver, self)._use_implicit_subnet(
context, subnet_specifics={'name': name},
is_proxy=False, clean_session=clean_session)
context.add_subnets(subs - set(context.current['subnets']))
for subnet in added:
self._sync_ptg_subnets(context, l2p)
@log.log_method_call
def ensure_tenant(self, plugin_context, tenant_id):
self.aim_mech_driver.ensure_tenant(plugin_context, tenant_id)
@@ -183,9 +102,11 @@ class AIMMappingDriver(nrd.CommonNeutronBase):
bd_name = str(self.name_mapper.network(
session, net['id'], net['name']))
bd_tenant_name = str(self._aim_tenant_name(context))
bd_tenant_name = str(self._aim_tenant_name(
session, context.current['tenant_id']))
epg = self._aim_endpoint_group(context, bd_name, bd_tenant_name)
epg = self._aim_endpoint_group(session, context.current, bd_name,
bd_tenant_name)
self.aim.create(aim_ctx, epg)
@log.log_method_call
@@ -201,12 +122,10 @@ class AIMMappingDriver(nrd.CommonNeutronBase):
session = context._plugin_context.session
aim_ctx = aim_context.AimContext(session)
epg = self._aim_endpoint_group(context)
epg = self._aim_endpoint_group(session, context.current)
self.aim.delete(aim_ctx, epg)
self.name_mapper.delete_apic_name(session, context.current['id'])
# REVISIT(Sumit): Delete app_profile if this is last PTG
subnet_ids = [assoc['subnet_id'] for assoc in ptg_db['subnets']]
context._plugin._remove_subnets_from_policy_target_group(
@@ -226,10 +145,23 @@ class AIMMappingDriver(nrd.CommonNeutronBase):
if not l2p_db['policy_target_groups']:
self._cleanup_l2_policy(context, l2p_id, clean_session=False)
@log.log_method_call
def extend_policy_target_group_dict(self, session, result):
epg = self._get_aim_endpoint_group(session, result)
if epg:
result[cisco_apic.DIST_NAMES] = {cisco_apic.EPG: epg.dn}
@log.log_method_call
def get_policy_target_group_status(self, context):
session = context._plugin_context.session
epg = self._get_aim_endpoint_group(session, context.current)
context.current['status'] = self._map_aim_status(session, epg)
@log.log_method_call
def create_policy_target_precommit(self, context):
if not context.current['port_id']:
ptg = context._plugin.get_policy_target_group(
ptg = self._db_plugin(
context._plugin).get_policy_target_group(
context._plugin_context,
context.current['policy_target_group_id'])
subnets = self._get_subnets(
@@ -285,3 +217,123 @@ class AIMMappingDriver(nrd.CommonNeutronBase):
# rn = self.mapper.tenant_filter(tenant, pr_id)
# tf = aim_resource.TenantFilter(tenant_rn=tenant, rn=rn)
# self.aim.delete(aim_context, tf)
def _aim_tenant_name(self, session, tenant_id):
tenant_name = self.name_mapper.tenant(session, tenant_id)
LOG.debug("Mapped tenant_id %(id)s to %(apic_name)s",
{'id': tenant_id, 'apic_name': tenant_name})
return tenant_name
def _aim_endpoint_group(self, session, ptg, bd_name=None,
bd_tenant_name=None):
# This returns a new AIM EPG resource
tenant_id = ptg['tenant_id']
tenant_name = self._aim_tenant_name(session, tenant_id)
id = ptg['id']
name = ptg['name']
epg_name = self.name_mapper.policy_target_group(session, id, name)
LOG.debug("Mapped ptg_id %(id)s with name %(name)s to %(apic_name)s",
{'id': id, 'name': name, 'apic_name': epg_name})
kwargs = {'tenant_name': str(tenant_name),
'name': str(epg_name),
'app_profile_name': aim_md.AP_NAME}
if bd_name:
kwargs['bd_name'] = bd_name
if bd_tenant_name:
kwargs['bd_tenant_name'] = bd_tenant_name
epg = aim_resource.EndpointGroup(**kwargs)
return epg
def _get_aim_endpoint_group(self, session, ptg):
# This gets an EPG from the AIM DB
epg = self._aim_endpoint_group(session, ptg)
aim_ctx = aim_context.AimContext(session)
epg_fetched = self.aim.get(aim_ctx, epg)
if not epg_fetched:
LOG.debug("No EPG found in AIM DB")
else:
LOG.debug("Got epg: %s", epg_fetched.__dict__)
return epg_fetched
def _aim_bridge_domain(self, session, tenant_id, network_id, network_name):
# This returns a new AIM BD resource
tenant_name = self._aim_tenant_name(session, tenant_id)
bd_name = self.name_mapper.network(session, network_id, network_name)
LOG.info(_LI("Mapped network_id %(id)s with name %(name)s to "
"%(apic_name)s"),
{'id': network_id, 'name': network_name,
'apic_name': bd_name})
bd = aim_resource.BridgeDomain(tenant_name=str(tenant_name),
name=str(bd_name))
return bd
def _get_l2p_subnets(self, context, l2p_id, clean_session=False):
plugin_context = context._plugin_context
l2p = context._plugin.get_l2_policy(plugin_context, l2p_id)
# REVISIT: The following should be a get_subnets call via local API
return self._core_plugin.get_subnets_by_network(
plugin_context, l2p['network_id'])
def _sync_ptg_subnets(self, context, l2p):
l2p_subnets = [x['id'] for x in
self._get_l2p_subnets(context, l2p['id'])]
ptgs = context._plugin._get_policy_target_groups(
context._plugin_context.elevated(), {'l2_policy_id': [l2p['id']]})
for sub in l2p_subnets:
# Add to PTG
for ptg in ptgs:
if sub not in ptg['subnets']:
try:
(context._plugin.
_add_subnet_to_policy_target_group(
context._plugin_context.elevated(),
ptg['id'], sub))
except gpolicy.PolicyTargetGroupNotFound as e:
LOG.warning(e)
def _use_implicit_subnet(self, context, force_add=False,
clean_session=False):
"""Implicit subnet for AIM.
The first PTG in a L2P will allocate a new subnet from the L3P.
Any subsequent PTG in the same L2P will use the same subnet.
Additional subnets will be allocated as and when the currently used
subnet runs out of IP addresses.
"""
l2p_id = context.current['l2_policy_id']
with lockutils.lock(l2p_id, external=True):
subs = self._get_l2p_subnets(context, l2p_id)
subs = set([x['id'] for x in subs])
added = []
if not subs or force_add:
l2p = context._plugin.get_l2_policy(
context._plugin_context, l2p_id)
name = APIC_OWNED + l2p['name']
added = super(
AIMMappingDriver, self)._use_implicit_subnet(
context, subnet_specifics={'name': name},
is_proxy=False, clean_session=clean_session)
context.add_subnets(subs - set(context.current['subnets']))
for subnet in added:
self._sync_ptg_subnets(context, l2p)
def _map_aim_status(self, session, aim_resource_obj):
# Note that this implementation assumes that this driver
# is the only policy driver configured, and no merging
# with any previous status is required.
aim_ctx = aim_context.AimContext(session)
aim_status = self.aim.get_status(aim_ctx, aim_resource_obj)
if not aim_status:
# REVIST(Sumit)
return gp_const.STATUS_BUILD
if aim_status.is_error():
return gp_const.STATUS_ERROR
elif aim_status.is_build():
return gp_const.STATUS_BUILD
else:
return gp_const.STATUS_ACTIVE
def _db_plugin(self, plugin_obj):
return super(gbp_plugin.GroupPolicyPlugin, plugin_obj)

49
gbpservice/neutron/services/grouppolicy/drivers/extensions/aim_mapping_extension_driver.py Normal file
View File

@@ -0,0 +1,49 @@
# 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.
from neutron._i18n import _LI
from neutron import manager as n_manager
from oslo_log import log as logging
from gbpservice.neutron.extensions import aim_driver_ext
from gbpservice.neutron.services.grouppolicy import (
group_policy_driver_api as api)
LOG = logging.getLogger(__name__)
class AIMExtensionDriver(api.ExtensionDriver):
_supported_extension_alias = aim_driver_ext.AIM_DRIVER_EXT
_extension_dict = aim_driver_ext.EXTENDED_ATTRIBUTES_2_0
def __init__(self):
LOG.info(_LI("AIM Extension __init__"))
self._policy_driver = None
@property
def _pd(self):
if not self._policy_driver:
gbp_plugin = (n_manager.NeutronManager.get_service_plugins()
.get("GROUP_POLICY"))
policy_mgr = gbp_plugin.policy_driver_manager
self._policy_driver = policy_mgr.policy_drivers['aim_mapping'].obj
return self._policy_driver
def initialize(self):
pass
@property
def extension_alias(self):
return self._supported_extension_alias
def extend_policy_target_group_dict(self, session, result):
self._pd.extend_policy_target_group_dict(session, result)

76
gbpservice/neutron/tests/unit/services/grouppolicy/test_aim_mapping_driver.py
View File

@@ -13,23 +13,24 @@
import mock
from aim import aim_manager
from aim.api import resource as aim_resource
from aim import context as aim_context
from aim.db import model_base as aim_model_base
from keystoneclient.v3 import client as ksc_client
from neutron import context as nctx
from neutron.db import api as db_api
from oslo_log import log as logging
import webob.exc
from gbpservice.neutron.plugins.ml2plus.drivers.apic_aim import (
mechanism_driver as aim_md)
from gbpservice.neutron.plugins.ml2plus.drivers.apic_aim import apic_mapper
from gbpservice.neutron.plugins.ml2plus.drivers.apic_aim import model
from gbpservice.neutron.services.grouppolicy.common import (
constants as gp_const)
from gbpservice.neutron.services.grouppolicy import config
from gbpservice.neutron.tests.unit.plugins.ml2plus import (
test_apic_aim as test_aim_md)
from gbpservice.neutron.tests.unit.services.grouppolicy import (
test_extension_driver_api as test_ext_base)
from gbpservice.neutron.tests.unit.services.grouppolicy import (
test_neutron_resources_driver as test_nr_base)
@@ -37,7 +38,10 @@ from gbpservice.neutron.tests.unit.services.grouppolicy import (
ML2PLUS_PLUGIN = 'gbpservice.neutron.plugins.ml2plus.plugin.Ml2PlusPlugin'
class AIMBaseTestCase(test_nr_base.CommonNeutronBaseTestCase):
class AIMBaseTestCase(test_nr_base.CommonNeutronBaseTestCase,
test_ext_base.ExtensionDriverTestBase):
_extension_drivers = ['aim_extension']
_extension_path = None
def setUp(self, policy_drivers=None, core_plugin=None, ml2_options=None,
sc_plugin=None, **kwargs):
@@ -65,16 +69,32 @@ class AIMBaseTestCase(test_nr_base.CommonNeutronBaseTestCase):
engine = db_api.get_engine()
aim_model_base.Base.metadata.create_all(engine)
self._aim = aim_manager.AimManager()
self._aim_mgr = None
self._aim_context = aim_context.AimContext(
self._neutron_context.session)
self._db = model.DbModel()
self._name_mapper = apic_mapper.APICNameMapper(self._db, logging)
self._name_mapper = None
def tearDown(self):
ksc_client.Client = self.saved_keystone_client
super(AIMBaseTestCase, self).tearDown()
@property
def aim_mgr(self):
if not self._aim_mgr:
self._aim_mgr = (
self._gbp_plugin.policy_driver_manager.policy_drivers[
'aim_mapping'].obj.aim)
return self._aim_mgr
@property
def name_mapper(self):
if not self._name_mapper:
self._name_mapper = (
self._gbp_plugin.policy_driver_manager.policy_drivers[
'aim_mapping'].obj.name_mapper)
return self._name_mapper
class TestL2Policy(test_nr_base.TestL2Policy, AIMBaseTestCase):
@@ -83,32 +103,48 @@ class TestL2Policy(test_nr_base.TestL2Policy, AIMBaseTestCase):
class TestPolicyTargetGroup(AIMBaseTestCase):
def _test_aim_resource_status(self, aim_resource_obj, gbp_resource):
aim_status = self.aim_mgr.get_status(self._aim_context,
aim_resource_obj)
if aim_status.is_error():
self.assertEqual(gp_const.STATUS_ERROR, gbp_resource['status'])
elif aim_status.is_build():
self.assertEqual(gp_const.STATUS_BUILD, gbp_resource['status'])
else:
self.assertEqual(gp_const.STATUS_ACTIVE, gbp_resource['status'])
def test_policy_target_group_lifecycle_implicit_l2p(self):
ptg = self.create_policy_target_group(
name="ptg1")['policy_target_group']
ptg_id = ptg['id']
self.show_policy_target_group(ptg_id, expected_res_status=200)
ptg_show = self.show_policy_target_group(
ptg_id, expected_res_status=200)['policy_target_group']
self.show_l2_policy(ptg['l2_policy_id'], expected_res_status=200)
req = self.new_show_request('subnets', ptg['subnets'][0], fmt=self.fmt)
res = self.deserialize(self.fmt, req.get_response(self.api))
self.assertIsNotNone(res['subnet']['id'])
ptg_name = ptg['name']
aim_epg_name = str(self._name_mapper.policy_target_group(
aim_epg_name = str(self.name_mapper.policy_target_group(
self._neutron_context.session, ptg_id, ptg_name))
aim_tenant_name = str(self._name_mapper.tenant(
aim_tenant_name = str(self.name_mapper.tenant(
self._neutron_context.session, self._tenant_id))
aim_app_profile_name = aim_md.AP_NAME
aim_app_profiles = self._aim.find(
aim_app_profiles = self.aim_mgr.find(
self._aim_context, aim_resource.ApplicationProfile,
tenant_name=aim_tenant_name, name=aim_app_profile_name)
self.assertEqual(1, len(aim_app_profiles))
aim_epgs = self._aim.find(
aim_epgs = self.aim_mgr.find(
self._aim_context, aim_resource.EndpointGroup, name=aim_epg_name)
self.assertEqual(1, len(aim_epgs))
self.assertEqual(aim_epg_name, aim_epgs[0].name)
self.assertEqual(aim_tenant_name, aim_epgs[0].tenant_name)
self._test_aim_resource_status(aim_epgs[0], ptg)
self.assertEqual(aim_epgs[0].dn,
ptg_show['apic:distinguished_names']['EndpointGroup'])
self._test_aim_resource_status(aim_epgs[0], ptg_show)
self.delete_policy_target_group(ptg_id, expected_res_status=204)
self.show_policy_target_group(ptg_id, expected_res_status=404)
# Implicitly created subnet should be deleted
@@ -118,7 +154,7 @@ class TestPolicyTargetGroup(AIMBaseTestCase):
# Implicitly created L2P should be deleted
self.show_l2_policy(ptg['l2_policy_id'], expected_res_status=404)
aim_epgs = self._aim.find(
aim_epgs = self.aim_mgr.find(
self._aim_context, aim_resource.EndpointGroup, name=aim_epg_name)
self.assertEqual(0, len(aim_epgs))
@@ -136,21 +172,23 @@ class TestPolicyTargetGroup(AIMBaseTestCase):
res = self.deserialize(self.fmt, req.get_response(self.api))
self.assertIsNotNone(res['subnet']['id'])
ptg_name = ptg['name']
aim_epg_name = str(self._name_mapper.policy_target_group(
aim_epg_name = str(self.name_mapper.policy_target_group(
self._neutron_context.session, ptg_id, ptg_name))
aim_tenant_name = str(self._name_mapper.tenant(
aim_tenant_name = str(self.name_mapper.tenant(
self._neutron_context.session, self._tenant_id))
aim_app_profile_name = aim_md.AP_NAME
aim_app_profiles = self._aim.find(
aim_app_profiles = self.aim_mgr.find(
self._aim_context, aim_resource.ApplicationProfile,
tenant_name=aim_tenant_name, name=aim_app_profile_name)
self.assertEqual(1, len(aim_app_profiles))
aim_epgs = self._aim.find(
aim_epgs = self.aim_mgr.find(
self._aim_context, aim_resource.EndpointGroup, name=aim_epg_name)
self.assertEqual(1, len(aim_epgs))
self.assertEqual(aim_epg_name, aim_epgs[0].name)
self.assertEqual(aim_tenant_name, aim_epgs[0].tenant_name)
self._test_aim_resource_status(aim_epgs[0], ptg)
self.delete_policy_target_group(ptg_id, expected_res_status=204)
self.show_policy_target_group(ptg_id, expected_res_status=404)
# Implicitly created subnet should be deleted
@@ -160,7 +198,7 @@ class TestPolicyTargetGroup(AIMBaseTestCase):
# Explicitly created L2P should not be deleted
self.show_l2_policy(ptg['l2_policy_id'], expected_res_status=200)
aim_epgs = self._aim.find(
aim_epgs = self.aim_mgr.find(
self._aim_context, aim_resource.EndpointGroup, name=aim_epg_name)
self.assertEqual(0, len(aim_epgs))
@@ -340,7 +378,7 @@ class TestPolicyRule(AIMBaseTestCase):
pr_id = pr['id']
pr_name = pr['name']
rn = self._aim_mapper.tenant_filter(tenant, pr_id, name=pr_name)
aim_pr = self._aim.find(
aim_pr = self.aim_mgr.find(
self._aim_context, aim_resource.TenantFilter, rn=rn)
self.assertEqual(1, len(aim_pr))
self.assertEqual(rn, aim_pr[0].rn)
@@ -349,6 +387,6 @@ class TestPolicyRule(AIMBaseTestCase):
self.delete_policy_rule(pr_id, expected_res_status=204)
self.show_policy_rule(pr_id, expected_res_status=404)
aim_pr = self._aim.find(
aim_pr = self.aim_mgr.find(
self._aim_context, aim_resource.TenantFilter, rn=rn)
self.assertEqual(0, len(aim_pr))

9
gbpservice/neutron/tests/unit/services/grouppolicy/test_extension_driver_api.py
View File

@@ -12,7 +12,7 @@
import os
from neutron.common import config # noqa
from neutron.common import config as neutron_config # noqa
from neutron.db import model_base
import sqlalchemy as sa
@@ -32,14 +32,17 @@ class ExtensionDriverTestBase(test_plugin.GroupPolicyPluginTestCase):
_extension_drivers = ['test']
_extension_path = os.path.dirname(os.path.abspath(test_ext.__file__))
def setUp(self):
def setUp(self, policy_drivers=None, core_plugin=None,
ml2_options=None, sc_plugin=None):
config.cfg.CONF.set_override('extension_drivers',
self._extension_drivers,
group='group_policy')
if self._extension_path:
config.cfg.CONF.set_override(
'api_extensions_path', self._extension_path)
super(ExtensionDriverTestBase, self).setUp()
super(ExtensionDriverTestBase, self).setUp(
core_plugin=core_plugin, ml2_options=ml2_options,
sc_plugin=sc_plugin)
class ExtensionDriverTestCase(ExtensionDriverTestBase):

1
setup.cfg
View File

@@ -48,6 +48,7 @@ neutron.service_plugins =
gbpservice.neutron.group_policy.extension_drivers =
test = gbpservice.neutron.tests.unit.services.grouppolicy.test_extension_driver_api:TestExtensionDriver
proxy_group = gbpservice.neutron.services.grouppolicy.drivers.extensions.proxy_group_driver:ProxyGroupDriver
aim_extension = gbpservice.neutron.services.grouppolicy.drivers.extensions.aim_mapping_extension_driver:AIMExtensionDriver
gbpservice.neutron.group_policy.policy_drivers =
dummy = gbpservice.neutron.services.grouppolicy.drivers.dummy_driver:NoopDriver
implicit_policy = gbpservice.neutron.services.grouppolicy.drivers.implicit_policy:ImplicitPolicyDriver