# Copyright 2015 VMware, Inc. # All Rights Reserved # # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import abc from unittest import mock 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 vmware_nsx._i18n import _ from vmware_nsx.common import config # noqa from vmware_nsx.services.lbaas.octavia import octavia_listener from vmware_nsx.shell.admin.plugins.common import utils as admin_utils from vmware_nsx.shell.admin.plugins.nsxp.resources import utils as nsxp_utils 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_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() octavia_listener.get_octavia_rpc_client = mock.Mock() @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): payload = admin_utils.MetadataEventPayload(kwargs) # Must call the internal notify_loop in order to get the errors return registry._get_callback_manager()._notify_loop( res_name, op, 'nsxadmin', payload=payload) 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 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", "tier0=newuuid", "metadata-proxy=mdproxy", "az=default", "dhcp-profile=openstack_dchp_profile"] # 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)