# vim: tabstop=4 shiftwidth=4 softtabstop=4
-# Copyright 2012 OpenStack Foundation.
+# Copyright 2013 OpenStack Foundation.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
+
+import os
+
+import neutron.plugins.nicira.api_client.client_eventlet as client
+import neutron.plugins.nicira.NeutronPlugin as plugin
+import neutron.plugins.nicira.NvpApiClient as nvpapi
+
+nvp_plugin = plugin.NvpPluginV2
+api_helper = nvpapi.NVPApiHelper
+nvp_client = client.NvpApiClientEventlet
+
+STUBS_PATH = os.path.join(os.path.dirname(__file__), 'etc')
+NVPEXT_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)),
+ "../../plugins/nicira/extensions")
+NVPAPI_NAME = '%s.%s' % (api_helper.__module__, api_helper.__name__)
+PLUGIN_NAME = '%s.%s' % (nvp_plugin.__module__, nvp_plugin.__name__)
+CLIENT_NAME = '%s.%s' % (nvp_client.__module__, nvp_client.__name__)
+
+
+def get_fake_conf(filename):
+ return os.path.join(STUBS_PATH, filename)
# under the License.
import mock
-import os
from neutron.common.test_lib import test_config
-import neutron.plugins.nicira as nvp_plugin
from neutron.tests.unit.nicira import fake_nvpapiclient
+from neutron.tests.unit.nicira import get_fake_conf
+from neutron.tests.unit.nicira import NVPAPI_NAME
+from neutron.tests.unit.nicira import PLUGIN_NAME
+from neutron.tests.unit.nicira import STUBS_PATH
from neutron.tests.unit.openvswitch import test_agent_scheduler as test_base
-NVP_MODULE_PATH = nvp_plugin.__name__
-NVP_INI_CONFIG_PATH = os.path.join(os.path.dirname(__file__),
- 'etc/nvp.ini.full.test')
-NVP_STUBS_PATH = os.path.join(os.path.dirname(__file__), 'etc')
-
-PLUGIN_NAME = '%s.NeutronPlugin.NvpPluginV2' % nvp_plugin.__name__
-
-
class NVPDhcpAgentNotifierTestCase(test_base.OvsDhcpAgentNotifierTestCase):
plugin_str = PLUGIN_NAME
def setUp(self):
test_config['plugin_name_v2'] = PLUGIN_NAME
- test_config['config_files'] = [NVP_INI_CONFIG_PATH]
+ test_config['config_files'] = [get_fake_conf('nvp.ini.full.test')]
# mock nvp api client
- self.fc = fake_nvpapiclient.FakeClient(NVP_STUBS_PATH)
- self.mock_nvpapi = mock.patch('%s.NvpApiClient.NVPApiHelper'
- % NVP_MODULE_PATH, autospec=True)
+ self.fc = fake_nvpapiclient.FakeClient(STUBS_PATH)
+ self.mock_nvpapi = mock.patch(NVPAPI_NAME, autospec=True)
instance = self.mock_nvpapi.start()
def _fake_request(*args, **kwargs):
import contextlib
import mock
-import os
from oslo.config import cfg
from neutron.common.test_lib import test_config
from neutron import context
from neutron.extensions import agent
-from neutron.openstack.common import log as logging
-import neutron.plugins.nicira as nvp_plugin
from neutron.plugins.nicira.NvpApiClient import NVPVersion
from neutron.tests.unit.nicira import fake_nvpapiclient
+from neutron.tests.unit.nicira import get_fake_conf
+from neutron.tests.unit.nicira import NVPAPI_NAME
+from neutron.tests.unit.nicira import NVPEXT_PATH
+from neutron.tests.unit.nicira import PLUGIN_NAME
+from neutron.tests.unit.nicira import STUBS_PATH
from neutron.tests.unit import test_db_plugin
-LOG = logging.getLogger(__name__)
-NVP_MODULE_PATH = nvp_plugin.__name__
-NVP_FAKE_RESPS_PATH = os.path.join(os.path.dirname(__file__), 'etc')
-NVP_INI_CONFIG_PATH = os.path.join(os.path.dirname(__file__),
- 'etc/nvp.ini.full.test')
-NVP_EXTENSIONS_PATH = os.path.join(os.path.dirname(__file__),
- '../../../plugins/nicira/extensions')
-
-
class MacLearningExtensionManager(object):
def get_resources(self):
def setUp(self):
self.adminContext = context.get_admin_context()
- test_config['config_files'] = [NVP_INI_CONFIG_PATH]
- test_config['plugin_name_v2'] = (
- 'neutron.plugins.nicira.NeutronPlugin.NvpPluginV2')
- cfg.CONF.set_override('api_extensions_path',
- NVP_EXTENSIONS_PATH)
+ test_config['config_files'] = [get_fake_conf('nvp.ini.full.test')]
+ test_config['plugin_name_v2'] = PLUGIN_NAME
+ cfg.CONF.set_override('api_extensions_path', NVPEXT_PATH)
# Save the original RESOURCE_ATTRIBUTE_MAP
self.saved_attr_map = {}
for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems():
ext_mgr = MacLearningExtensionManager()
test_config['extension_manager'] = ext_mgr
# mock nvp api client
- self.fc = fake_nvpapiclient.FakeClient(NVP_FAKE_RESPS_PATH)
- self.mock_nvpapi = mock.patch('%s.NvpApiClient.NVPApiHelper'
- % NVP_MODULE_PATH, autospec=True)
+ self.fc = fake_nvpapiclient.FakeClient(STUBS_PATH)
+ self.mock_nvpapi = mock.patch(NVPAPI_NAME, autospec=True)
instance = self.mock_nvpapi.start()
def _fake_request(*args, **kwargs):
# limitations under the License.
import contextlib
-import os
import mock
import netaddr
from neutron.extensions import securitygroup as secgrp
from neutron import manager
from neutron.openstack.common import uuidutils
-import neutron.plugins.nicira as nvp_plugin
from neutron.plugins.nicira.dbexts import nicira_qos_db as qos_db
from neutron.plugins.nicira.extensions import nvp_networkgw
from neutron.plugins.nicira.extensions import nvp_qos as ext_qos
from neutron.plugins.nicira import nvplib
from neutron.tests.unit import _test_extension_portbindings as test_bindings
from neutron.tests.unit.nicira import fake_nvpapiclient
+from neutron.tests.unit.nicira import get_fake_conf
+from neutron.tests.unit.nicira import NVPAPI_NAME
+from neutron.tests.unit.nicira import NVPEXT_PATH
+from neutron.tests.unit.nicira import PLUGIN_NAME
+from neutron.tests.unit.nicira import STUBS_PATH
import neutron.tests.unit.nicira.test_networkgw as test_l2_gw
import neutron.tests.unit.test_db_plugin as test_plugin
import neutron.tests.unit.test_extension_ext_gw_mode as test_ext_gw_mode
import neutron.tests.unit.test_l3_plugin as test_l3_plugin
from neutron.tests.unit import testlib_api
-NICIRA_PKG_PATH = nvp_plugin.__name__
-NICIRA_EXT_PATH = "../../plugins/nicira/extensions"
-
class NiciraPluginV2TestCase(test_plugin.NeutronDbPluginV2TestCase):
- _plugin_name = ('%s.NeutronPlugin.NvpPluginV2' % NICIRA_PKG_PATH)
-
def _create_network(self, fmt, name, admin_state_up,
arg_list=None, providernet_args=None, **kwargs):
data = {'network': {'name': name,
return network_req.get_response(self.api)
def setUp(self):
- etc_path = os.path.join(os.path.dirname(__file__), 'etc')
- test_lib.test_config['config_files'] = [os.path.join(etc_path,
- 'nvp.ini.test')]
+ test_lib.test_config['config_files'] = [get_fake_conf('nvp.ini.test')]
# mock nvp api client
- self.fc = fake_nvpapiclient.FakeClient(etc_path)
- self.mock_nvpapi = mock.patch('%s.NvpApiClient.NVPApiHelper'
- % NICIRA_PKG_PATH, autospec=True)
+ self.fc = fake_nvpapiclient.FakeClient(STUBS_PATH)
+ self.mock_nvpapi = mock.patch(NVPAPI_NAME, autospec=True)
instance = self.mock_nvpapi.start()
def _fake_request(*args, **kwargs):
# Emulate tests against NVP 2.x
instance.return_value.get_nvp_version.return_value = NVPVersion("2.9")
instance.return_value.request.side_effect = _fake_request
- super(NiciraPluginV2TestCase, self).setUp(self._plugin_name)
+ super(NiciraPluginV2TestCase, self).setUp(PLUGIN_NAME)
cfg.CONF.set_override('metadata_mode', None, 'NVP')
self.addCleanup(self.fc.reset_all)
self.addCleanup(self.mock_nvpapi.stop)
class NiciraPortSecurityTestCase(psec.PortSecurityDBTestCase):
- _plugin_name = ('%s.NeutronPlugin.NvpPluginV2' % NICIRA_PKG_PATH)
-
def setUp(self):
- etc_path = os.path.join(os.path.dirname(__file__), 'etc')
- test_lib.test_config['config_files'] = [os.path.join(etc_path,
- 'nvp.ini.test')]
+ test_lib.test_config['config_files'] = [get_fake_conf('nvp.ini.test')]
# mock nvp api client
- fc = fake_nvpapiclient.FakeClient(etc_path)
- self.mock_nvpapi = mock.patch('%s.NvpApiClient.NVPApiHelper'
- % NICIRA_PKG_PATH, autospec=True)
+ fc = fake_nvpapiclient.FakeClient(STUBS_PATH)
+ self.mock_nvpapi = mock.patch(NVPAPI_NAME, autospec=True)
instance = self.mock_nvpapi.start()
instance.return_value.login.return_value = "the_cookie"
return fc.fake_request(*args, **kwargs)
instance.return_value.request.side_effect = _fake_request
- super(NiciraPortSecurityTestCase, self).setUp(self._plugin_name)
+ super(NiciraPortSecurityTestCase, self).setUp(PLUGIN_NAME)
self.addCleanup(self.mock_nvpapi.stop)
class NiciraSecurityGroupsTestCase(ext_sg.SecurityGroupDBTestCase):
- _plugin_name = ('%s.NeutronPlugin.NvpPluginV2' % NICIRA_PKG_PATH)
-
def setUp(self):
- etc_path = os.path.join(os.path.dirname(__file__), 'etc')
- test_lib.test_config['config_files'] = [os.path.join(etc_path,
- 'nvp.ini.test')]
+ test_lib.test_config['config_files'] = [get_fake_conf('nvp.ini.test')]
# mock nvp api client
- fc = fake_nvpapiclient.FakeClient(etc_path)
- self.mock_nvpapi = mock.patch('%s.NvpApiClient.NVPApiHelper'
- % NICIRA_PKG_PATH, autospec=True)
+ fc = fake_nvpapiclient.FakeClient(STUBS_PATH)
+ self.mock_nvpapi = mock.patch(NVPAPI_NAME, autospec=True)
instance = self.mock_nvpapi.start()
instance.return_value.login.return_value = "the_cookie"
return fc.fake_request(*args, **kwargs)
instance.return_value.request.side_effect = _fake_request
- super(NiciraSecurityGroupsTestCase, self).setUp(self._plugin_name)
+ super(NiciraSecurityGroupsTestCase, self).setUp(PLUGIN_NAME)
def tearDown(self):
super(NiciraSecurityGroupsTestCase, self).tearDown()
class TestNiciraQoSQueue(NiciraPluginV2TestCase):
def setUp(self, plugin=None):
- ext_path = os.path.join(os.path.dirname(os.path.dirname(__file__)),
- NICIRA_EXT_PATH)
- cfg.CONF.set_override('api_extensions_path', ext_path)
+ cfg.CONF.set_override('api_extensions_path', NVPEXT_PATH)
super(TestNiciraQoSQueue, self).setUp()
ext_mgr = NvpQoSTestExtensionManager()
self.ext_api = test_extensions.setup_extensions_middleware(ext_mgr)
NiciraPluginV2TestCase):
def setUp(self):
- ext_path = os.path.join(os.path.dirname(os.path.dirname(__file__)),
- NICIRA_EXT_PATH)
- cfg.CONF.set_override('api_extensions_path', ext_path)
+ cfg.CONF.set_override('api_extensions_path', NVPEXT_PATH)
super(TestNiciraNetworkGateway, self).setUp()
def test_create_network_gateway_name_exceeds_40_chars(self):
from mock import Mock
from mock import patch
-from neutron.plugins.nicira.api_client import (
- client_eventlet as nace,
- request_eventlet as nare,
-)
+from neutron.plugins.nicira.api_client import client_eventlet as nace
+from neutron.plugins.nicira.api_client import request_eventlet as nare
from neutron.tests import base
+from neutron.tests.unit.nicira import CLIENT_NAME
logging.basicConfig(level=logging.DEBUG)
self.assertTrue(retval is None)
def test_redirect_params_setup_https_with_cooki(self):
- with patch('neutron.plugins.nicira.api_client.'
- 'client_eventlet.NvpApiClientEventlet') as mock:
+ with patch(CLIENT_NAME) as mock:
api_client = mock.return_value
self.req._api_client = api_client
myconn = Mock()
self.assertTrue(api_client.acquire_redirect_connection.called)
def test_redirect_params_setup_htttps_and_query(self):
- with patch('neutron.plugins.nicira.api_client.'
- 'client_eventlet.NvpApiClientEventlet') as mock:
+ with patch(CLIENT_NAME) as mock:
api_client = mock.return_value
self.req._api_client = api_client
myconn = Mock()
self.assertTrue(api_client.acquire_redirect_connection.called)
def test_redirect_params_setup_https_connection_no_cookie(self):
- with patch('neutron.plugins.nicira.api_client.'
- 'client_eventlet.NvpApiClientEventlet') as mock:
+ with patch(CLIENT_NAME) as mock:
api_client = mock.return_value
self.req._api_client = api_client
myconn = Mock()
self.assertTrue(api_client.acquire_redirect_connection.called)
def test_redirect_params_setup_https_and_query_no_cookie(self):
- with patch('neutron.plugins.nicira.api_client.'
- 'client_eventlet.NvpApiClientEventlet') as mock:
+ with patch(CLIENT_NAME) as mock:
api_client = mock.return_value
self.req._api_client = api_client
myconn = Mock()
self.assertTrue(api_client.acquire_redirect_connection.called)
def test_redirect_params_path_only_with_query(self):
- with patch('neutron.plugins.nicira.api_client.'
- 'client_eventlet.NvpApiClientEventlet') as mock:
+ with patch(CLIENT_NAME) as mock:
api_client = mock.return_value
api_client.wait_for_login.return_value = None
api_client.auth_cookie = None
# @author: Salvatore Orlando, VMware
import mock
-import os
from neutron.common import constants
from neutron.common import exceptions
-import neutron.plugins.nicira as nvp_plugin
from neutron.plugins.nicira.common import config # noqa
from neutron.plugins.nicira.common import exceptions as nvp_exc
from neutron.plugins.nicira import nvp_cluster
from neutron.plugins.nicira import nvplib
from neutron.tests import base
from neutron.tests.unit.nicira import fake_nvpapiclient
+from neutron.tests.unit.nicira import NVPAPI_NAME
+from neutron.tests.unit.nicira import STUBS_PATH
from neutron.tests.unit import test_api_v2
-NICIRA_PKG_PATH = nvp_plugin.__name__
+
_uuid = test_api_v2._uuid
def setUp(self):
# mock nvp api client
- etc_path = os.path.join(os.path.dirname(__file__), 'etc')
- self.fc = fake_nvpapiclient.FakeClient(etc_path)
- self.mock_nvpapi = mock.patch('%s.NvpApiClient.NVPApiHelper'
- % NICIRA_PKG_PATH, autospec=True)
+ self.fc = fake_nvpapiclient.FakeClient(STUBS_PATH)
+ self.mock_nvpapi = mock.patch(NVPAPI_NAME, autospec=True)
instance = self.mock_nvpapi.start()
instance.return_value.login.return_value = "the_cookie"
fake_version = getattr(self, 'fake_version', "2.9")
def setUp(self):
# mock nvp api client
- etc_path = os.path.join(os.path.dirname(__file__), 'etc')
- self.fc = fake_nvpapiclient.FakeClient(etc_path)
- self.mock_nvpapi = mock.patch('%s.NvpApiClient.NVPApiHelper'
- % NICIRA_PKG_PATH, autospec=True)
+ self.fc = fake_nvpapiclient.FakeClient(STUBS_PATH)
+ self.mock_nvpapi = mock.patch(NVPAPI_NAME, autospec=True)
instance = self.mock_nvpapi.start()
instance.return_value.login.return_value = "the_cookie"
# Choose 2.9, but the version is irrelevant for the aim of
#
import fixtures
-import os
import testtools
from oslo.config import cfg
from neutron.plugins.nicira.common import config # noqa
from neutron.plugins.nicira.common import exceptions
from neutron.plugins.nicira import nvp_cluster
+from neutron.tests.unit.nicira import get_fake_conf
+from neutron.tests.unit.nicira import PLUGIN_NAME
-BASE_CONF_PATH = os.path.join(os.path.dirname(__file__),
- '../../etc/neutron.conf.test')
-NVP_BASE_CONF_PATH = os.path.join(os.path.dirname(__file__),
- 'etc/neutron.conf.test')
-NVP_INI_PATH = os.path.join(os.path.dirname(__file__),
- 'etc/nvp.ini.basic.test')
-NVP_INI_FULL_PATH = os.path.join(os.path.dirname(__file__),
- 'etc/nvp.ini.full.test')
-NVP_INI_DEPR_PATH = os.path.join(os.path.dirname(__file__),
- 'etc/nvp.ini.grizzly.test')
-NVP_PLUGIN_PATH = ('neutron.plugins.nicira.nicira_nvp_plugin.'
- 'NeutronPlugin.NvpPluginV2')
+BASE_CONF_PATH = get_fake_conf('neutron.conf.test')
+NVP_BASE_CONF_PATH = get_fake_conf('neutron.conf.test')
+NVP_INI_PATH = get_fake_conf('nvp.ini.basic.test')
+NVP_INI_FULL_PATH = get_fake_conf('nvp.ini.full.test')
+NVP_INI_DEPR_PATH = get_fake_conf('nvp.ini.grizzly.test')
class NVPClusterTest(testtools.TestCase):
def test_load_plugin_with_full_options(self):
q_config.parse(['--config-file', BASE_CONF_PATH,
'--config-file', NVP_INI_FULL_PATH])
- cfg.CONF.set_override('core_plugin', NVP_PLUGIN_PATH)
+ cfg.CONF.set_override('core_plugin', PLUGIN_NAME)
plugin = NeutronManager().get_plugin()
cluster = plugin.cluster
self._assert_required_options(cluster)
def test_load_plugin_with_required_options_only(self):
q_config.parse(['--config-file', BASE_CONF_PATH,
'--config-file', NVP_INI_PATH])
- cfg.CONF.set_override('core_plugin', NVP_PLUGIN_PATH)
+ cfg.CONF.set_override('core_plugin', PLUGIN_NAME)
plugin = NeutronManager().get_plugin()
self._assert_required_options(plugin.cluster)
def test_load_api_extensions(self):
q_config.parse(['--config-file', NVP_BASE_CONF_PATH,
'--config-file', NVP_INI_FULL_PATH])
- cfg.CONF.set_override('core_plugin', NVP_PLUGIN_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_load_plugin_with_deprecated_options(self):
q_config.parse(['--config-file', BASE_CONF_PATH,
'--config-file', NVP_INI_DEPR_PATH])
- cfg.CONF.set_override('core_plugin', NVP_PLUGIN_PATH)
+ cfg.CONF.set_override('core_plugin', PLUGIN_NAME)
plugin = NeutronManager().get_plugin()
cluster = plugin.cluster
self._assert_required_options(cluster)