x
/
vmware-nsx
Code Issues Proposed changes
Browse Source

Revert "Temporarily disable admin utils tests" 

This reverts commit 09fbad47d8.

Reason for revert: Re-Enable the tests

Change-Id: I48e9e7f7e41084c8def0d46804a05dc17bdb5ade
changes/43/765743/7
Adit Sarfaty 2 years ago
parent
eeac28dd80
commit
2a1991a3a6
4 changed files with 416 additions and 413 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 2
      .coveragerc
  2. 7
      tox.ini
  3. 3
      vmware_nsx/shell/admin/plugins/nsxp/resources/migration.py
  4. 817
      vmware_nsx/tests/unit/shell/test_admin_utils.py

2
.coveragerc
Unescape View File

@ -1,7 +1,7 @@
[run]
branch = True
source = vmware_nsx
omit = vmware_nsx/tests/*,vmware_nsx/*dvs*,vmware_nsx/api_replay/*,vmware_nsx/dhcp_meta/*,vmware_nsx/nsxlib/*,vmware_nsx/*lsn*,vmware_nsx/*tv*,vmware_nsx/api_client/*,vmware_nsx/common/profile*,vmware_nsx/shell/nsx_instance_if_migrate*,vmware_nsx/plugins/nsx_v/vshield/vcns.*,vmware_nsx/db/migration/alembic_migrations/*,vmware_nsx/shell/admin/plugins*,vmware_nsx/shell/admin/plugins/nsxv3/resources/migration*
omit = vmware_nsx/tests/*,vmware_nsx/*dvs*,vmware_nsx/api_replay/*,vmware_nsx/dhcp_meta/*,vmware_nsx/nsxlib/*,vmware_nsx/*lsn*,vmware_nsx/*tv*,vmware_nsx/api_client/*,vmware_nsx/common/profile*,vmware_nsx/shell/nsx_instance_if_migrate*,vmware_nsx/plugins/nsx_v/vshield/vcns.*,vmware_nsx/db/migration/alembic_migrations/*,vmware_nsx/shell/admin/plugins/nsxv3/resources/migration*
[report]
ignore_errors = True

7
tox.ini
Unescape View File

@ -22,6 +22,7 @@ commands =
[testenv:common]
# Fake job to define environment variables shared between dsvm/non-dsvm jobs
setenv = OS_TEST_TIMEOUT=180
EDITOR=vi
commands = false
[testenv:dev]
@ -69,14 +70,17 @@ commands = sphinx-build -a -E -W -d releasenotes/build/doctrees -b html releasen
[testenv:py36]
basepython = python3.6
setenv = OS_FAIL_ON_MISSING_DEPS=1
EDITOR=vi
[testenv:py37]
basepython = python3.7
setenv = OS_FAIL_ON_MISSING_DEPS=1
EDITOR=vi
[testenv:py38]
basepython = python3.8
setenv = OS_FAIL_ON_MISSING_DEPS=1
EDITOR=vi
[testenv:py3-dev]
commands =
@ -85,6 +89,7 @@ commands =
stestr run {posargs}
whitelist_externals =
stestr
setenv = EDITOR=vi
[testenv:py36-dev]
basepython = python3.6
@ -132,6 +137,7 @@ envdir = {toxworkdir}/shared
setenv = {[testenv]setenv}
{[testenv:common]setenv}
PYTHON=coverage run --source vmware_nsx --parallel-mode
EDITOR=vi
commands =
{[testenv:dev]commands}
coverage erase
@ -194,6 +200,7 @@ deps =
-r{toxinidir}/test-requirements.txt
-r{toxinidir}/doc/requirements.txt
-r{toxinidir}/requirements.txt
setenv = EDITOR=vi
[testenv:lower-constraints-dev]
commands =

3
vmware_nsx/shell/admin/plugins/nsxp/resources/migration.py
Unescape View File

@ -23,7 +23,8 @@ from vmware_nsx.shell import resources as shell
@admin_utils.output_header
def cleanup_db_mappings(resource, event, trigger, **kwargs):
"""Delete all entries from nsx-t mapping tables in DB"""
return migration.cleanup_db_mappings(resource, event, trigger, **kwargs)
return migration. MP2Policy_cleanup_db_mappings(
resource, event, trigger, **kwargs)
registry.subscribe(cleanup_db_mappings,

817
vmware_nsx/tests/unit/shell/test_admin_utils.py
Unescape View File

@ -13,419 +13,414 @@
# License for the specific language governing permissions and limitations
# under the License.
# TODO(asarfaty) - Re-enable those tests when cmd2 problem is solved
# import abc
import abc
# from unittest import mock
from unittest import mock
# from oslo_config import cfg
# from oslo_log import _options
from oslo_config import cfg
from oslo_log import _options
from oslo_log import log as logging
# from oslo_utils import uuidutils
# from neutron.common import config as neutron_config
# from neutron.db import servicetype_db # noqa
# from neutron.quota import resource_registry
# from neutron.tests import base
# from neutron_lib.callbacks import registry
# from neutron_lib.plugins import constants
# from vmware_nsx._i18n import _
# from vmware_nsx.common import config # noqa
# from vmware_nsx.db import nsxv_db
# from vmware_nsx.dvs import dvs_utils
# from vmware_nsx.shell.admin.plugins.nsxp.resources import utils as nsxp_utils
# from vmware_nsx.shell.admin.plugins.nsxv.resources import migration
# from vmware_nsx.shell.admin.plugins.nsxv.resources import utils as nsxv_utils
# from vmware_nsx.shell.admin.plugins.nsxv3.resources import utils as \
# nsxv3_utils
# from vmware_nsx.shell import resources
# from vmware_nsx.tests import unit as vmware
# from vmware_nsx.tests.unit.nsx_p import test_plugin as test_p_plugin
# from vmware_nsx.tests.unit.nsx_v import test_plugin as test_v_plugin
# from vmware_nsx.tests.unit.nsx_v3 import test_plugin as test_v3_plugin
# from vmware_nsxlib.v3 import client as v3_client
# from vmware_nsxlib.v3 import core_resources
# from vmware_nsxlib.v3 import resources as nsx_v3_resources
# from vmware_nsxlib.v3 import security as nsx_v3_security
from oslo_utils import uuidutils
from neutron.common import config as neutron_config
from neutron.db import servicetype_db # noqa
from neutron.quota import resource_registry
from neutron.tests import base
from neutron_lib.callbacks import registry
from neutron_lib.plugins import constants
from vmware_nsx._i18n import _
from vmware_nsx.common import config # noqa
from vmware_nsx.db import nsxv_db
from vmware_nsx.dvs import dvs_utils
from vmware_nsx.shell.admin.plugins.nsxp.resources import utils as nsxp_utils
from vmware_nsx.shell.admin.plugins.nsxv.resources import migration
from vmware_nsx.shell.admin.plugins.nsxv.resources import utils as nsxv_utils
from vmware_nsx.shell.admin.plugins.nsxv3.resources import utils as nsxv3_utils
from vmware_nsx.shell import resources
from vmware_nsx.tests import unit as vmware
from vmware_nsx.tests.unit.nsx_p import test_plugin as test_p_plugin
from vmware_nsx.tests.unit.nsx_v import test_plugin as test_v_plugin
from vmware_nsx.tests.unit.nsx_v3 import test_plugin as test_v3_plugin
from vmware_nsxlib.v3 import client as v3_client
from vmware_nsxlib.v3 import core_resources
from vmware_nsxlib.v3 import resources as nsx_v3_resources
from vmware_nsxlib.v3 import security as nsx_v3_security
LOG = logging.getLogger(__name__)
# NSX_INI_PATH = vmware.get_fake_conf('nsx.ini.test')
# BASE_CONF_PATH = vmware.get_fake_conf('neutron.conf.test')
# class AbstractTestAdminUtils(base.BaseTestCase, metaclass=abc.ABCMeta):
# def setUp(self):
# cfg.CONF.unregister_opts(_options.common_cli_opts)
# cfg.CONF.register_cli_opts(resources.cli_opts)
# super(AbstractTestAdminUtils, self).setUp()
# # remove resource registration conflicts
# resource_registry.unregister_all_resources()
# self.edgeapi = nsxv_utils.NeutronDbClient()
# # Init the neutron config
# neutron_config.init(args=['--config-file', BASE_CONF_PATH,
# '--config-file', NSX_INI_PATH])
# self._init_mock_plugin()
# self._init_resource_plugin()
# self.addCleanup(resource_registry.unregister_all_resources)
# def _init_mock_plugin(self):
# mock_query = mock.patch(
# "vmware_nsx.shell.admin.plugins.common.utils.query_yes_no")
# mock_query.start()
# @abc.abstractmethod
# def _get_plugin_name(self):
# pass
# def _init_resource_plugin(self):
# plugin_name = self._get_plugin_name()
# resources.init_resource_plugin(
# plugin_name,
# resources.get_plugin_dir(plugin_name))
# def _test_resource(self, res_name, op, **kwargs):
# errors = self._test_resource_with_errors(res_name, op, **kwargs)
# if len(errors) > 0:
# msg = (_("admin util %(res)s/%(op)s failed with message: "
# "%(err)s") % {'res': res_name,
# 'op': op,
# 'err': errors[0]})
# self.fail(msg=msg)
# def _test_resource_with_errors(self, res_name, op, **kwargs):
# # Must call the internal notify_loop in order to get the errors
# return registry._get_callback_manager()._notify_loop(
# res_name, op, 'nsxadmin', **kwargs)
# def _test_resources(self, res_dict):
# for res in res_dict.keys():
# res_name = res_dict[res].name
# for op in res_dict[res].supported_ops:
# self._test_resource(res_name, op)
# def _test_resources_with_args(self, res_dict, func_args):
# for res in res_dict.keys():
# res_name = res_dict[res].name
# for op in res_dict[res].supported_ops:
# args = {'property': func_args}
# self._test_resource(res_name, op, **args)
# def _create_router(self):
# tenant_id = uuidutils.generate_uuid()
# data = {'router': {'tenant_id': tenant_id}}
# data['router']['name'] = 'dummy'
# data['router']['admin_state_up'] = True
# return self._plugin.create_router(self.edgeapi.context, data)
# class TestNsxvAdminUtils(AbstractTestAdminUtils,
# test_v_plugin.NsxVPluginV2TestCase):
# def _get_plugin_name(self):
# return 'nsxv'
# def _init_mock_plugin(self, *mocks):
# super(TestNsxvAdminUtils, self)._init_mock_plugin()
# # support the dvs manager:
# mock.patch.object(dvs_utils, 'dvs_create_session').start()
# # override metadata get-object
# dummy_lb = {
# 'enabled': True,
# 'enableServiceInsertion': True,
# 'accelerationEnabled': True,
# 'virtualServer': [],
# 'applicationProfile': [],
# 'pool': [],
# 'applicationRule': []
# }
# mock.patch('vmware_nsx.plugins.nsx_v.vshield.nsxv_edge_cfg_obj.'
# 'NsxvEdgeCfgObj.get_object',
# return_value=dummy_lb).start()
# # Tests shouldn't wait for dummy spawn jobs to finish
# mock.patch('vmware_nsx.shell.admin.plugins.nsxv.resources.utils.'
# 'NsxVPluginWrapper.count_spawn_jobs',
# return_value=0).start()
# self._plugin = nsxv_utils.NsxVPluginWrapper()
# def get_plugin_mock(alias=constants.CORE):
# if alias in (constants.CORE, constants.L3):
# return self._plugin
# mock.patch("neutron_lib.plugins.directory.get_plugin",
# side_effect=get_plugin_mock).start()
# # Create a router to make sure we have deployed an edge
# self.router = self._create_router()
# self.dist_router = self._create_router(dist=True)
# self.network = self._create_net()
# def tearDown(self):
# if self.router and self.router.get('id'):
# self._plugin.delete_router(
# self.edgeapi.context, self.router['id'])
# if self.dist_router and self.dist_router.get('id'):
# self._plugin.delete_router(
# self.edgeapi.context, self.dist_router['id'])
# if self.network and self.network.get('id'):
# self._plugin.delete_network(
# self.edgeapi.context, self.network['id'])
# super(TestNsxvAdminUtils, self).tearDown()
# def test_nsxv_resources(self):
# self._test_resources(resources.nsxv_resources)
# def _test_edge_nsx_update(self, edge_id, params):
# args = {'property': ["edge-id=%s" % edge_id]}
# args['property'].extend(params)
# self._test_resource('edges', 'nsx-update', **args)
# def _create_router(self, dist=False):
# # Create an exclusive router (with an edge)
# tenant_id = uuidutils.generate_uuid()
# data = {'router': {'tenant_id': tenant_id}}
# data['router']['name'] = 'dummy'
# data['router']['admin_state_up'] = True
# if dist:
# data['router']['distributes'] = True
# else:
# data['router']['router_type'] = 'exclusive'
# return self._plugin.create_router(self.edgeapi.context, data)
# def _create_net(self):
# tenant_id = uuidutils.generate_uuid()
# data = {'network': {'tenant_id': tenant_id,
# 'name': 'dummy',
# 'admin_state_up': True,
# 'shared': False}}
# net = self._plugin.create_network(self.edgeapi.context, data)
# data = {'subnet': {'tenant_id': tenant_id,
# 'name': 'dummy',
# 'admin_state_up': True,
# 'network_id': net['id'],
# 'cidr': '1.1.1.0/16',
# 'enable_dhcp': True,
# 'ip_version': 4,
# 'dns_nameservers': None,
# 'host_routes': None,
# 'allocation_pools': None}}
# self._plugin.create_subnet(self.edgeapi.context, data)
# return net
# def get_edge_id(self):
# bindings = nsxv_db.get_nsxv_router_bindings(
# self.edgeapi.context.session)
# for binding in bindings:
# if binding.edge_id:
# return binding.edge_id
# # use a dummy edge
# return "edge-1"
# def test_edge_nsx_updates(self):
# """Test eges/nsx-update utility with different inputs."""
# edge_id = self.get_edge_id()
# self._test_edge_nsx_update(edge_id, ["appliances=true"])
# self._test_edge_nsx_update(edge_id, ["size=compact"])
# self._test_edge_nsx_update(edge_id, ["hostgroup=update"])
# self._test_edge_nsx_update(edge_id, ["hostgroup=all"])
# self._test_edge_nsx_update(edge_id, ["hostgroup=clean"])
# self._test_edge_nsx_update(edge_id, ["highavailability=True"])
# self._test_edge_nsx_update(edge_id, ["resource=cpu", "limit=100"])
# self._test_edge_nsx_update(edge_id, ["syslog-server=1.1.1.1",
# "syslog-proto=tcp",
# "log-level=debug"])
# def test_bad_args(self):
# args = {'property': ["xxx"]}
# errors = self._test_resource_with_errors(
# 'networks', 'nsx-update', **args)
# self.assertEqual(1, len(errors))
# def test_resources_with_common_args(self):
# """Run all nsxv admin utilities with some common arguments
# Using arguments like edge-id which many apis need
# This improves the test coverage
# """
# edge_id = self.get_edge_id()
# args = ["edge-id=%s" % edge_id,
# "router-id=e5b9b249-0034-4729-8ab6-fe4dacaa3a12",
# "policy-id=1",
# "network_id=net-1",
# "net-id=net-1",
# "network=net-1",
# "port=port-1",
# "security-group-id=sg-1",
# "dvs-id=dvs-1",
# "moref=virtualwire-1",
# "teamingpolicy=LACP_ACTIVE",
# "log-allowed-traffic=true",
# "az-name=default",
# "transit-network=abc",
# "moref=abc",
# ]
# self._test_resources_with_args(
# resources.nsxv_resources, args)
# def test_router_recreate(self):
# # Testing router-recreate separately because it may change the
# # edge-id
# edge_id = self.get_edge_id()
# args = {'property': ["edge-id=%s" % edge_id]}
# self._test_resource('routers', 'nsx-recreate', **args)
# def test_migration_validation(self):
# # check that validation fails
# args = {'property': ["transit-network=1.1.1.0/24"]}
# try:
# migration.validate_config_for_migration(
# 'nsx-migrate-v2t', 'validate', None, **args)
# except SystemExit:
# return
# else:
# self.fail()
# class TestNsxv3AdminUtils(AbstractTestAdminUtils,
# test_v3_plugin.NsxV3PluginTestCaseMixin):
# def _patch_object(self, *args, **kwargs):
# patcher = mock.patch.object(*args, **kwargs)
# patcher.start()
# self._patchers.append(patcher)
# def _init_mock_plugin(self):
# test_v3_plugin._mock_nsx_backend_calls()
# # mock resources
# for cls in (nsx_v3_resources.LogicalPort,
# nsx_v3_resources.LogicalDhcpServer,
# core_resources.NsxLibLogicalRouter,
# core_resources.NsxLibSwitchingProfile,
# v3_client.RESTClient):
# self._patch_object(cls, 'list', return_value={'results': []})
# self._patch_object(cls, 'get',
# return_value={
# 'id': uuidutils.generate_uuid()})
# self._patch_object(cls, 'update')
# self._patch_object(core_resources.NsxLibSwitchingProfile,
# 'find_by_display_name',
# return_value=[{'id': uuidutils.generate_uuid()}])
# self._patch_object(nsx_v3_security.NsxLibFirewallSection,
# 'get_excludelist',
# return_value={'members': [{
# 'target_type': 'LogicalPort',
# 'target_id': 'port_id'}]})
# # Mocks for MP2P migration
# mock.patch("vmware_nsxlib.v3.NsxLib.get_version",
# return_value='3.1.0').start()
# mock.patch("vmware_nsx.shell.admin.plugins.nsxv3."
# "resources.migration."
# "ensure_migration_state_ready", return_value=True).start()
# mock.patch("vmware_nsx.shell.admin.plugins.nsxv3.resources."
# "migration."
# "change_migration_service_status").start()
# cfg.CONF.set_override('nsx_api_managers', ['1.1.1.1'], 'nsx_v3')
# cfg.CONF.set_override('nsx_api_user', ['admin'], 'nsx_v3')
# cfg.CONF.set_override('nsx_api_password', ['dummy'], 'nsx_v3')
# super(TestNsxv3AdminUtils, self)._init_mock_plugin()
# self._plugin = nsxv3_utils.NsxV3PluginWrapper()
# mock_nm_get_plugin = mock.patch(
# "neutron_lib.plugins.directory.get_plugin")
# self.mock_nm_get_plugin = mock_nm_get_plugin.start()
# self.mock_nm_get_plugin.return_value = self._plugin
# def _get_plugin_name(self):
# return 'nsxv3'
# def test_nsxv3_resources(self):
# self._test_resources(resources.nsxv3_resources)
# def test_resources_with_common_args(self):
# """Run all nsxv3 admin utilities with some common arguments
# Using arguments like dhcp_profile_uuid which many apis need
# This improves the test coverage
# """
# args = ["dhcp_profile_uuid=e5b9b249-0034-4729-8ab6-fe4dacaa3a12",
# "metadata_proxy_uuid=e5b9b249-0034-4729-8ab6-fe4dacaa3a12",
# "nsx-id=e5b9b249-0034-4729-8ab6-fe4dacaa3a12",
# "net-id=e5b9b249-0034-4729-8ab6-fe4dacaa3a12",
# "availability-zone=default",
# "server-ip=1.1.1.1",
# "log-allowed-traffic=true",
# "value=10",
# "old-tier0=olduuid",
# "new-tier0=newuuid",
# "project-id=aaa",
# "host-moref=dummy-moref"
# ]
# # Create some neutron objects for the utilities to run on
# self._create_router()
# with self._create_l3_ext_network() as network:
# with self.subnet(network=network) as subnet:
# with self.port(subnet=subnet):
# # Run all utilities with backend objects
# self._test_resources_with_args(
# resources.nsxv3_resources, args)
# class TestNsxtvdAdminUtils(AbstractTestAdminUtils):
# def _get_plugin_name(self):
# return 'nsxtvd'
# def test_nsxtv_resources(self):
# self._test_resources(resources.nsxtvd_resources)
# class TestNsxpAdminUtils(AbstractTestAdminUtils,
# test_p_plugin.NsxPPluginTestCaseMixin):
# def _get_plugin_name(self):
# return 'nsxp'
# def _init_mock_plugin(self):
# self._plugin = nsxp_utils.NsxPolicyPluginWrapper()
# def test_nsxp_resources(self):
# self._test_resources(resources.nsxp_resources)
# def test_nsxp_resources_with_objects(self):
# # Create some neutron objects for the utilities to run on
# self._create_router()
# with self.network():
# # Run all utilities with backend objects
# self._test_resources(resources.nsxp_resources)
# def test_resources_with_common_args(self):
# """Run all nsxp admin utilities with some common arguments
# Using arguments some apis need to improves the test coverage
# """
# args = ["realization_interval=1",
# "dhcp-config=dumyuuid",
# "old-tier0=olduuid",
# "new-tier0=newuuid",
# "firewall-match=internal"]
# # Create some neutron objects for the utilities to run on
# self._create_router()
# with self._create_l3_ext_network() as network:
# with self.subnet(network=network, enable_dhcp=False) as subnet:
# with self.port(subnet=subnet):
# # Run all utilities with backend objects
# self._test_resources_with_args(
# resources.nsxp_resources, args)
NSX_INI_PATH = vmware.get_fake_conf('nsx.ini.test')
BASE_CONF_PATH = vmware.get_fake_conf('neutron.conf.test')
class AbstractTestAdminUtils(base.BaseTestCase, metaclass=abc.ABCMeta):
def setUp(self):
cfg.CONF.unregister_opts(_options.common_cli_opts)
cfg.CONF.register_cli_opts(resources.cli_opts)
super(AbstractTestAdminUtils, self).setUp()
# remove resource registration conflicts
resource_registry.unregister_all_resources()
self.edgeapi = nsxv_utils.NeutronDbClient()
# Init the neutron config
neutron_config.init(args=['--config-file', BASE_CONF_PATH,
'--config-file', NSX_INI_PATH])
self._init_mock_plugin()
self._init_resource_plugin()
self.addCleanup(resource_registry.unregister_all_resources)
def _init_mock_plugin(self):
mock_query = mock.patch(
"vmware_nsx.shell.admin.plugins.common.utils.query_yes_no")
mock_query.start()
@abc.abstractmethod
def _get_plugin_name(self):
pass
def _init_resource_plugin(self):
plugin_name = self._get_plugin_name()
resources.init_resource_plugin(
plugin_name,
resources.get_plugin_dir(plugin_name))
def _test_resource(self, res_name, op, **kwargs):
errors = self._test_resource_with_errors(res_name, op, **kwargs)
if len(errors) > 0:
msg = (_("admin util %(res)s/%(op)s failed with message: "
"%(err)s") % {'res': res_name,
'op': op,
'err': errors[0]})
self.fail(msg=msg)
def _test_resource_with_errors(self, res_name, op, **kwargs):
# Must call the internal notify_loop in order to get the errors
return registry._get_callback_manager()._notify_loop(
res_name, op, 'nsxadmin', **kwargs)
def _test_resources(self, res_dict):
for res in res_dict.keys():
res_name = res_dict[res].name
for op in res_dict[res].supported_ops:
self._test_resource(res_name, op)
def _test_resources_with_args(self, res_dict, func_args):
for res in res_dict.keys():
res_name = res_dict[res].name
for op in res_dict[res].supported_ops:
args = {'property': func_args}
self._test_resource(res_name, op, **args)
def _create_router(self):
tenant_id = uuidutils.generate_uuid()
data = {'router': {'tenant_id': tenant_id}}
data['router']['name'] = 'dummy'
data['router']['admin_state_up'] = True
return self._plugin.create_router(self.edgeapi.context, data)
class TestNsxvAdminUtils(AbstractTestAdminUtils,
test_v_plugin.NsxVPluginV2TestCase):
def _get_plugin_name(self):
return 'nsxv'
def _init_mock_plugin(self, *mocks):
super(TestNsxvAdminUtils, self)._init_mock_plugin()
# support the dvs manager:
mock.patch.object(dvs_utils, 'dvs_create_session').start()
# override metadata get-object
dummy_lb = {
'enabled': True,
'enableServiceInsertion': True,
'accelerationEnabled': True,
'virtualServer': [],
'applicationProfile': [],
'pool': [],
'applicationRule': []
}
mock.patch('vmware_nsx.plugins.nsx_v.vshield.nsxv_edge_cfg_obj.'
'NsxvEdgeCfgObj.get_object',
return_value=dummy_lb).start()
# Tests shouldn't wait for dummy spawn jobs to finish
mock.patch('vmware_nsx.shell.admin.plugins.nsxv.resources.utils.'
'NsxVPluginWrapper.count_spawn_jobs',
return_value=0).start()
self._plugin = nsxv_utils.NsxVPluginWrapper()
def get_plugin_mock(alias=constants.CORE):
if alias in (constants.CORE, constants.L3):
return self._plugin
mock.patch("neutron_lib.plugins.directory.get_plugin",
side_effect=get_plugin_mock).start()
# Create a router to make sure we have deployed an edge
self.router = self._create_router()
self.dist_router = self._create_router(dist=True)
self.network = self._create_net()
def tearDown(self):
if self.router and self.router.get('id'):
self._plugin.delete_router(
self.edgeapi.context, self.router['id'])
if self.dist_router and self.dist_router.get('id'):
self._plugin.delete_router(
self.edgeapi.context, self.dist_router['id'])
if self.network and self.network.get('id'):
self._plugin.delete_network(
self.edgeapi.context, self.network['id'])
super(TestNsxvAdminUtils, self).tearDown()
def test_nsxv_resources(self):
self._test_resources(resources.nsxv_resources)
def _test_edge_nsx_update(self, edge_id, params):
args = {'property': ["edge-id=%s" % edge_id]}
args['property'].extend(params)
self._test_resource('edges', 'nsx-update', **args)
def _create_router(self, dist=False):
# Create an exclusive router (with an edge)
tenant_id = uuidutils.generate_uuid()
data = {'router': {'tenant_id': tenant_id}}
data['router']['name'] = 'dummy'
data['router']['admin_state_up'] = True
if dist:
data['router']['distributes'] = True
else:
data['router']['router_type'] = 'exclusive'
return self._plugin.create_router(self.edgeapi.context, data)
def _create_net(self):
tenant_id = uuidutils.generate_uuid()
data = {'network': {'tenant_id': tenant_id,
'name': 'dummy',
'admin_state_up': True,
'shared': False}}
net = self._plugin.create_network(self.edgeapi.context, data)
data = {'subnet': {'tenant_id': tenant_id,
'name': 'dummy',
'admin_state_up': True,
'network_id': net['id'],
'cidr': '1.1.1.0/16',
'enable_dhcp': True,
'ip_version': 4,
'dns_nameservers': None,
'host_routes': None,
'allocation_pools': None}}
self._plugin.create_subnet(self.edgeapi.context, data)
return net
def get_edge_id(self):
bindings = nsxv_db.get_nsxv_router_bindings(
self.edgeapi.context.session)
for binding in bindings:
if binding.edge_id:
return binding.edge_id
# use a dummy edge
return "edge-1"
def test_edge_nsx_updates(self):
"""Test eges/nsx-update utility with different inputs."""
edge_id = self.get_edge_id()
self._test_edge_nsx_update(edge_id, ["appliances=true"])
self._test_edge_nsx_update(edge_id, ["size=compact"])
self._test_edge_nsx_update(edge_id, ["hostgroup=update"])
self._test_edge_nsx_update(edge_id, ["hostgroup=all"])
self._test_edge_nsx_update(edge_id, ["hostgroup=clean"])
self._test_edge_nsx_update(edge_id, ["highavailability=True"])
self._test_edge_nsx_update(edge_id, ["resource=cpu", "limit=100"])
self._test_edge_nsx_update(edge_id, ["syslog-server=1.1.1.1",
"syslog-proto=tcp",
"log-level=debug"])
def test_bad_args(self):
args = {'property': ["xxx"]}
errors = self._test_resource_with_errors(
'networks', 'nsx-update', **args)
self.assertEqual(1, len(errors))
def test_resources_with_common_args(self):
"""Run all nsxv admin utilities with some common arguments
Using arguments like edge-id which many apis need
This improves the test coverage
"""
edge_id = self.get_edge_id()
args = ["edge-id=%s" % edge_id,
"router-id=e5b9b249-0034-4729-8ab6-fe4dacaa3a12",
"policy-id=1",
"network_id=net-1",
"net-id=net-1",
"network=net-1",
"port=port-1",
"security-group-id=sg-1",
"dvs-id=dvs-1",
"moref=virtualwire-1",
"teamingpolicy=LACP_ACTIVE",
"log-allowed-traffic=true",
"az-name=default",
"transit-network=abc",
"moref=abc",
]
self._test_resources_with_args(
resources.nsxv_resources, args)
def test_router_recreate(self):
# Testing router-recreate separately because it may change the edge-id
edge_id = self.get_edge_id()
args = {'property': ["edge-id=%s" % edge_id]}
self._test_resource('routers', 'nsx-recreate', **args)
def test_migration_validation(self):
# check that validation fails
args = {'property': ["transit-network=1.1.1.0/24"]}
try:
migration.validate_config_for_migration(
'nsx-migrate-v2t', 'validate', None, **args)
except SystemExit:
return
else:
self.fail()
class TestNsxv3AdminUtils(AbstractTestAdminUtils,
test_v3_plugin.NsxV3PluginTestCaseMixin):
def _patch_object(self, *args, **kwargs):
patcher = mock.patch.object(*args, **kwargs)
patcher.start()
self._patchers.append(patcher)
def _init_mock_plugin(self):
test_v3_plugin._mock_nsx_backend_calls()
# mock resources
for cls in (nsx_v3_resources.LogicalPort,
nsx_v3_resources.LogicalDhcpServer,
core_resources.NsxLibLogicalRouter,
core_resources.NsxLibSwitchingProfile,
v3_client.RESTClient):
self._patch_object(cls, 'list', return_value={'results': []})
self._patch_object(cls, 'get',
return_value={'id': uuidutils.generate_uuid()})
self._patch_object(cls, 'update')
self._patch_object(core_resources.NsxLibSwitchingProfile,
'find_by_display_name',
return_value=[{'id': uuidutils.generate_uuid()}])
self._patch_object(nsx_v3_security.NsxLibFirewallSection,
'get_excludelist',
return_value={'members': [{
'target_type': 'LogicalPort',
'target_id': 'port_id'}]})
# Mocks for MP2P migration
mock.patch("vmware_nsxlib.v3.NsxLib.get_version",
return_value='3.1.0').start()
mock.patch("vmware_nsx.shell.admin.plugins.nsxv3.resources.migration."
"ensure_migration_state_ready", return_value=True).start()
mock.patch("vmware_nsx.shell.admin.plugins.nsxv3.resources.migration."
"change_migration_service_status").start()
cfg.CONF.set_override('nsx_api_managers', ['1.1.1.1'], 'nsx_v3')
cfg.CONF.set_override('nsx_api_user', ['admin'], 'nsx_v3')
cfg.CONF.set_override('nsx_api_password', ['dummy'], 'nsx_v3')
super(TestNsxv3AdminUtils, self)._init_mock_plugin()
self._plugin = nsxv3_utils.NsxV3PluginWrapper()
mock_nm_get_plugin = mock.patch(
"neutron_lib.plugins.directory.get_plugin")
self.mock_nm_get_plugin = mock_nm_get_plugin.start()
self.mock_nm_get_plugin.return_value = self._plugin
def _get_plugin_name(self):
return 'nsxv3'
def test_nsxv3_resources(self):
self._test_resources(resources.nsxv3_resources)
def test_resources_with_common_args(self):
"""Run all nsxv3 admin utilities with some common arguments
Using arguments like dhcp_profile_uuid which many apis need
This improves the test coverage
"""
args = ["dhcp_profile_uuid=e5b9b249-0034-4729-8ab6-fe4dacaa3a12",
"metadata_proxy_uuid=e5b9b249-0034-4729-8ab6-fe4dacaa3a12",
"nsx-id=e5b9b249-0034-4729-8ab6-fe4dacaa3a12",
"net-id=e5b9b249-0034-4729-8ab6-fe4dacaa3a12",
"availability-zone=default",
"server-ip=1.1.1.1",
"log-allowed-traffic=true",
"value=10",
"old-tier0=olduuid",
"new-tier0=newuuid",
"project-id=aaa",
"host-moref=dummy-moref"
]
# Create some neutron objects for the utilities to run on
self._create_router()
with self._create_l3_ext_network() as network:
with self.subnet(network=network) as subnet:
with self.port(subnet=subnet):
# Run all utilities with backend objects
self._test_resources_with_args(
resources.nsxv3_resources, args)
class TestNsxtvdAdminUtils(AbstractTestAdminUtils):
def _get_plugin_name(self):
return 'nsxtvd'
def test_nsxtv_resources(self):
self._test_resources(resources.nsxtvd_resources)
class TestNsxpAdminUtils(AbstractTestAdminUtils,
test_p_plugin.NsxPPluginTestCaseMixin):
def _get_plugin_name(self):
return 'nsxp'
def _init_mock_plugin(self):
super(TestNsxpAdminUtils, self)._init_mock_plugin()
self._plugin = nsxp_utils.NsxPolicyPluginWrapper()
def test_nsxp_resources(self):
self._test_resources(resources.nsxp_resources)
def test_nsxp_resources_with_objects(self):
# Create some neutron objects for the utilities to run on
self._create_router()
with self.network():
# Run all utilities with backend objects
self._test_resources(resources.nsxp_resources)
def test_resources_with_common_args(self):
"""Run all nsxp admin utilities with some common arguments
Using arguments some apis need to improves the test coverage
"""
args = ["realization_interval=1",
"dhcp-config=dumyuuid",
"old-tier0=olduuid",
"new-tier0=newuuid",
"firewall-match=internal"]
# Create some neutron objects for the utilities to run on
self._create_router()
with self._create_l3_ext_network() as network:
with self.subnet(network=network, enable_dhcp=False) as subnet:
with self.port(subnet=subnet):
# Run all utilities with backend objects
self._test_resources_with_args(
resources.nsxp_resources, args)

Write Preview
Loading…
Cancel
Save