Cleanup schema loading so unit tests run consistently

Ensure that all unit test modules can be run independently or in any
combination. A new all_models module is imported before Neutron's
SqlFixture loads DB schema from model files that derive from Neutron's
model_base.BASE_V2. Code that otherwise loads the schema for these
models is eliminated. Note that AIM and NFP models do not derive from
Neutron's base class, and are handled by their own fixtures.

Change-Id: I0b1c20429f75f77929a78b75c7e361bcc5e9b6b7
This commit is contained in:
Robert Kukura 2017-12-06 09:56:47 -05:00
parent c31162c9b1
commit 84353cca91
20 changed files with 372 additions and 314 deletions

View File

@ -0,0 +1,66 @@
# 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.
# This module must import all modules from the GBP repo that define
# SQLAlchemy database models using Neutron's
# neutron_lib.db.model_base.BASEV2 base class. It also must itself be
# imported, directly or indirectly, by every test module for which
# Neutron's neutron.tests.unit.testlib_api.SqlFixture, or any derived
# class, manages the creation of the DB schema prior to running tests
# and the clearing of DB tables between tests.
from gbpservice.neutron.db.grouppolicy.extensions import ( # noqa
apic_allowed_vm_name_db,
apic_auto_ptg_db,
apic_intra_ptg_db,
apic_reuse_bd_db,
apic_segmentation_label_db,
group_proxy_db
)
from gbpservice.neutron.db.grouppolicy import ( # noqa
group_policy_db,
group_policy_mapping_db
)
from gbpservice.neutron.db import ( # noqa
implicitsubnetpool_db,
servicechain_db
)
from gbpservice.neutron.plugins.ml2plus.drivers.apic_aim import ( # noqa
data_migrations,
db,
extension_db
)
from gbpservice.neutron.services.grouppolicy.drivers import ( # noqa
chain_mapping,
implicit_policy,
nsp_manager,
resource_mapping
)
from gbpservice.neutron.services.grouppolicy.drivers.cisco.apic import ( # noqa
port_ha_ipaddress_binding
)
from gbpservice.neutron.services.servicechain.plugins.ncp import ( # noqa
model
)
from gbpservice.neutron.services.servicechain.plugins.ncp.node_drivers import ( # noqa
heat_node_driver,
nfp_node_driver
)
from gbpservice.neutron.tests.unit.plugins.ml2plus.drivers import ( # noqa
extension_test
)
from gbpservice.neutron.tests.unit.services.grouppolicy.drivers import ( # noqa
extension_test
)
# Note that the models in gbpservice.nfp.orchestrator.db.nfp_db_model
# are managed by a separate fixture, so are not imported here.

View File

@ -23,7 +23,6 @@ import sqlalchemy as sa
from gbpservice.common import utils
from gbpservice.network.neutronv2 import local_api
from gbpservice.neutron.db.grouppolicy import group_policy_mapping_db as gpdb
from gbpservice.neutron.db import servicechain_db # noqa
from gbpservice.neutron.services.grouppolicy import (
group_policy_driver_api as api)
from gbpservice.neutron.services.grouppolicy.common import constants as gconst

View File

@ -37,7 +37,6 @@ from gbpservice.common import utils
from gbpservice.network.neutronv2 import local_api
from gbpservice.neutron.db.grouppolicy import group_policy_db as gpdb
from gbpservice.neutron.db.grouppolicy import group_policy_mapping_db as gpmdb
from gbpservice.neutron.db import servicechain_db # noqa
from gbpservice.neutron.extensions import driver_proxy_group as proxy_ext
from gbpservice.neutron.extensions import group_policy as gp_ext
from gbpservice.neutron.services.grouppolicy import (

View File

@ -20,17 +20,16 @@ import mock
from neutron.api import extensions
from neutron.api.rpc.callbacks.producer import registry
from neutron import context
from neutron.db import api as db_api
from neutron.plugins.common import constants
from neutron import policy
from neutron.tests.unit.api import test_extensions
from neutron.tests.unit.db import test_db_base_plugin_v2
from neutron_lib import constants as nl_constants
from neutron_lib.db import model_base
from neutron_lib.plugins import directory
from oslo_utils import importutils
from oslo_utils import uuidutils
from gbpservice.neutron.db import all_models # noqa
from gbpservice.neutron.db.grouppolicy import group_policy_db as gpdb
from gbpservice.neutron.db import servicechain_db as svcchain_db
from gbpservice.neutron.extensions import group_policy as gpolicy
@ -366,8 +365,6 @@ class GroupPolicyDbTestCase(GroupPolicyDBTestBase,
if not ext_mgr:
ext_mgr = extensions.PluginAwareExtensionManager.get_instance()
self.ext_api = test_extensions.setup_extensions_middleware(ext_mgr)
engine = db_api.context_manager.writer.get_engine()
model_base.BASEV2.metadata.create_all(engine)
plugins = directory.get_plugins()
self._gbp_plugin = plugins.get(constants.GROUP_POLICY)

View File

@ -14,9 +14,7 @@
import webob.exc
from neutron import context as nctx
from neutron.db import api as db_api
from neutron.tests.unit.extensions import test_l3
from neutron_lib.db import model_base
from gbpservice.neutron.db.grouppolicy import group_policy_mapping_db as gpmdb
from gbpservice.neutron.services.grouppolicy.common import exceptions as gpexc
@ -59,8 +57,6 @@ class GroupPolicyMappingDbTestCase(tgpdb.GroupPolicyDbTestCase,
core_plugin=core_plugin, gp_plugin=gp_plugin,
service_plugins=service_plugins
)
engine = db_api.context_manager.writer.get_engine()
model_base.BASEV2.metadata.create_all(engine)
class TestMappedGroupResources(GroupPolicyMappingDbTestCase,

View File

@ -16,6 +16,7 @@ from neutron.tests.unit import testlib_api
from oslo_db import exception as exc
from oslo_utils import importutils
from gbpservice.neutron.db import all_models # noqa
from gbpservice.neutron.services.grouppolicy.drivers.cisco.apic import (
port_ha_ipaddress_binding as ha)

View File

@ -50,7 +50,7 @@ from opflexagent import constants as ofcst
from oslo_config import cfg
import webob.exc
from gbpservice.neutron.db import implicitsubnetpool_db # noqa
from gbpservice.neutron.db import all_models # noqa
from gbpservice.neutron.extensions import cisco_apic_l3 as l3_ext
from gbpservice.neutron.plugins.ml2plus.drivers.apic_aim import (
extension_db as extn_db)

View File

@ -23,7 +23,7 @@ from neutron.tests.unit.db import test_db_base_plugin_v2 as test_plugin
from neutron.tests.unit.extensions import test_address_scope
from neutron_lib.plugins import directory
from gbpservice.neutron.db import implicitsubnetpool_db # noqa
from gbpservice.neutron.db import all_models # noqa
import gbpservice.neutron.extensions
from gbpservice.neutron.tests.unit.plugins.ml2plus.drivers import (
mechanism_logger as mech_logger)

View File

@ -18,7 +18,6 @@ from neutron.tests.unit.plugins.ml2 import (
test_tracked_resources as n_tracked)
from neutron.tests.unit.plugins.ml2 import test_plugin
from gbpservice.neutron.db import implicitsubnetpool_db # noqa
import gbpservice.neutron.extensions
PLUGIN_NAME = 'ml2plus'

View File

@ -0,0 +1,282 @@
# 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_lib.db import model_base
import sqlalchemy as sa
from gbpservice.neutron.services.grouppolicy import (
group_policy_driver_api as api)
from gbpservice.neutron.tests.unit.services.grouppolicy.extensions import (
test_extension as test_extension)
class TestPolicyTargetExtension(model_base.BASEV2):
__tablename__ = 'test_policy_target_extension'
policy_target_id = sa.Column(sa.String(36),
sa.ForeignKey('gp_policy_targets.id',
ondelete="CASCADE"),
primary_key=True)
pt_extension = sa.Column(sa.String(64))
class TestPolicyTargetGroupExtension(model_base.BASEV2):
__tablename__ = 'test_policy_target_group_extension'
policy_target_group_id = sa.Column(
sa.String(36), sa.ForeignKey('gp_policy_target_groups.id',
ondelete="CASCADE"),
primary_key=True)
ptg_extension = sa.Column(sa.String(64))
class TestL2PolicyExtension(model_base.BASEV2):
__tablename__ = 'test_l2_policy_extension'
l2_policy_id = sa.Column(sa.String(36), sa.ForeignKey('gp_l2_policies.id',
ondelete="CASCADE"),
primary_key=True)
l2p_extension = sa.Column(sa.String(64))
class TestL3PolicyExtension(model_base.BASEV2):
__tablename__ = 'test_l3_policy_extension'
l3_policy_id = sa.Column(sa.String(36),
sa.ForeignKey('gp_l3_policies.id',
ondelete="CASCADE"),
primary_key=True)
l3p_extension = sa.Column(sa.String(64))
class TestPolicyClassifierExtension(model_base.BASEV2):
__tablename__ = 'test_policy_classifier_extension'
policy_classifier_id = sa.Column(sa.String(36),
sa.ForeignKey('gp_policy_classifiers.id',
ondelete="CASCADE"),
primary_key=True)
pc_extension = sa.Column(sa.String(64))
class TestPolicyActionExtension(model_base.BASEV2):
__tablename__ = 'test_policy_action_extension'
policy_action_id = sa.Column(sa.String(36),
sa.ForeignKey('gp_policy_actions.id',
ondelete="CASCADE"),
primary_key=True)
pa_extension = sa.Column(sa.String(64))
class TestPolicyRuleExtension(model_base.BASEV2):
__tablename__ = 'test_policy_rule_extension'
policy_rule_id = sa.Column(sa.String(36),
sa.ForeignKey('gp_policy_rules.id',
ondelete="CASCADE"),
primary_key=True)
pr_extension = sa.Column(sa.String(64))
class TestPolicyRuleSetExtension(model_base.BASEV2):
__tablename__ = 'test_policy_rule_set_extension'
policy_rule_set_id = sa.Column(sa.String(36),
sa.ForeignKey('gp_policy_rule_sets.id',
ondelete="CASCADE"),
primary_key=True)
prs_extension = sa.Column(sa.String(64))
class TestNetworkServicePolicyExtension(model_base.BASEV2):
__tablename__ = 'test_network_service_policy_extension'
network_service_policy_id = sa.Column(sa.String(36),
sa.ForeignKey('gp_network_service_policies.id',
ondelete="CASCADE"),
primary_key=True)
nsp_extension = sa.Column(sa.String(64))
class TestExternalSegmentExtension(model_base.BASEV2):
__tablename__ = 'test_external_segment_extension'
external_segment_id = sa.Column(sa.String(36),
sa.ForeignKey('gp_external_segments.id',
ondelete="CASCADE"),
primary_key=True)
es_extension = sa.Column(sa.String(64))
class TestExternalPolicyExtension(model_base.BASEV2):
__tablename__ = 'test_external_policy_extension'
external_policy_id = sa.Column(sa.String(36),
sa.ForeignKey('gp_external_policies.id',
ondelete="CASCADE"),
primary_key=True)
ep_extension = sa.Column(sa.String(64))
class TestNatPoolExtension(model_base.BASEV2):
__tablename__ = 'test_nat_pool_extension'
nat_pool_id = sa.Column(sa.String(36),
sa.ForeignKey('gp_nat_pools.id',
ondelete="CASCADE"),
primary_key=True)
np_extension = sa.Column(sa.String(64))
class TestExtensionDriver(api.ExtensionDriver):
_supported_extension_alias = 'test_extension'
_extension_dict = test_extension.EXTENDED_ATTRIBUTES_2_0
def initialize(self):
pass
@property
def extension_alias(self):
return self._supported_extension_alias
@api.default_extension_behavior(TestPolicyTargetExtension)
def process_create_policy_target(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyTargetExtension)
def process_update_policy_target(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyTargetExtension)
def extend_policy_target_dict(self, session, result):
pass
@api.default_extension_behavior(TestPolicyTargetGroupExtension)
def process_create_policy_target_group(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyTargetGroupExtension)
def process_update_policy_target_group(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyTargetGroupExtension)
def extend_policy_target_group_dict(self, session, result):
pass
@api.default_extension_behavior(TestL2PolicyExtension)
def process_create_l2_policy(self, session, data, result):
pass
@api.default_extension_behavior(TestL2PolicyExtension)
def process_update_l2_policy(self, session, data, result):
pass
@api.default_extension_behavior(TestL2PolicyExtension)
def extend_l2_policy_dict(self, session, result):
pass
@api.default_extension_behavior(TestL3PolicyExtension)
def process_create_l3_policy(self, session, data, result):
pass
@api.default_extension_behavior(TestL3PolicyExtension)
def process_update_l3_policy(self, session, data, result):
pass
@api.default_extension_behavior(TestL3PolicyExtension)
def extend_l3_policy_dict(self, session, result):
pass
@api.default_extension_behavior(TestPolicyClassifierExtension)
def process_create_policy_classifier(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyClassifierExtension)
def process_update_policy_classifier(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyClassifierExtension)
def extend_policy_classifier_dict(self, session, result):
pass
@api.default_extension_behavior(TestPolicyActionExtension)
def process_create_policy_action(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyActionExtension)
def process_update_policy_action(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyActionExtension)
def extend_policy_action_dict(self, session, result):
pass
@api.default_extension_behavior(TestPolicyRuleExtension)
def process_create_policy_rule(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyRuleExtension)
def process_update_policy_rule(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyRuleExtension)
def extend_policy_rule_dict(self, session, result):
pass
@api.default_extension_behavior(TestPolicyRuleSetExtension)
def process_create_policy_rule_set(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyRuleSetExtension)
def process_update_policy_rule_set(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyRuleSetExtension)
def extend_policy_rule_set_dict(self, session, result):
pass
@api.default_extension_behavior(TestNetworkServicePolicyExtension)
def process_create_network_service_policy(self, session, data, result):
pass
@api.default_extension_behavior(TestNetworkServicePolicyExtension)
def process_update_network_service_policy(self, session, data, result):
pass
@api.default_extension_behavior(TestNetworkServicePolicyExtension)
def extend_network_service_policy_dict(self, session, result):
pass
@api.default_extension_behavior(TestExternalSegmentExtension)
def process_create_external_segment(self, session, data, result):
pass
@api.default_extension_behavior(TestExternalSegmentExtension)
def process_update_external_segment(self, session, data, result):
pass
@api.default_extension_behavior(TestExternalSegmentExtension)
def extend_external_segment_dict(self, session, result):
pass
@api.default_extension_behavior(TestExternalPolicyExtension)
def process_create_external_policy(self, session, data, result):
pass
@api.default_extension_behavior(TestExternalPolicyExtension)
def process_update_external_policy(self, session, data, result):
pass
@api.default_extension_behavior(TestExternalPolicyExtension)
def extend_external_policy_dict(self, session, result):
pass
@api.default_extension_behavior(TestNatPoolExtension)
def process_create_nat_pool(self, session, data, result):
pass
@api.default_extension_behavior(TestNatPoolExtension)
def process_update_nat_pool(self, session, data, result):
pass
@api.default_extension_behavior(TestNatPoolExtension)
def extend_nat_pool_dict(self, session, result):
pass

View File

@ -13,19 +13,13 @@
import os
from neutron.common import config as neutron_config # noqa
from neutron_lib.db import model_base
import sqlalchemy as sa
from gbpservice.neutron.services.grouppolicy import (
group_policy_driver_api as api)
from gbpservice.neutron.services.grouppolicy import config
from gbpservice.neutron.tests.unit import common as cm
from gbpservice.neutron.tests.unit.services.grouppolicy import (
extensions as test_ext)
from gbpservice.neutron.tests.unit.services.grouppolicy import (
test_grouppolicy_plugin as test_plugin)
from gbpservice.neutron.tests.unit.services.grouppolicy.extensions import (
test_extension as test_extension)
class ExtensionDriverTestBase(test_plugin.GroupPolicyPluginTestCase):
@ -449,268 +443,5 @@ class ExtensionDriverTestCase(ExtensionDriverTestBase):
self.assertEqual("def", val)
class TestPolicyTargetExtension(model_base.BASEV2):
__tablename__ = 'test_policy_target_extension'
policy_target_id = sa.Column(sa.String(36),
sa.ForeignKey('gp_policy_targets.id',
ondelete="CASCADE"),
primary_key=True)
pt_extension = sa.Column(sa.String(64))
class TestPolicyTargetGroupExtension(model_base.BASEV2):
__tablename__ = 'test_policy_target_group_extension'
policy_target_group_id = sa.Column(
sa.String(36), sa.ForeignKey('gp_policy_target_groups.id',
ondelete="CASCADE"),
primary_key=True)
ptg_extension = sa.Column(sa.String(64))
class TestL2PolicyExtension(model_base.BASEV2):
__tablename__ = 'test_l2_policy_extension'
l2_policy_id = sa.Column(sa.String(36), sa.ForeignKey('gp_l2_policies.id',
ondelete="CASCADE"),
primary_key=True)
l2p_extension = sa.Column(sa.String(64))
class TestL3PolicyExtension(model_base.BASEV2):
__tablename__ = 'test_l3_policy_extension'
l3_policy_id = sa.Column(sa.String(36),
sa.ForeignKey('gp_l3_policies.id',
ondelete="CASCADE"),
primary_key=True)
l3p_extension = sa.Column(sa.String(64))
class TestPolicyClassifierExtension(model_base.BASEV2):
__tablename__ = 'test_policy_classifier_extension'
policy_classifier_id = sa.Column(sa.String(36),
sa.ForeignKey('gp_policy_classifiers.id',
ondelete="CASCADE"),
primary_key=True)
pc_extension = sa.Column(sa.String(64))
class TestPolicyActionExtension(model_base.BASEV2):
__tablename__ = 'test_policy_action_extension'
policy_action_id = sa.Column(sa.String(36),
sa.ForeignKey('gp_policy_actions.id',
ondelete="CASCADE"),
primary_key=True)
pa_extension = sa.Column(sa.String(64))
class TestPolicyRuleExtension(model_base.BASEV2):
__tablename__ = 'test_policy_rule_extension'
policy_rule_id = sa.Column(sa.String(36),
sa.ForeignKey('gp_policy_rules.id',
ondelete="CASCADE"),
primary_key=True)
pr_extension = sa.Column(sa.String(64))
class TestPolicyRuleSetExtension(model_base.BASEV2):
__tablename__ = 'test_policy_rule_set_extension'
policy_rule_set_id = sa.Column(sa.String(36),
sa.ForeignKey('gp_policy_rule_sets.id',
ondelete="CASCADE"),
primary_key=True)
prs_extension = sa.Column(sa.String(64))
class TestNetworkServicePolicyExtension(model_base.BASEV2):
__tablename__ = 'test_network_service_policy_extension'
network_service_policy_id = sa.Column(sa.String(36),
sa.ForeignKey('gp_network_service_policies.id',
ondelete="CASCADE"),
primary_key=True)
nsp_extension = sa.Column(sa.String(64))
class TestExternalSegmentExtension(model_base.BASEV2):
__tablename__ = 'test_external_segment_extension'
external_segment_id = sa.Column(sa.String(36),
sa.ForeignKey('gp_external_segments.id',
ondelete="CASCADE"),
primary_key=True)
es_extension = sa.Column(sa.String(64))
class TestExternalPolicyExtension(model_base.BASEV2):
__tablename__ = 'test_external_policy_extension'
external_policy_id = sa.Column(sa.String(36),
sa.ForeignKey('gp_external_policies.id',
ondelete="CASCADE"),
primary_key=True)
ep_extension = sa.Column(sa.String(64))
class TestNatPoolExtension(model_base.BASEV2):
__tablename__ = 'test_nat_pool_extension'
nat_pool_id = sa.Column(sa.String(36),
sa.ForeignKey('gp_nat_pools.id',
ondelete="CASCADE"),
primary_key=True)
np_extension = sa.Column(sa.String(64))
class TestExtensionDriver(api.ExtensionDriver):
_supported_extension_alias = 'test_extension'
_extension_dict = test_extension.EXTENDED_ATTRIBUTES_2_0
def initialize(self):
pass
@property
def extension_alias(self):
return self._supported_extension_alias
@api.default_extension_behavior(TestPolicyTargetExtension)
def process_create_policy_target(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyTargetExtension)
def process_update_policy_target(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyTargetExtension)
def extend_policy_target_dict(self, session, result):
pass
@api.default_extension_behavior(TestPolicyTargetGroupExtension)
def process_create_policy_target_group(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyTargetGroupExtension)
def process_update_policy_target_group(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyTargetGroupExtension)
def extend_policy_target_group_dict(self, session, result):
pass
@api.default_extension_behavior(TestL2PolicyExtension)
def process_create_l2_policy(self, session, data, result):
pass
@api.default_extension_behavior(TestL2PolicyExtension)
def process_update_l2_policy(self, session, data, result):
pass
@api.default_extension_behavior(TestL2PolicyExtension)
def extend_l2_policy_dict(self, session, result):
pass
@api.default_extension_behavior(TestL3PolicyExtension)
def process_create_l3_policy(self, session, data, result):
pass
@api.default_extension_behavior(TestL3PolicyExtension)
def process_update_l3_policy(self, session, data, result):
pass
@api.default_extension_behavior(TestL3PolicyExtension)
def extend_l3_policy_dict(self, session, result):
pass
@api.default_extension_behavior(TestPolicyClassifierExtension)
def process_create_policy_classifier(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyClassifierExtension)
def process_update_policy_classifier(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyClassifierExtension)
def extend_policy_classifier_dict(self, session, result):
pass
@api.default_extension_behavior(TestPolicyActionExtension)
def process_create_policy_action(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyActionExtension)
def process_update_policy_action(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyActionExtension)
def extend_policy_action_dict(self, session, result):
pass
@api.default_extension_behavior(TestPolicyRuleExtension)
def process_create_policy_rule(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyRuleExtension)
def process_update_policy_rule(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyRuleExtension)
def extend_policy_rule_dict(self, session, result):
pass
@api.default_extension_behavior(TestPolicyRuleSetExtension)
def process_create_policy_rule_set(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyRuleSetExtension)
def process_update_policy_rule_set(self, session, data, result):
pass
@api.default_extension_behavior(TestPolicyRuleSetExtension)
def extend_policy_rule_set_dict(self, session, result):
pass
@api.default_extension_behavior(TestNetworkServicePolicyExtension)
def process_create_network_service_policy(self, session, data, result):
pass
@api.default_extension_behavior(TestNetworkServicePolicyExtension)
def process_update_network_service_policy(self, session, data, result):
pass
@api.default_extension_behavior(TestNetworkServicePolicyExtension)
def extend_network_service_policy_dict(self, session, result):
pass
@api.default_extension_behavior(TestExternalSegmentExtension)
def process_create_external_segment(self, session, data, result):
pass
@api.default_extension_behavior(TestExternalSegmentExtension)
def process_update_external_segment(self, session, data, result):
pass
@api.default_extension_behavior(TestExternalSegmentExtension)
def extend_external_segment_dict(self, session, result):
pass
@api.default_extension_behavior(TestExternalPolicyExtension)
def process_create_external_policy(self, session, data, result):
pass
@api.default_extension_behavior(TestExternalPolicyExtension)
def process_update_external_policy(self, session, data, result):
pass
@api.default_extension_behavior(TestExternalPolicyExtension)
def extend_external_policy_dict(self, session, result):
pass
@api.default_extension_behavior(TestNatPoolExtension)
def process_create_nat_pool(self, session, data, result):
pass
@api.default_extension_behavior(TestNatPoolExtension)
def process_update_nat_pool(self, session, data, result):
pass
@api.default_extension_behavior(TestNatPoolExtension)
def extend_nat_pool_dict(self, session, result):
pass
def _acronim(type):
return ''.join([x[0] for x in type.split('_')])

View File

@ -11,14 +11,9 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from neutron.db import api as db_api
from neutron_lib.db import model_base
from oslo_config import cfg
import webob.exc
from gbpservice.neutron.db import servicechain_db # noqa
from gbpservice.neutron.services.servicechain.plugins.ncp import ( # noqa
model) # noqa
from gbpservice.neutron.tests.unit.services.grouppolicy import (
test_grouppolicy_plugin as test_plugin)
@ -30,8 +25,6 @@ class ImplicitPolicyTestCase(
cfg.CONF.set_override('policy_drivers', ['implicit_policy'],
group='group_policy')
super(ImplicitPolicyTestCase, self).setUp()
engine = db_api.context_manager.writer.get_engine()
model_base.BASEV2.metadata.create_all(engine)
class TestImplicitL2Policy(ImplicitPolicyTestCase):

View File

@ -13,10 +13,8 @@
import mock
from neutron import context as nctx
from neutron.db import api as db_api
from neutron.plugins.common import constants as pconst
from neutron_lib import constants
from neutron_lib.db import model_base
from neutron_lib.plugins import directory
import webob.exc
@ -52,8 +50,6 @@ class CommonNeutronBaseTestCase(test_plugin.GroupPolicyPluginTestBase):
ml2_options=ml2_options,
sc_plugin=sc_plugin,
qos_plugin=qos_plugin)
engine = db_api.context_manager.writer.get_engine()
model_base.BASEV2.metadata.create_all(engine)
res = mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.'
'_check_router_needs_rescheduling').start()
res.return_value = None

View File

@ -15,8 +15,6 @@ import mock
from mock import call
import webob.exc
from neutron.db import api as db_api
from neutron_lib.db import model_base
from oslo_config import cfg
from vmware_nsx.common import config
from vmware_nsxlib.v3 import exceptions as nsxlib_exc
@ -41,9 +39,6 @@ class NsxPolicyMappingTestCase(test_rmd.ResourceMappingTestCase):
# REVISIT (annak): currently run with ML2 plugin
# core_plugin='vmware_nsx.plugin.NsxV3Plugin'
engine = db_api.context_manager.writer.get_engine()
model_base.BASEV2.metadata.create_all(engine)
self.driver = self._gbp_plugin.policy_driver_manager.policy_drivers[
'nsx_policy'].obj
self.nsx_policy = self.driver.nsx_policy

View File

@ -19,7 +19,6 @@ from keystonemiddleware import auth_token # noqa
import mock
import netaddr
from neutron import context as nctx
from neutron.db import api as db_api
from neutron.db.qos import models as qos_models
from neutron.extensions import external_net as external_net
from neutron.extensions import securitygroup as ext_sg
@ -29,7 +28,6 @@ from neutron.tests.unit.extensions import test_l3
from neutron.tests.unit.extensions import test_securitygroup
from neutron.tests.unit.plugins.ml2 import test_plugin as n_test_plugin
from neutron_lib import constants as cst
from neutron_lib.db import model_base
from neutron_lib.plugins import directory
from oslo_utils import uuidutils
import unittest2
@ -102,8 +100,6 @@ class ResourceMappingTestCase(test_plugin.GroupPolicyPluginTestCase):
sc_plugin=sc_plugin,
qos_plugin=qos_plugin)
engine = db_api.context_manager.writer.get_engine()
model_base.BASEV2.metadata.create_all(engine)
res = mock.patch('neutron.db.l3_db.L3_NAT_dbonly_mixin.'
'_check_router_needs_rescheduling').start()
res.return_value = None

View File

@ -14,6 +14,7 @@
import ast
import collections
from neutron.common import config
from neutron import context as n_ctx
from oslo_config import cfg
from oslo_serialization import jsonutils
@ -49,6 +50,24 @@ class ServiceChainPluginTestCase(test_servicechain_db.ServiceChainDbTestCase):
sc_plugin=sc_plugin or
SC_PLUGIN_KLASS,
gp_plugin=gp_plugin)
try:
config.cfg.CONF.keystone_authtoken.username
except config.cfg.NoSuchOptError:
config.cfg.CONF.register_opt(
config.cfg.StrOpt('username'),
'keystone_authtoken')
try:
config.cfg.CONF.keystone_authtoken.password
except config.cfg.NoSuchOptError:
config.cfg.CONF.register_opt(
config.cfg.StrOpt('password'),
'keystone_authtoken')
try:
config.cfg.CONF.keystone_authtoken.project_name
except config.cfg.NoSuchOptError:
config.cfg.CONF.register_opt(
config.cfg.StrOpt('project_name'),
'keystone_authtoken')
class BaseTestGroupPolicyPluginGroupResources(

View File

@ -16,15 +16,12 @@ import webob.exc
import mock
from neutron.common import config # noqa
from neutron import context as n_context
from neutron.db import api as db_api
from neutron.plugins.common import constants as pconst
from neutron_lib.db import model_base
from neutron_lib import exceptions as n_exc
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_serialization import jsonutils
from gbpservice.neutron.db.grouppolicy import group_policy_mapping_db # noqa
from gbpservice.neutron.services.grouppolicy import config as gpconfig # noqa
from gbpservice.neutron.services.servicechain.plugins.ncp import (
context as ncp_context)
@ -143,8 +140,6 @@ class NodeCompositionPluginTestCase(
core_plugin=core_plugin or CORE_PLUGIN,
gp_plugin=gp_plugin or GP_PLUGIN_KLASS,
sc_plugin=SC_PLUGIN_KLASS)
engine = db_api.context_manager.writer.get_engine()
model_base.BASEV2.metadata.create_all(engine)
self.driver = self.sc_plugin.driver_manager.ordered_drivers[0].obj
def _create_simple_chain(self):
@ -828,8 +823,6 @@ class TestQuotasForServiceChain(test_base.ServiceChainPluginTestCase):
core_plugin=core_plugin or CORE_PLUGIN,
gp_plugin=gp_plugin or GP_PLUGIN_KLASS,
sc_plugin=SC_PLUGIN_KLASS)
engine = db_api.context_manager.writer.get_engine()
model_base.BASEV2.metadata.create_all(engine)
self.driver = self.sc_plugin.driver_manager.ordered_drivers[0].obj
cfg.CONF.set_override('quota_servicechain_node', 1,
group='QUOTAS')

View File

@ -12,9 +12,7 @@
# limitations under the License.
import mock
from neutron.db import api as db_api
from neutron.plugins.common import constants
from neutron_lib.db import model_base
from oslo_serialization import jsonutils
import webob
@ -193,8 +191,6 @@ class NFPNodeDriverTestCase(
core_plugin=CORE_PLUGIN,
gp_plugin=GP_PLUGIN_KLASS,
sc_plugin=SC_PLUGIN_KLASS)
engine = db_api.context_manager.writer.get_engine()
model_base.BASEV2.metadata.create_all(engine)
def test_manager_initialized(self):
mgr = self.plugin.driver_manager

View File

@ -49,7 +49,7 @@ neutron.service_plugins =
nfp_fwaas = gbpservice.contrib.nfp.service_plugins.firewall.nfp_fwaas_plugin:NFPFirewallPlugin
apic_aim_l3 = gbpservice.neutron.services.apic_aim.l3_plugin:ApicL3Plugin
gbpservice.neutron.group_policy.extension_drivers =
test = gbpservice.neutron.tests.unit.services.grouppolicy.test_extension_driver_api:TestExtensionDriver
test = gbpservice.neutron.tests.unit.services.grouppolicy.drivers.extension_test: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
apic_segmentation_label = gbpservice.neutron.services.grouppolicy.drivers.extensions.apic_segmentation_label_driver:ApicSegmentationLabelExtensionDriver