x/vmware-nsx
Code Issues Proposed changes

Unit test for nsx|v + nsx|t admin utils

Browse Source 
New units tests that loads and runs all the admin utilities
resources and operations.
We do not check specific arguments and scenarios for now. Just making sure
it can run.

Change-Id: Iac7fcf4fc2546856ba5f3fc22e868c0ed7839443
This commit is contained in:
Adit Sarfaty 2016-06-23 18:23:35 +03:00
parent 2ca5f17dd7
commit db262e5b95
19 changed files with 346 additions and 159 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
vmware_nsx/shell/admin/plugins/nsxv/resources/backup_edges.py
View File

@ -31,7 +31,7 @@ from vmware_nsx.shell.admin.plugins.common import constants
from vmware_nsx.shell.admin.plugins.common import formatters
import vmware_nsx.shell.admin.plugins.common.utils as admin_utils
import vmware_nsx.shell.admin.plugins.nsxv.resources.utils as utils
import vmware_nsx.shell.nsxadmin as shell
import vmware_nsx.shell.resources as shell
LOG = logging.getLogger(__name__)

2
vmware_nsx/shell/admin/plugins/nsxv/resources/dhcp_binding.py
View File

@ -20,7 +20,7 @@ from vmware_nsx.shell.admin.plugins.common import constants
import vmware_nsx.shell.admin.plugins.common.utils as admin_utils
import vmware_nsx.shell.admin.plugins.nsxv.resources.utils as utils
import vmware_nsx.shell.nsxadmin as shell
import vmware_nsx.shell.resources as shell
from neutron.callbacks import registry
from neutron.db import db_base_plugin_v2

2
vmware_nsx/shell/admin/plugins/nsxv/resources/edges.py
View File

@ -21,7 +21,7 @@ from vmware_nsx.shell.admin.plugins.common import formatters
import vmware_nsx.shell.admin.plugins.common.utils as admin_utils
import vmware_nsx.shell.admin.plugins.nsxv.resources.utils as utils
import vmware_nsx.shell.nsxadmin as shell
import vmware_nsx.shell.resources as shell
from neutron.callbacks import registry
from neutron_lib import exceptions

2
vmware_nsx/shell/admin/plugins/nsxv/resources/metadata.py
View File

@ -31,7 +31,7 @@ from vmware_nsx.plugins.nsx_v.vshield import nsxv_loadbalancer as nsxv_lb
from vmware_nsx.shell.admin.plugins.common import constants
from vmware_nsx.shell.admin.plugins.common import utils as admin_utils
from vmware_nsx.shell.admin.plugins.nsxv.resources import utils as utils
from vmware_nsx.shell import nsxadmin as shell
from vmware_nsx.shell import resources as shell
LOG = logging.getLogger(__name__)

2
vmware_nsx/shell/admin/plugins/nsxv/resources/networks.py
View File

@ -23,7 +23,7 @@ from vmware_nsx.shell.admin.plugins.common import constants
from vmware_nsx.shell.admin.plugins.common import formatters
from vmware_nsx.shell.admin.plugins.common import utils as admin_utils
from vmware_nsx.shell.admin.plugins.nsxv.resources import utils as utils
from vmware_nsx.shell import nsxadmin as shell
from vmware_nsx.shell import resources as shell
from neutron.callbacks import registry

5
vmware_nsx/shell/admin/plugins/nsxv/resources/securitygroups.py
View File

@ -29,7 +29,7 @@ from vmware_nsx.shell.admin.plugins.common import constants
from vmware_nsx.shell.admin.plugins.common import formatters
from vmware_nsx.shell.admin.plugins.common import utils as admin_utils
from vmware_nsx.shell.admin.plugins.nsxv.resources import utils
from vmware_nsx.shell import nsxadmin
from vmware_nsx.shell import resources as nsxadmin
LOG = logging.getLogger(__name__)
@ -39,6 +39,9 @@ class NsxVPluginWrapper(plugin.NsxVPlugin):
def _start_rpc_listeners(self):
pass
def _validate_config(self):
pass
class NeutronSecurityGroupDB(utils.NeutronDbClient,
securitygroups_db.SecurityGroupDbMixin):

2
vmware_nsx/shell/admin/plugins/nsxv/resources/spoofguard_policy.py
View File

@ -21,7 +21,7 @@ from vmware_nsx.shell.admin.plugins.common import formatters
import vmware_nsx.shell.admin.plugins.common.utils as admin_utils
import vmware_nsx.shell.admin.plugins.nsxv.resources.utils as utils
import vmware_nsx.shell.nsxadmin as shell
import vmware_nsx.shell.resources as shell
from neutron.callbacks import registry
from neutron_lib import exceptions

2
vmware_nsx/shell/admin/plugins/nsxv3/resources/dhcp_binding.py
View File

@ -31,7 +31,7 @@ from vmware_nsx.shell.admin.plugins.common import constants
from vmware_nsx.shell.admin.plugins.common import formatters
from vmware_nsx.shell.admin.plugins.common import utils as admin_utils
from vmware_nsx.shell.admin.plugins.nsxv3.resources import utils
import vmware_nsx.shell.nsxadmin as shell
import vmware_nsx.shell.resources as shell
LOG = logging.getLogger(__name__)
neutron_client = utils.NeutronDbClient()

2
vmware_nsx/shell/admin/plugins/nsxv3/resources/metadata_proxy.py
View File

@ -26,7 +26,7 @@ from vmware_nsx.nsxlib.v3 import resources
from vmware_nsx.shell.admin.plugins.common import constants
from vmware_nsx.shell.admin.plugins.common import utils as admin_utils
from vmware_nsx.shell.admin.plugins.nsxv3.resources import utils
import vmware_nsx.shell.nsxadmin as shell
import vmware_nsx.shell.resources as shell
LOG = logging.getLogger(__name__)
neutron_client = utils.NeutronDbClient()

2
vmware_nsx/shell/admin/plugins/nsxv3/resources/networks.py
View File

@ -22,7 +22,7 @@ from vmware_nsx.nsxlib import v3 as nsxlib
from vmware_nsx.shell.admin.plugins.common import constants
from vmware_nsx.shell.admin.plugins.common import formatters
from vmware_nsx.shell.admin.plugins.common import utils as admin_utils
from vmware_nsx.shell import nsxadmin as shell
from vmware_nsx.shell import resources as shell
from neutron.callbacks import registry
from neutron import context as neutron_context

2
vmware_nsx/shell/admin/plugins/nsxv3/resources/ports.py
View File

@ -29,7 +29,7 @@ from vmware_nsx.services.qos.common import utils as qos_utils
from vmware_nsx.shell.admin.plugins.common import constants
from vmware_nsx.shell.admin.plugins.common import formatters
from vmware_nsx.shell.admin.plugins.common import utils as admin_utils
from vmware_nsx.shell import nsxadmin as shell
from vmware_nsx.shell import resources as shell
from neutron.callbacks import registry
from neutron import context as neutron_context

2
vmware_nsx/shell/admin/plugins/nsxv3/resources/routers.py
View File

@ -24,7 +24,7 @@ from vmware_nsx.nsxlib.v3 import resources as nsx_resources
from vmware_nsx.shell.admin.plugins.common import constants
from vmware_nsx.shell.admin.plugins.common import formatters
from vmware_nsx.shell.admin.plugins.common import utils as admin_utils
from vmware_nsx.shell import nsxadmin as shell
from vmware_nsx.shell import resources as shell
from neutron.callbacks import registry
from neutron import context as neutron_context

2
vmware_nsx/shell/admin/plugins/nsxv3/resources/securitygroups.py
View File

@ -18,7 +18,7 @@ from vmware_nsx.common import utils
from vmware_nsx.shell.admin.plugins.common import constants
from vmware_nsx.shell.admin.plugins.common import formatters
from vmware_nsx.shell.admin.plugins.common import utils as admin_utils
from vmware_nsx.shell import nsxadmin as shell
from vmware_nsx.shell import resources as shell
from neutron.callbacks import registry
from neutron import context as neutron_context

157
vmware_nsx/shell/nsxadmin.py
View File

@ -24,11 +24,7 @@ TODO: Add support for other resources, ports, logical switches etc.
TODO: Autocomplete command line args
"""
import enum
import glob
import importlib
import logging
import os
import requests
import sys
@ -43,6 +39,7 @@ from oslo_log import _options
from vmware_nsx.shell.admin.plugins.common import constants
from vmware_nsx.shell.admin import version
from vmware_nsx.shell import resources
# Suppress the Insecure request warning
requests.packages.urllib3.disable_warnings()
@ -50,94 +47,6 @@ requests.packages.urllib3.disable_warnings()
LOG = logging.getLogger(__name__)
class Operations(enum.Enum):
LIST = 'list'
CLEAN = 'clean'
LIST_MISMATCHES = 'list-mismatches'
FIX_MISMATCH = 'fix-mismatch'
NEUTRON_LIST = 'neutron-list'
NEUTRON_CLEAN = 'neutron-clean'
NEUTRON_UPDATE = 'neutron-update'
NSX_LIST = 'nsx-list'
NSX_CLEAN = 'nsx-clean'
NSX_UPDATE = 'nsx-update'
NSX_UPDATE_SECRET = 'nsx-update-secret'
ops = [op.value for op in Operations]
class Resource(object):
def __init__(self, name, ops):
self.name = name
self.supported_ops = ops
# Add supported NSX-V3 resources in this dictionary
nsxv3_resources = {
constants.SECURITY_GROUPS: Resource(constants.SECURITY_GROUPS,
[Operations.CLEAN.value,
Operations.LIST.value,
Operations.NSX_LIST.value,
Operations.NSX_CLEAN.value,
Operations.NEUTRON_LIST.value,
Operations.NEUTRON_CLEAN.value]),
constants.NETWORKS: Resource(constants.NETWORKS,
[Operations.LIST_MISMATCHES.value]),
constants.PORTS: Resource(constants.PORTS,
[Operations.LIST_MISMATCHES.value]),
constants.ROUTERS: Resource(constants.ROUTERS,
[Operations.LIST_MISMATCHES.value]),
constants.DHCP_BINDING: Resource(constants.DHCP_BINDING,
[Operations.LIST.value,
Operations.NSX_UPDATE.value]),
}
# Add supported NSX-V resources in this dictionary
nsxv_resources = {
constants.EDGES: Resource(constants.EDGES,
[Operations.NSX_LIST.value,
Operations.NEUTRON_LIST.value,
Operations.NSX_UPDATE.value]),
constants.BACKUP_EDGES: Resource(constants.BACKUP_EDGES,
[Operations.LIST.value,
Operations.CLEAN.value,
Operations.LIST_MISMATCHES.value,
Operations.FIX_MISMATCH.value]),
constants.ORPHANED_EDGES: Resource(constants.ORPHANED_EDGES,
[Operations.LIST.value,
Operations.CLEAN.value]),
constants.MISSING_EDGES: Resource(constants.MISSING_EDGES,
[Operations.LIST.value]),
constants.SPOOFGUARD_POLICY: Resource(constants.SPOOFGUARD_POLICY,
[Operations.LIST.value,
Operations.CLEAN.value]),
constants.DHCP_BINDING: Resource(constants.DHCP_BINDING,
[Operations.LIST.value,
Operations.NSX_UPDATE.value]),
constants.NETWORKS: Resource(constants.NETWORKS,
[Operations.LIST.value,
Operations.NSX_UPDATE.value]),
constants.SECURITY_GROUPS: Resource(constants.SECURITY_GROUPS,
[Operations.LIST.value,
Operations.FIX_MISMATCH.value]),
constants.FIREWALL_SECTIONS: Resource(constants.FIREWALL_SECTIONS,
[Operations.LIST.value,
Operations.LIST_MISMATCHES.value]),
constants.FIREWALL_NSX_GROUPS: Resource(
constants.FIREWALL_NSX_GROUPS, [Operations.LIST.value,
Operations.LIST_MISMATCHES.value]),
constants.METADATA: Resource(
constants.METADATA, [Operations.NSX_UPDATE.value,
Operations.NSX_UPDATE_SECRET]),
}
nsxv3_resources_names = map(lambda res: res.name, nsxv3_resources.itervalues())
nsxv_resources_names = map(lambda res: res.name, nsxv_resources.itervalues())
def _get_plugin():
plugin = cfg.CONF.core_plugin
plugin_name = ''
@ -148,54 +57,8 @@ def _get_plugin():
return plugin_name
def _get_plugin_dir():
plugin_dir = os.path.dirname(os.path.realpath(__file__)) + "/admin/plugins"
return '{}/{}/resources'.format(plugin_dir, _get_plugin())
def _get_resources():
modules = glob.glob(_get_plugin_dir() + "/*.py")
return map(lambda module: os.path.splitext(os.path.basename(module))[0],
modules)
cli_opts = [cfg.StrOpt('fmt',
short='f',
default='psql',
choices=['psql', 'json'],
help='Supported output formats: json, psql'),
cfg.StrOpt('resource',
short='r',
choices=nsxv_resources_names + nsxv3_resources_names,
help='Supported list of resources: NSX-V3: %s '
'NSX-V: %s' % (', '.join(nsxv3_resources_names),
', '.join(nsxv_resources_names))),
cfg.StrOpt('operation',
short='o',
help='Supported list of operations: {}'
.format(', '.join(ops))),
cfg.BoolOpt('force',
default=False,
help='Enables \'force\' mode. No confirmations will '
'be made before deletions.'),
cfg.MultiStrOpt('property',
short='p',
help='Key-value pair containing the information '
'to be updated. For ex: key=value.')
]
def _init_resource_plugin():
resources = _get_resources()
for resource in resources:
if resource != '__init__':
importlib.import_module("." + resource,
"vmware_nsx.shell.admin.plugins."
"{}.resources".format(_get_plugin()))
def _init_cfg():
cfg.CONF.register_cli_opts(cli_opts)
cfg.CONF.register_cli_opts(resources.cli_opts)
# NOTE(gangila): neutron.common.config registers some options by default
# which are then shown in the help message. We don't need them
@ -217,20 +80,20 @@ def _init_cfg():
def _validate_resource_choice(resource, nsx_plugin):
if nsx_plugin == 'nsxv' and resource not in nsxv_resources:
if nsx_plugin == 'nsxv' and resource not in resources.nsxv_resources:
LOG.error(_LE('Supported list of NSX-V resources: %s'),
nsxv_resources_names)
resources.nsxv_resources_names)
sys.exit(1)
elif nsx_plugin == 'nsxv3'and resource not in nsxv3_resources:
elif nsx_plugin == 'nsxv3'and resource not in resources.nsxv3_resources:
LOG.error(_LE('Supported list of NSX-V3 resources: %s'),
nsxv3_resources_names)
resources.nsxv3_resources_names)
sys.exit(1)
def _validate_op_choice(choice, nsx_plugin):
if nsx_plugin == 'nsxv':
supported_resource_ops = \
nsxv_resources[cfg.CONF.resource].supported_ops
resources.nsxv_resources[cfg.CONF.resource].supported_ops
if choice not in supported_resource_ops:
LOG.error(_LE('Supported list of operations for the NSX-V '
'resource %s'), supported_resource_ops)
@ -238,7 +101,7 @@ def _validate_op_choice(choice, nsx_plugin):
elif nsx_plugin == 'nsxv3':
supported_resource_ops = \
nsxv3_resources[cfg.CONF.resource].supported_ops
resources.nsxv3_resources[cfg.CONF.resource].supported_ops
if choice not in supported_resource_ops:
LOG.error(_LE('Supported list of operations for the NSX-V3 '
'resource %s'), supported_resource_ops)
@ -247,8 +110,10 @@ def _validate_op_choice(choice, nsx_plugin):
def main(argv=sys.argv[1:]):
_init_cfg()
_init_resource_plugin()
nsx_plugin_in_use = _get_plugin()
resources.init_resource_plugin(
nsx_plugin_in_use,
resources.get_plugin_dir(nsx_plugin_in_use))
LOG.info(_LI('NSX Plugin in use: %s'), nsx_plugin_in_use)
_validate_resource_choice(cfg.CONF.resource, nsx_plugin_in_use)

165
vmware_nsx/shell/resources.py Normal file
View File

@ -0,0 +1,165 @@
# 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 enum
import glob
import importlib
import logging
import os
from oslo_config import cfg
import requests
from vmware_nsx.common import config # noqa
from vmware_nsx.shell.admin.plugins.common import constants
# Suppress the Insecure request warning
requests.packages.urllib3.disable_warnings()
LOG = logging.getLogger(__name__)
class Operations(enum.Enum):
LIST = 'list'
CLEAN = 'clean'
LIST_MISMATCHES = 'list-mismatches'
FIX_MISMATCH = 'fix-mismatch'
NEUTRON_LIST = 'neutron-list'
NEUTRON_CLEAN = 'neutron-clean'
NEUTRON_UPDATE = 'neutron-update'
NSX_LIST = 'nsx-list'
NSX_CLEAN = 'nsx-clean'
NSX_UPDATE = 'nsx-update'
NSX_UPDATE_SECRET = 'nsx-update-secret'
ops = [op.value for op in Operations]
class Resource(object):
def __init__(self, name, ops):
self.name = name
self.supported_ops = ops
# Add supported NSX-V3 resources in this dictionary
nsxv3_resources = {
constants.SECURITY_GROUPS: Resource(constants.SECURITY_GROUPS,
[Operations.CLEAN.value,
Operations.LIST.value,
Operations.NSX_LIST.value,
Operations.NSX_CLEAN.value,
Operations.NEUTRON_LIST.value,
Operations.NEUTRON_CLEAN.value]),
constants.NETWORKS: Resource(constants.NETWORKS,
[Operations.LIST_MISMATCHES.value]),
constants.PORTS: Resource(constants.PORTS,
[Operations.LIST_MISMATCHES.value]),
constants.ROUTERS: Resource(constants.ROUTERS,
[Operations.LIST_MISMATCHES.value]),
constants.DHCP_BINDING: Resource(constants.DHCP_BINDING,
[Operations.LIST.value,
Operations.NSX_UPDATE.value]),
}
# Add supported NSX-V resources in this dictionary
nsxv_resources = {
constants.EDGES: Resource(constants.EDGES,
[Operations.NSX_LIST.value,
Operations.NEUTRON_LIST.value,
Operations.NSX_UPDATE.value]),
constants.BACKUP_EDGES: Resource(constants.BACKUP_EDGES,
[Operations.LIST.value,
Operations.CLEAN.value,
Operations.LIST_MISMATCHES.value,
Operations.FIX_MISMATCH.value]),
constants.ORPHANED_EDGES: Resource(constants.ORPHANED_EDGES,
[Operations.LIST.value,
Operations.CLEAN.value]),
constants.MISSING_EDGES: Resource(constants.MISSING_EDGES,
[Operations.LIST.value]),
constants.SPOOFGUARD_POLICY: Resource(constants.SPOOFGUARD_POLICY,
[Operations.LIST.value,
Operations.CLEAN.value]),
constants.DHCP_BINDING: Resource(constants.DHCP_BINDING,
[Operations.LIST.value,
Operations.NSX_UPDATE.value]),
constants.NETWORKS: Resource(constants.NETWORKS,
[Operations.LIST.value,
Operations.NSX_UPDATE.value]),
constants.SECURITY_GROUPS: Resource(constants.SECURITY_GROUPS,
[Operations.LIST.value,
Operations.FIX_MISMATCH.value]),
constants.FIREWALL_SECTIONS: Resource(constants.FIREWALL_SECTIONS,
[Operations.LIST.value,
Operations.LIST_MISMATCHES.value]),
constants.FIREWALL_NSX_GROUPS: Resource(
constants.FIREWALL_NSX_GROUPS, [Operations.LIST.value,
Operations.LIST_MISMATCHES.value]),
constants.METADATA: Resource(
constants.METADATA, [Operations.NSX_UPDATE.value,
Operations.NSX_UPDATE_SECRET.value]),
}
nsxv3_resources_names = map(lambda res: res.name, nsxv3_resources.itervalues())
nsxv_resources_names = map(lambda res: res.name, nsxv_resources.itervalues())
def get_resources(plugin_dir):
modules = glob.glob(plugin_dir + "/*.py")
return map(lambda module: os.path.splitext(os.path.basename(module))[0],
modules)
cli_opts = [cfg.StrOpt('fmt',
short='f',
default='psql',
choices=['psql', 'json'],
help='Supported output formats: json, psql'),
cfg.StrOpt('resource',
short='r',
choices=nsxv_resources_names + nsxv3_resources_names,
help='Supported list of resources: NSX-V3: %s '
'NSX-V: %s' % (', '.join(nsxv3_resources_names),
', '.join(nsxv_resources_names))),
cfg.StrOpt('operation',
short='o',
help='Supported list of operations: {}'
.format(', '.join(ops))),
cfg.BoolOpt('force',
default=False,
help='Enables \'force\' mode. No confirmations will '
'be made before deletions.'),
cfg.MultiStrOpt('property',
short='p',
help='Key-value pair containing the information '
'to be updated. For ex: key=value.')
]
def init_resource_plugin(plugin_name, plugin_dir):
plugin_resources = get_resources(plugin_dir)
for resource in plugin_resources:
if (resource != '__init__'):
importlib.import_module(
"." + resource,
"vmware_nsx.shell.admin.plugins."
"{}.resources".format(plugin_name))
def get_plugin_dir(plugin_name):
plugin_dir = (os.path.dirname(os.path.realpath(__file__)) +
"/admin/plugins")
return '{}/{}/resources'.format(plugin_dir, plugin_name)

2
vmware_nsx/tests/unit/nsx_v3/test_plugin.py
View File

@ -100,6 +100,8 @@ class NsxV3PluginTestCaseMixin(test_plugin.NeutronDbPluginV2TestCase,
mock_client_module.NSX3Client.return_value = mocked_client
_patch_object(nsx_plugin, 'nsx_client', new=mock_client_module)
_patch_object(nsx_plugin, 'nsx_cluster', new=mock_cluster_module)
self._mock_client_module = mock_client_module
self._mock_cluster_module = mock_cluster_module
# Mock the nsx v3 version
mock_nsxlib_get_version = mock.patch(

10
vmware_nsx/tests/unit/nsxlib/v3/nsxlib_testcase.py
View File

@ -94,6 +94,10 @@ class MemoryMockAPIProvider(nsx_cluster.AbstractHTTPProvider):
class NsxClientTestCase(NsxLibTestCase):
class MockBridge(object):
"""The MockBridge class is used to mock the nsxlib/v3/client.py file,
and includes the relevant functions & classes APIs
"""
def __init__(self, api_client):
self._client = api_client
@ -113,6 +117,12 @@ class NsxClientTestCase(NsxLibTestCase):
return nsx_client.update_resource(
resource, data, client=self._client)
def NSX3Client(self, cluster_api):
return self._client
def _set_default_api_cluster(self, cluster_api):
pass
class MockNSXClusteredAPI(nsx_cluster.NSXClusteredAPI):
def __init__(self, session_response=None):

0
vmware_nsx/tests/unit/shell/__init__.py Normal file
View File

142
vmware_nsx/tests/unit/shell/test_admin_utils.py Normal file
View File

@ -0,0 +1,142 @@
# 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
import logging
import mock
import six
from oslo_config import cfg
from neutron.callbacks import registry
from neutron.common import config as neutron_config
from neutron.tests import base
from neutron.tests.unit.db import test_db_base_plugin_v2 as test_n_plugin
from vmware_nsx._i18n import _
from vmware_nsx.common import config # noqa
from vmware_nsx.nsxlib.v3 import client as nsx_v3_client
from vmware_nsx.nsxlib.v3 import resources as nsx_v3_resources
from vmware_nsx.plugins.nsx_v3 import plugin as nsx_v3_plugin
from vmware_nsx.shell import resources
from vmware_nsx.tests import unit as vmware
from vmware_nsx.tests.unit.nsx_v.vshield import fake_vcns
from vmware_nsx.tests.unit.nsx_v3 import test_plugin as test_v3_plugin
LOG = logging.getLogger(__name__)
NSX_INI_PATH = vmware.get_fake_conf('nsx.ini.test')
BASE_CONF_PATH = vmware.get_fake_conf('neutron.conf.test')
@six.add_metaclass(abc.ABCMeta)
class AbstractTestAdminUtils(base.BaseTestCase):
def setUp(self):
cfg.CONF.register_cli_opts(resources.cli_opts)
super(AbstractTestAdminUtils, self).setUp()
# 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()
@abc.abstractmethod
def _init_mock_plugin(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 _test_resource(self, res_name, op, **kwargs):
# Must call the internal notify_loop in order to get the errors
errors = registry._get_callback_manager()._notify_loop(
res_name, op, 'nsxadmin', **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_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)
class TestNsxvAdminUtils(AbstractTestAdminUtils,
test_n_plugin.NeutronDbPluginV2TestCase):
def _init_mock_plugin(self):
mock_vcns = mock.patch(vmware.VCNS_NAME, autospec=True)
mock_vcns_instance = mock_vcns.start()
self.fc = fake_vcns.FakeVcns()
mock_vcns_instance.return_value = self.fc
self.addCleanup(self.fc.reset_all)
def _get_plugin_name(self):
return 'nsxv'
def test_nsxv_resources(self):
self._test_resources(resources.nsxv_resources)
# This is an example how to test a specific utility with arguments
def test_with_args(self):
args = {'property': ["xxx=yyy"]}
self._test_resource('networks', 'list', **args)
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):
self._patch_object(nsx_v3_client, 'NSX3Client',
new=self._mock_client_module)
self._patch_object(nsx_v3_plugin, 'nsx_cluster',
new=self._mock_cluster_module)
# mock resources
self._patch_object(nsx_v3_resources.LogicalPort,
'__init__', return_value=None)
self._patch_object(nsx_v3_resources.LogicalDhcpServer,
'__init__', return_value=None)
self._patch_object(nsx_v3_resources.LogicalRouter,
'__init__', return_value=None)
self._patch_object(nsx_v3_resources.SwitchingProfile,
'__init__', return_value=None)
self._patch_object(nsx_v3_resources.SwitchingProfile,
'find_by_display_name', return_value=None)
def _get_plugin_name(self):
return 'nsxv3'
def test_nsxv3_resources(self):
self._test_resources(resources.nsxv3_resources)