From 2a1991a3a62a7d85ac6ad3b2c43010e3455c6aa7 Mon Sep 17 00:00:00 2001 From: Adit Sarfaty Date: Mon, 7 Dec 2020 11:16:00 +0000 Subject: [PATCH] Revert "Temporarily disable admin utils tests" This reverts commit 09fbad47d822d7234d9e69b038ab45e7885a0cf1. Reason for revert: Re-Enable the tests Change-Id: I48e9e7f7e41084c8def0d46804a05dc17bdb5ade --- .coveragerc | 2 +- tox.ini | 7 + .../admin/plugins/nsxp/resources/migration.py | 3 +- .../tests/unit/shell/test_admin_utils.py | 687 +++++++++--------- 4 files changed, 351 insertions(+), 348 deletions(-) diff --git a/.coveragerc b/.coveragerc index c1a02adbde..297011e10c 100644 --- a/.coveragerc +++ b/.coveragerc @@ -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 diff --git a/tox.ini b/tox.ini index cbe0b58c40..bd1f84d4db 100644 --- a/tox.ini +++ b/tox.ini @@ -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 = diff --git a/vmware_nsx/shell/admin/plugins/nsxp/resources/migration.py b/vmware_nsx/shell/admin/plugins/nsxp/resources/migration.py index c064b8aa3e..84ee2aa1f3 100644 --- a/vmware_nsx/shell/admin/plugins/nsxp/resources/migration.py +++ b/vmware_nsx/shell/admin/plugins/nsxp/resources/migration.py @@ -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, diff --git a/vmware_nsx/tests/unit/shell/test_admin_utils.py b/vmware_nsx/tests/unit/shell/test_admin_utils.py index 2acaa234ed..7c55a3da3a 100644 --- a/vmware_nsx/tests/unit/shell/test_admin_utils.py +++ b/vmware_nsx/tests/unit/shell/test_admin_utils.py @@ -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 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 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 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') +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): +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) + def setUp(self): + cfg.CONF.unregister_opts(_options.common_cli_opts) + cfg.CONF.register_cli_opts(resources.cli_opts) -# super(AbstractTestAdminUtils, self).setUp() + super(AbstractTestAdminUtils, self).setUp() -# # remove resource registration conflicts -# resource_registry.unregister_all_resources() + # 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) + 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() + 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 + @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 _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(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_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(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 _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) + 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): +class TestNsxvAdminUtils(AbstractTestAdminUtils, + test_v_plugin.NsxVPluginV2TestCase): -# def _get_plugin_name(self): -# return 'nsxv' + def _get_plugin_name(self): + return 'nsxv' -# def _init_mock_plugin(self, *mocks): -# super(TestNsxvAdminUtils, self)._init_mock_plugin() + 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() + # 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() + # 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() + self._plugin = nsxv_utils.NsxVPluginWrapper() -# def get_plugin_mock(alias=constants.CORE): -# if alias in (constants.CORE, constants.L3): -# return self._plugin + 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() + 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() + # 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 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_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 _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' + 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) + 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 _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 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_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_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 + 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) + 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_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() + 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): +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 _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() + 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') + # 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'}]}) + 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() + # 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') + 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() + 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 + 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 _get_plugin_name(self): + return 'nsxv3' -# def test_nsxv3_resources(self): -# self._test_resources(resources.nsxv3_resources) + 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 + 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) + 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): +class TestNsxtvdAdminUtils(AbstractTestAdminUtils): -# def _get_plugin_name(self): -# return 'nsxtvd' + def _get_plugin_name(self): + return 'nsxtvd' -# def test_nsxtv_resources(self): -# self._test_resources(resources.nsxtvd_resources) + def test_nsxtv_resources(self): + self._test_resources(resources.nsxtvd_resources) -# class TestNsxpAdminUtils(AbstractTestAdminUtils, -# test_p_plugin.NsxPPluginTestCaseMixin): +class TestNsxpAdminUtils(AbstractTestAdminUtils, + test_p_plugin.NsxPPluginTestCaseMixin): -# def _get_plugin_name(self): -# return 'nsxp' + def _get_plugin_name(self): + return 'nsxp' -# def _init_mock_plugin(self): -# self._plugin = nsxp_utils.NsxPolicyPluginWrapper() + 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(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_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 + 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) + 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)