import contextlib
import logging
import os
+import os.path
import sys
import eventlet.timeout
from oslo.config import cfg
import testtools
+from neutron.common import config
from neutron.common import constants as const
from neutron import manager
from neutron.openstack.common.notifier import api as notifier_api
from neutron.openstack.common.notifier import test_notifier
+from neutron.openstack.common import rpc
+from neutron.openstack.common.rpc import impl_fake
from neutron.tests import post_mortem_debug
TRUE_STRING = ['True', '1']
LOG_FORMAT = "%(asctime)s %(levelname)8s [%(name)s] %(message)s"
+ROOTDIR = os.path.dirname(__file__)
+ETCDIR = os.path.join(ROOTDIR, 'etc')
+
+
+def etcdir(*p):
+ return os.path.join(ETCDIR, *p)
+
def fake_use_fatal_exceptions(*args):
return True
notification_driver = [test_notifier.__name__]
cfg.CONF.set_override("notification_driver", notification_driver)
+ @staticmethod
+ def config_parse(conf=None, args=None):
+ """Create the default configurations."""
+ # neutron.conf.test includes rpc_backend which needs to be cleaned up
+ if args is None:
+ args = ['--config-file', etcdir('neutron.conf.test')]
+ if conf is None:
+ config.parse(args=args)
+ else:
+ conf(args)
+
+ def _cleanup_rpc_backend(self):
+ rpc._RPCIMPL = None
+ impl_fake.CONSUMERS.clear()
+
def setUp(self):
super(BaseTestCase, self).setUp()
+ self.addCleanup(self._cleanup_rpc_backend)
# Configure this first to ensure pm debugging support for setUp()
if os.environ.get('OS_POST_MORTEM_DEBUG') in TRUE_STRING:
import mock
from oslo.config import cfg
-from neutron.common import config as neutron_config
from neutron.plugins.cisco.common import config as cisco_config
from neutron.tests import base
-from neutron.tests.unit import test_api_v2
class TestCiscoNexusPluginConfig(base.BaseTestCase):
def setUp(self):
# Point neutron config file to: neutron/tests/etc/neutron.conf.test
- args = ['--config-file', test_api_v2.etcdir('neutron.conf.test')]
- neutron_config.parse(args=args)
+ self.config_parse()
super(TestCiscoNexusPluginConfig, self).setUp()
import mock
-from neutron.common import config
from neutron import context
from neutron.plugins.cisco.common import cisco_constants as const
from neutron.plugins.cisco.common import config as cisco_config
from neutron.plugins.cisco.models import virt_phy_sw_v2
from neutron.plugins.cisco.nexus import cisco_nexus_plugin_v2
from neutron.tests import base
-from neutron.tests.unit import test_api_v2
class TestCiscoPluginModel(base.BaseTestCase):
def setUp(self):
# Point config file to: neutron/tests/etc/neutron.conf.test
- args = ['--config-file', test_api_v2.etcdir('neutron.conf.test')]
- config.parse(args=args)
+ self.config_parse()
super(TestCiscoPluginModel, self).setUp()
import contextlib
import logging
-import os
import mock
from oslo.config import cfg
)
NOOP_DRIVER_KLASS = ('neutron.tests.unit.db.loadbalancer.test_db_loadbalancer.'
'NoopLbaaSDriver')
-ROOTDIR = os.path.dirname(__file__) + '../../../..'
-ETCDIR = os.path.join(ROOTDIR, 'etc')
extensions_path = ':'.join(neutron.extensions.__path__)
_subnet_id = "0c798ed8-33ba-11e2-8b28-000c291c4d14"
-def etcdir(*p):
- return os.path.join(ETCDIR, *p)
-
-
class NoopLbaaSDriver(abstract_driver.LoadBalancerAbstractDriver):
"""A dummy lbass driver that that only performs object deletion."""
# License for the specific language governing permissions and limitations
# under the License.
-import os
-
import mock
from oslo.config import cfg
import testtools
from neutron.tests import base
CONF_FILE = ""
-ROOTDIR = os.path.dirname(os.path.dirname(__file__))
-ETCDIR = os.path.join(ROOTDIR, 'etc')
META_PATH = "neutron.plugins.metaplugin"
FAKE_PATH = "neutron.tests.unit.metaplugin"
PROXY_PATH = "%s.proxy_neutron_plugin.ProxyPluginV2" % META_PATH
""".strip() % (FAKE_PATH, FAKE_PATH)
-def etcdir(*p):
- return os.path.join(ETCDIR, *p)
-
-
def setup_metaplugin_conf(has_l3=True):
cfg.CONF.set_override('auth_url', 'http://localhost:35357/v2.0',
'PROXY')
import mock
from oslo.config import cfg
-from neutron.common import config as neutron_config
from neutron.plugins.ml2.drivers.cisco.nexus import config as cisco_config
from neutron.tests import base
-from neutron.tests.unit import test_api_v2
class TestCiscoNexusPluginConfig(base.BaseTestCase):
def setUp(self):
- # Point neutron config file to: neutron/tests/etc/neutron.conf.test
- args = ['--config-file', test_api_v2.etcdir('neutron.conf.test')]
- neutron_config.parse(args=args)
+ self.config_parse()
super(TestCiscoNexusPluginConfig, self).setUp()
def test_config_parse_error(self):
from neutron.api.v2 import attributes
from neutron.api.v2 import base as v2_base
from neutron.api.v2 import router
-from neutron.common import config
from neutron.common import exceptions as n_exc
from neutron import context
from neutron.manager import NeutronManager
ROOTDIR = os.path.dirname(os.path.dirname(__file__))
-ETCDIR = os.path.join(ROOTDIR, 'etc')
EXTDIR = os.path.join(ROOTDIR, 'unit/extensions')
_uuid = uuidutils.generate_uuid
-def etcdir(*p):
- return os.path.join(ETCDIR, *p)
-
-
def _get_path(resource, id=None, action=None, fmt=None):
path = '/%s' % resource
# Ensure existing ExtensionManager is not used
PluginAwareExtensionManager._instance = None
# Create the default configurations
- args = ['--config-file', etcdir('neutron.conf.test')]
- config.parse(args=args)
+ self.config_parse()
# Update the plugin
self.setup_coreplugin(plugin)
cfg.CONF.set_override('allow_pagination', True)
for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems():
self.saved_attr_map[resource] = attrs.copy()
- args = ['--config-file', etcdir('neutron.conf.test')]
- config.parse(args=args)
+ self.config_parse()
self.setup_coreplugin(plugin)
self._plugin_patcher = mock.patch(plugin, autospec=True)
self.saved_attr_map[resource] = attrs.copy()
# Create the default configurations
- args = ['--config-file', etcdir('neutron.conf.test')]
- config.parse(args=args)
+ self.config_parse()
# Update the plugin and extensions path
self.setup_coreplugin(plugin)
from neutron.api import extensions
from neutron.api.v2 import attributes
-from neutron.common import config
from neutron import quota
from neutron.tests.unit import test_api_v2
from neutron.tests.unit import test_extensions
self.addCleanup(self._resotre_attr_map)
# Create the default configurations
- args = ['--config-file', test_api_v2.etcdir('neutron.conf.test')]
- config.parse(args)
+ self.config_parse()
#just stubbing core plugin with plugin
self.setup_coreplugin(plugin)
import contextlib
import copy
-import os
import mock
from oslo.config import cfg
from neutron.api.v2 import attributes
from neutron.api.v2.attributes import ATTR_NOT_SPECIFIED
from neutron.api.v2.router import APIRouter
-from neutron.common import config
from neutron.common import constants
from neutron.common import exceptions as n_exc
from neutron.common.test_lib import test_config
from neutron.tests.unit import testlib_api
DB_PLUGIN_KLASS = 'neutron.db.db_base_plugin_v2.NeutronDbPluginV2'
-ROOTDIR = os.path.dirname(os.path.dirname(__file__))
-ETCDIR = os.path.join(ROOTDIR, 'etc')
def optional_ctx(obj, fallback):
return context_wrapper()
-def etcdir(*p):
- return os.path.join(ETCDIR, *p)
-
-
def _fake_get_pagination_helper(self, request):
return api_common.PaginationEmulatedHelper(request, self._primary_key)
plugin = DB_PLUGIN_KLASS
# Create the default configurations
- args = ['--config-file', etcdir('neutron.conf.test')]
+ args = ['--config-file', base.etcdir('neutron.conf.test')]
# If test_config specifies some config-file, use it, as well
for config_file in test_config.get('config_files', []):
args.extend(['--config-file', config_file])
- config.parse(args=args)
+ self.config_parse(args=args)
# Update the plugin
self.setup_coreplugin(plugin)
cfg.CONF.set_override(
# under the License.
import copy
-import os
import sys
import uuid
from neutron.tests import base
-ROOTDIR = os.path.dirname(os.path.dirname(__file__))
-ETCDIR = os.path.join(ROOTDIR, 'etc')
HOSTNAME = 'hostname'
dev_man = dhcp.DeviceManager
rpc_api = dhcp_agent.DhcpPluginApi
DHCP_PLUGIN = '%s.%s' % (rpc_api.__module__, rpc_api.__name__)
-def etcdir(*p):
- return os.path.join(ETCDIR, *p)
-
-
fake_tenant_id = 'aaaaaaaa-aaaa-aaaa-aaaaaaaaaaaa'
fake_subnet1_allocation_pools = dhcp.DictModel(dict(id='', start='172.9.9.2',
end='172.9.9.254'))
with mock.patch.object(sys, 'argv') as sys_argv:
sys_argv.return_value = [
'dhcp', '--config-file',
- etcdir('neutron.conf.test')]
+ base.etcdir('neutron.conf.test')]
cfg.CONF.register_opts(dhcp_agent.DhcpAgent.OPTS)
config.register_interface_driver_opts_helper(cfg.CONF)
config.register_agent_state_opts_helper(cfg.CONF)
with mock.patch.object(sys, 'argv') as sys_argv:
with mock.patch(launcher_str) as launcher:
sys_argv.return_value = ['dhcp', '--config-file',
- etcdir('neutron.conf.test')]
+ base.etcdir('neutron.conf.test')]
dhcp_agent.main()
launcher.assert_has_calls(
[mock.call(), mock.call().launch_service(mock.ANY),
)
# point config file to: neutron/tests/etc/neutron.conf.test
- args = ['--config-file', test_api_v2.etcdir('neutron.conf.test')]
- config.parse(args=args)
+ self.config_parse()
self.setup_coreplugin(plugin)
# limitations under the License.
import contextlib
-import os
import mock
import webob.exc
DB_PLUGIN_KLASS = ('neutron.tests.unit.test_extension_security_group.'
'SecurityGroupTestPlugin')
-ROOTDIR = os.path.dirname(os.path.dirname(__file__))
-ETCDIR = os.path.join(ROOTDIR, 'etc')
-
-
-def etcdir(*p):
- return os.path.join(ETCDIR, *p)
class SecurityGroupTestExtensionManager(object):
# License for the specific language governing permissions and limitations
# under the License.
-import os
-
import mock
import routes
import webob
LOG = logging.getLogger(__name__)
-
-ROOTDIR = os.path.dirname(os.path.dirname(__file__))
-ETCDIR = os.path.join(ROOTDIR, 'etc')
-
-
-def etcdir(*p):
- return os.path.join(ETCDIR, *p)
-
extensions_path = ':'.join(neutron.tests.unit.extensions.__path__)
res.body = jsonutils.dumps(data)
return res
- base_app = webtest.TestApp(setup_base_app())
+ base_app = webtest.TestApp(setup_base_app(self))
response = base_app.put("/dummy_resources/1",
{'uneditable': "new_value"})
self.assertEqual(response.json['uneditable'], "original_value")
return ExtensionsTestApp(conf)
-def setup_base_app():
- config_file = 'neutron.conf.test'
- args = ['--config-file', etcdir(config_file)]
- config.parse(args=args)
+def setup_base_app(test):
+ base.BaseTestCase.config_parse()
app = config.load_paste_app('extensions_test_app')
return app
extensions.PluginAwareExtensionManager(
extensions_path,
{constants.CORE: FakePluginWithExtension()}))
- config_file = 'neutron.conf.test'
- args = ['--config-file', etcdir(config_file)]
- config.parse(args=args)
+ base.BaseTestCase.config_parse()
app = config.load_paste_app('extensions_test_app')
return extensions.ExtensionMiddleware(app, ext_mgr=extension_manager)
class TestBase(base.BaseTestCase):
def setUp(self):
super(TestBase, self).setUp()
- root = os.path.dirname(os.path.dirname(__file__))
- args = ['--config-file',
- os.path.join(root, 'etc', 'neutron.conf.test')]
self.conf = config.setup_conf()
self.conf.register_opts(base_config.core_opts)
self.conf.register_opts(dhcp.OPTS)
self.mock_mgr = instance.start()
self.conf.register_opt(cfg.BoolOpt('enable_isolated_metadata',
default=True))
- self.conf(args=args)
+ self.config_parse(self.conf)
self.conf.set_override('state_path', '')
self.conf.use_namespaces = True
# License for the specific language governing permissions and limitations
# under the License.
-import os
import types
import fixtures
from oslo.config import cfg
-from neutron.common import config
from neutron.manager import NeutronManager
from neutron.manager import validate_post_plugin_load
from neutron.manager import validate_pre_plugin_load
LOG = logging.getLogger(__name__)
DB_PLUGIN_KLASS = 'neutron.db.db_base_plugin_v2.NeutronDbPluginV2'
-ROOTDIR = os.path.dirname(os.path.dirname(__file__))
-ETCDIR = os.path.join(ROOTDIR, 'etc')
-
-
-def etcdir(*p):
- return os.path.join(ETCDIR, *p)
class MultiServiceCorePlugin(object):
def setUp(self):
super(NeutronManagerTestCase, self).setUp()
- args = ['--config-file', etcdir('neutron.conf.test')]
- # If test_config specifies some config-file, use it, as well
- config.parse(args=args)
+ self.config_parse()
self.setup_coreplugin()
self.useFixture(
fixtures.MonkeyPatch('neutron.manager.NeutronManager._instance'))
from neutron import quota
from neutron.tests import base
from neutron.tests.unit import test_api_v2
-from neutron.tests.unit import test_extensions
from neutron.tests.unit import testlib_api
TARGET_PLUGIN = ('neutron.plugins.linuxbridge.lb_neutron_plugin'
self.saved_attr_map[resource] = attrs.copy()
# Create the default configurations
- args = ['--config-file', test_extensions.etcdir('neutron.conf.test')]
- config.parse(args=args)
+ self.config_parse()
# Update the plugin and extensions path
self.setup_coreplugin(TARGET_PLUGIN)
)
# point config file to: neutron/tests/etc/neutron.conf.test
- args = ['--config-file', test_api_v2.etcdir('neutron.conf.test')]
- config.parse(args=args)
+ self.config_parse()
#just stubbing core plugin with LoadBalancer plugin
self.setup_coreplugin(plugin)
from neutron.api import extensions
from neutron.api.extensions import PluginAwareExtensionManager
from neutron.api.v2 import attributes
-from neutron.common import config
from neutron import context
from neutron.db import api as db_api
from neutron.db import db_base_plugin_v2
extensions.PluginAwareExtensionManager._instance = None
# Create the default configurations
- args = ['--config-file', test_api_v2.etcdir('neutron.conf.test')]
- config.parse(args=args)
+ self.config_parse()
# Update the plugin and extensions path
self.setup_coreplugin(plugin)
import mock
from oslo.config import cfg
-from neutron.common import config as q_config
from neutron.manager import NeutronManager
from neutron.openstack.common import uuidutils
from neutron.plugins.vmware.api_client import client
self.assertEqual('whatever', cluster.default_interface_name)
def test_load_plugin_with_full_options(self):
- q_config.parse(['--config-file', BASE_CONF_PATH,
- '--config-file', NSX_INI_FULL_PATH])
+ self.config_parse(args=['--config-file', BASE_CONF_PATH,
+ '--config-file', NSX_INI_FULL_PATH])
cfg.CONF.set_override('core_plugin', PLUGIN_NAME)
plugin = NeutronManager().get_plugin()
cluster = plugin.cluster
self._assert_extra_options(cluster)
def test_load_plugin_with_required_options_only(self):
- q_config.parse(['--config-file', BASE_CONF_PATH,
- '--config-file', NSX_INI_PATH])
+ self.config_parse(args=['--config-file', BASE_CONF_PATH,
+ '--config-file', NSX_INI_PATH])
cfg.CONF.set_override('core_plugin', PLUGIN_NAME)
plugin = NeutronManager().get_plugin()
self._assert_required_options(plugin.cluster)
self.assertEqual('breth0', cfg.CONF.default_interface_name)
def test_load_api_extensions(self):
- q_config.parse(['--config-file', BASE_CONF_PATH,
- '--config-file', NSX_INI_FULL_PATH])
+ self.config_parse(args=['--config-file', BASE_CONF_PATH,
+ '--config-file', NSX_INI_FULL_PATH])
cfg.CONF.set_override('core_plugin', PLUGIN_NAME)
# Load the configuration, and initialize the plugin
NeutronManager().get_plugin()
self.assertIn('extensions', cfg.CONF.api_extensions_path)
def test_agentless_extensions(self):
- q_config.parse(['--config-file', BASE_CONF_PATH,
- '--config-file', NSX_INI_AGENTLESS_PATH])
+ self.config_parse(args=['--config-file', BASE_CONF_PATH,
+ '--config-file', NSX_INI_AGENTLESS_PATH])
cfg.CONF.set_override('core_plugin', PLUGIN_NAME)
self.assertEqual(config.AgentModes.AGENTLESS,
cfg.CONF.NSX.agent_mode)
plugin.supported_extension_aliases)
def test_agentless_extensions_version_fail(self):
- q_config.parse(['--config-file', BASE_CONF_PATH,
- '--config-file', NSX_INI_AGENTLESS_PATH])
+ self.config_parse(args=['--config-file', BASE_CONF_PATH,
+ '--config-file', NSX_INI_AGENTLESS_PATH])
cfg.CONF.set_override('core_plugin', PLUGIN_NAME)
self.assertEqual(config.AgentModes.AGENTLESS,
cfg.CONF.NSX.agent_mode)
self.assertRaises(exceptions.NsxPluginException, NeutronManager)
def test_agentless_extensions_unmet_deps_fail(self):
- q_config.parse(['--config-file', BASE_CONF_PATH,
- '--config-file', NSX_INI_AGENTLESS_PATH])
+ self.config_parse(args=['--config-file', BASE_CONF_PATH,
+ '--config-file', NSX_INI_AGENTLESS_PATH])
cfg.CONF.set_override('core_plugin', PLUGIN_NAME)
self.assertEqual(config.AgentModes.AGENTLESS,
cfg.CONF.NSX.agent_mode)
NeutronManager)
def test_agent_extensions(self):
- q_config.parse(['--config-file', BASE_CONF_PATH,
- '--config-file', NSX_INI_FULL_PATH])
+ self.config_parse(args=['--config-file', BASE_CONF_PATH,
+ '--config-file', NSX_INI_FULL_PATH])
cfg.CONF.set_override('core_plugin', PLUGIN_NAME)
self.assertEqual(config.AgentModes.AGENT,
cfg.CONF.NSX.agent_mode)
self.assertEqual(cluster.default_tz_uuid, 'fake_tz_uuid')
def test_load_plugin_with_deprecated_options(self):
- q_config.parse(['--config-file', BASE_CONF_PATH,
- '--config-file', NVP_INI_DEPR_PATH])
+ self.config_parse(args=['--config-file', BASE_CONF_PATH,
+ '--config-file', NVP_INI_DEPR_PATH])
cfg.CONF.set_override('core_plugin', PLUGIN_NAME)
plugin = NeutronManager().get_plugin()
cluster = plugin.cluster
from oslo.config import cfg
from neutron.api.v2 import attributes as attr
-from neutron.common import config
from neutron.common import constants
from neutron.common import exceptions as n_exc
from neutron import context
# and setup needed config variables
args = ['--config-file', get_fake_conf('neutron.conf.test'),
'--config-file', get_fake_conf('nsx.ini.test')]
- config.parse(args=args)
+ self.config_parse(args=args)
cfg.CONF.set_override('allow_overlapping_ips', True)
self._plugin = plugin.NsxPlugin()
# Mock neutron manager plugin load functions to speed up tests
import mock
import webob.exc
-from neutron.common import config as n_config
from neutron import context
from neutron.db.firewall import firewall_db
from neutron.openstack.common import uuidutils
def setUp(self):
- n_config.parse(['--config-file', VCNS_CONFIG_FILE])
+ self.config_parse(args=['--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)
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
def setUp(self):
- n_config.parse(['--config-file', VCNS_CONFIG_FILE])
+ self.config_parse(args=['--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)
from eventlet import greenthread
import mock
-from neutron.common import config as n_config
from neutron.plugins.vmware.vshield.common import constants as vcns_const
from neutron.plugins.vmware.vshield.common.constants import RouterStatus
from neutron.plugins.vmware.vshield.tasks.constants import TaskState
def setUp(self):
super(VcnsDriverTestCase, self).setUp()
- n_config.parse(['--config-file', VCNS_CONFIG_FILE])
+ self.config_parse(args=['--config-file', VCNS_CONFIG_FILE])
self.fc = fake_vcns.FakeVcns()
self.mock_vcns = mock.patch(VCNS_NAME, autospec=True)
Code Review / openstack-build / neutron-build.git / commitdiff