trivial: Remove unused inner functions of UT and FT

Picked up unused fuctions with vulture [1].
In this patch, from the picked up functions,
the inner-functions prefixed with '_' are deleted as dead code.
Only target the inner-function of unittest and functional-test.
Other functions are left because it's difficult
to determine if they should be mechanically deleted.

[1] https://pypi.org/project/vulture/

Change-Id: I0deb0bdc9e7a7dd1cf7f98f22686ca7727088e22
Closes-Bug: #1879871
changes/07/729507/10
Hiroo Kitamura 3 years ago
parent 41a21bc807
commit 527ccfc3a7
  1. 123
      tacker/tests/functional/nfvo/test_nfvo.py
  2. 115
      tacker/tests/unit/api/test_v2_extension.py
  3. 12
      tacker/tests/unit/nfvo/test_nfvo_plugin.py
  4. 53
      tacker/tests/unit/vnfm/infra_drivers/openstack/test_openstack.py
  5. 6
      tacker/tests/unit/vnfm/infra_drivers/openstack/test_openstack_driver.py
  6. 11
      tacker/tests/unit/vnfm/test_plugin.py

@ -72,129 +72,6 @@ class NsdTestCreate(base.BaseTackerTest):
self.verify_vnfd_events(vnfd_id, evt_constants.RES_EVT_DELETE,
evt_constants.RES_EVT_NA_STATE)
def _wait_until_ns_status(self, ns_id, target_status, timeout,
sleep_interval):
start_time = int(time.time())
while True:
ns_result = self.client.show_ns(ns_id)
status = ns_result['ns']['status']
if (status == target_status) or (
(int(time.time()) - start_time) > timeout):
break
time.sleep(sleep_interval)
self.assertEqual(status, target_status,
"ns %(ns_id)s with status %(status)s is"
" expected to be %(target)s" %
{"ns_id": ns_id, "status": status,
"target": target_status})
def _wait_until_ns_delete(self, ns_id, timeout):
start_time = int(time.time())
while True:
try:
ns_result = self.client.show_ns(ns_id)
time.sleep(2)
except Exception:
return
status = ns_result['ns']['status']
if (status != 'PENDING_DELETE') or ((
int(time.time()) - start_time) > timeout):
raise Exception("Failed with status: %s" % status)
def _test_create_delete_ns(self, nsd_file, ns_name,
template_source='onboarded'):
vnfd1_id = self._test_create_tosca_vnfd(
'test-ns-vnfd1.yaml',
'test-ns-vnfd1')
vnfd2_id = self._test_create_tosca_vnfd(
'test-ns-vnfd2.yaml',
'test-ns-vnfd2')
if template_source == 'onboarded':
nsd_id = self._test_create_nsd(
nsd_file,
'test-ns-nsd')
ns_arg = {'ns': {
'nsd_id': nsd_id,
'name': ns_name,
'attributes': {"param_values": {
"nsd": {
"vl2_name": "net0",
"vl1_name": "net_mgmt"}}}}}
ns_instance = self.client.create_ns(body=ns_arg)
ns_id = ns_instance['ns']['id']
if template_source == 'inline':
input_yaml = read_file(nsd_file)
template = yaml.safe_load(input_yaml)
ns_arg = {'ns': {
'name': ns_name,
'attributes': {"param_values": {
"nsd": {
"vl2_name": "net0",
"vl1_name": "net_mgmt"}}},
'nsd_template': template}}
ns_instance = self.client.create_ns(body=ns_arg)
ns_id = ns_instance['ns']['id']
self._wait_until_ns_status(ns_id, 'ACTIVE',
constants.NS_CREATE_TIMEOUT,
constants.ACTIVE_SLEEP_TIME)
ns_show_out = self.client.show_ns(ns_id)['ns']
self.assertIsNotNone(ns_show_out['mgmt_ip_addresses'])
try:
self.client.delete_ns(ns_id)
except Exception as e:
print("Exception:", e)
assert False, "ns Delete failed"
if template_source == 'onboarded':
self._wait_until_ns_delete(ns_id, constants.NS_DELETE_TIMEOUT)
self._test_delete_nsd(nsd_id)
self._test_delete_vnfd(vnfd1_id)
self._test_delete_vnfd(vnfd2_id)
def _test_create_delete_ns_vnffg(self, nsd_file, ns_name,
source_port_id, dest_ip_prefix):
vnfd1_id = self._test_create_tosca_vnfd(
'tosca-ns-vnffg-vnfd1-sample.yaml',
'sample-vnfd1')
vnfd2_id = self._test_create_tosca_vnfd(
'tosca-ns-vnffg-vnfd2-sample.yaml',
'sample-vnfd2')
nsd_id = self._test_create_nsd(nsd_file, ns_name)
ns_arg = {'ns': {
'nsd_id': nsd_id,
'name': ns_name,
'attributes': {"param_values": {
"nsd": {
"vl2_name": "net0",
"vl1_name": "net_mgmt",
"net_src_port_id": source_port_id,
"ip_dest_prefix": dest_ip_prefix}}}}}
ns_instance = self.client.create_ns(body=ns_arg)
ns_id = ns_instance['ns']['id']
self._wait_until_ns_status(ns_id, 'ACTIVE',
constants.NS_CREATE_TIMEOUT,
constants.ACTIVE_SLEEP_TIME)
ns_show_out = self.client.show_ns(ns_id)['ns']
self.assertIsNotNone(ns_show_out['mgmt_ip_addresses'])
vnffg = self.client.list_vnffgs()
self.assertIsNotNone(vnffg)
try:
self.client.delete_ns(ns_id)
except Exception as e:
print("Exception:", e)
assert False, "ns Delete failed"
self._wait_until_ns_delete(ns_id, constants.NS_DELETE_TIMEOUT)
self._test_delete_nsd(nsd_id)
self._test_delete_vnfd(vnfd1_id)
self._test_delete_vnfd(vnfd2_id)
def test_create_delete_nsd(self):
vnfd1_id = self._test_create_tosca_vnfd(
'test-nsd-vnfd1.yaml',

@ -1,115 +0,0 @@
# Copyright 2014 Intel Corporation.
# 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.
from unittest import mock
from oslo_config import cfg
from oslo_utils import uuidutils
from webob import exc
import webtest
from tacker.api import extensions
from tacker.api.v1 import attributes
from tacker.tests.unit.api import test_extensions
from tacker.tests.unit.api.v2 import test_api_v2
from tacker.tests.unit import testlib_api
class ExtensionTestCase(testlib_api.WebTestCase):
def _resotre_attr_map(self):
attributes.RESOURCE_ATTRIBUTE_MAP = self._saved_attr_map
def _setUpExtension(self, plugin, service_type,
resource_attribute_map, extension_class,
resource_prefix, plural_mappings=None,
translate_resource_name=False,
allow_pagination=False, allow_sorting=False,
supported_extension_aliases=None,
):
self._resource_prefix = resource_prefix
self._plural_mappings = plural_mappings or {}
self._translate_resource_name = translate_resource_name
# Ensure existing ExtensionManager is not used
extensions.PluginAwareExtensionManager._instance = None
# Save the global RESOURCE_ATTRIBUTE_MAP
self._saved_attr_map = attributes.RESOURCE_ATTRIBUTE_MAP.copy()
# Restore the global RESOURCE_ATTRIBUTE_MAP
self.addCleanup(self._resotre_attr_map)
# Create the default configurations
self.config_parse()
# just stubbing core plugin with plugin
self.setup_coreplugin(plugin)
cfg.CONF.set_override('core_plugin', plugin)
if service_type:
cfg.CONF.set_override('service_plugins', [plugin])
self._plugin_patcher = mock.patch(plugin, autospec=True)
self.plugin = self._plugin_patcher.start()
instance = self.plugin.return_value
if service_type:
instance.get_plugin_type.return_value = service_type
if supported_extension_aliases is not None:
instance.supported_extension_aliases = supported_extension_aliases
if allow_pagination:
cfg.CONF.set_override('allow_pagination', True)
# instance.__native_pagination_support = True
native_pagination_attr_name = ("_%s__native_pagination_support"
% instance.__class__.__name__)
setattr(instance, native_pagination_attr_name, True)
if allow_sorting:
cfg.CONF.set_override('allow_sorting', True)
# instance.__native_sorting_support = True
native_sorting_attr_name = ("_%s__native_sorting_support"
% instance.__class__.__name__)
setattr(instance, native_sorting_attr_name, True)
class ExtensionTestExtensionManager(object):
def get_resources(self):
# Add the resources to the global attribute map
# This is done here as the setup process won't
# initialize the main API router which extends
# the global attribute map
attributes.RESOURCE_ATTRIBUTE_MAP.update(
resource_attribute_map)
return extension_class.get_resources()
def get_actions(self):
return []
def get_request_extensions(self):
return []
ext_mgr = ExtensionTestExtensionManager()
self.ext_mdw = test_extensions.setup_extensions_middleware(ext_mgr)
self.api = webtest.TestApp(self.ext_mdw)
def _test_entity_delete(self, entity):
"""Does the entity deletion based on naming convention."""
entity_id = uuidutils.generate_uuid()
path = self._resource_prefix + '/' if self._resource_prefix else ''
path += self._plural_mappings.get(entity, entity + 's')
if self._translate_resource_name:
path = path.replace('_', '-')
res = self.api.delete(
test_api_v2._get_path(path, id=entity_id, fmt=self.fmt))
delete_entity = getattr(self.plugin.return_value, "delete_" + entity)
delete_entity.assert_called_with(mock.ANY, entity_id)
self.assertEqual(exc.HTTPNoContent.code, res.status_int)

@ -448,18 +448,6 @@ class TestNfvoPlugin(db_base.SqlTestCase):
session.flush()
return vnffg_template
def _insert_dummy_vnffg_str_param_template(self):
session = self.context.session
vnffg_template = vnffg_db.VnffgTemplate(
id='eb094833-995e-49f0-a047-dfb56aaf7c4e',
tenant_id='ad7ebc56538745a08ef7c5e97f8bd437',
name='fake_template',
description='fake_template_description',
template={u'vnffgd': utils.vnffgd_tosca_str_param_template})
session.add(vnffg_template)
session.flush()
return vnffg_template
def _insert_dummy_vnffg_multi_param_template(self):
session = self.context.session
vnffg_template = vnffg_db.VnffgTemplate(

@ -147,42 +147,6 @@ class TestOpenStack(base.TestCase):
'tenant_id': 'ad7ebc56538745a08ef7c5e97f8bd437',
'id': 'fb048660-dc1b-4f0f-bd89-b023666650ec'}
def _get_expected_fields(self):
return {'stack_name':
'test_openwrt_eb84260e-5ff7-4332-b032-50a14d6c1123',
'template': self.hot_template}
def _get_expected_fields_user_data(self):
return {'stack_name':
'test_userdata_18685f68-2b2a-4185-8566-74f54e548811',
'template': self.hot_param_template}
def _get_expected_fields_ipaddr_data(self):
return {'stack_name': 'test_ip_d1337add-d5a1-4fd4-9447-bb9243c8460b',
'template': self.hot_ipparam_template}
def _get_expected_vnf_wait_obj(self, param_values=''):
return {'status': 'PENDING_CREATE',
'instance_id': None,
'name': u'test_openwrt',
'tenant_id': u'ad7ebc56538745a08ef7c5e97f8bd437',
'vnfd_id': u'eb094833-995e-49f0-a047-dfb56aaf7c4e',
'vnfd': {
'service_types': [{
'service_type': u'vnfd',
'id': u'4a4c2d44-8a52-4895-9a75-9d1c76c3e738'}],
'description': u'OpenWRT with services',
'tenant_id': u'ad7ebc56538745a08ef7c5e97f8bd437',
'mgmt_driver': u'openwrt',
'attributes': {u'vnfd': self.tosca_vnfd_openwrt},
'id': u'fb048660-dc1b-4f0f-bd89-b023666650ec',
'name': u'OpenWRT'},
'mgmt_ip_address': '{"vdu1": "192.168.120.31"}',
'service_context': [],
'attributes': {u'param_values': param_values},
'id': 'eb84260e-5ff7-4332-b032-50a14d6c1123',
'description': u'OpenWRT with services'}
def _get_expected_vnf_update_obj(self):
return {'status': 'PENDING_CREATE', 'instance_id': None, 'name':
u'test_openwrt', 'tenant_id':
@ -200,23 +164,6 @@ class TestOpenStack(base.TestCase):
'id': 'eb84260e-5ff7-4332-b032-50a14d6c1123', 'description':
u'OpenWRT with services'}
def _get_expected_vnf_update_param_obj(self):
return {'status': 'PENDING_CREATE', 'instance_id': None, 'name':
u'test_openwrt', 'tenant_id':
u'ad7ebc56538745a08ef7c5e97f8bd437', 'vnfd_id':
u'eb094833-995e-49f0-a047-dfb56aaf7c4e', 'vnfd': {
'service_types': [{'service_type': u'vnfd', 'id':
u'4a4c2d44-8a52-4895-9a75-9d1c76c3e738'}], 'description':
u'OpenWRT with services', 'tenant_id':
u'ad7ebc56538745a08ef7c5e97f8bd437', 'mgmt_driver': u'openwrt',
'attributes': {u'vnfd': self.tosca_vnfd_openwrt_param},
'id': u'fb048660-dc1b-4f0f-bd89-b023666650ec', 'name':
u'openwrt_services'}, 'mgmt_url': None, 'service_context': [],
'attributes': {'heat_template': utils.hot_data,
'param_values': utils.update_param_data},
'id': 'eb84260e-5ff7-4332-b032-50a14d6c1123', 'description':
u'OpenWRT with services'}
def _get_expected_vnf_update_new_param_obj(self):
return {'status': 'PENDING_CREATE', 'instance_id': None, 'name':
u'test_openwrt', 'tenant_id':

@ -1011,12 +1011,6 @@ class TestOpenStack(base.FixturedTestCase):
self.assertEqual(import_image_url.call_count, 1)
self.assertEqual(get_image_url.call_count, 10)
def _exception_response_in_import_image(self):
url = os.path.join(self.glance_url, 'images', uuidsentinel.image_id,
'import')
return self.requests_mock.register_uri(
'POST', url, exc=requests.exceptions.ConnectTimeout)
def _response_in_delete_image(self, resource_id, exception=False):
# response for glance_client's delete()
url = os.path.join(

@ -245,17 +245,6 @@ class TestVNFMPlugin(db_base.SqlTestCase):
session.flush()
return vnf_template
def _insert_dummy_vnfd_attributes(self, template):
session = self.context.session
vnfd_attr = vnfm_db.VNFDAttribute(
id='eb094833-995e-49f0-a047-dfb56aaf7c4e',
vnfd_id='eb094833-995e-49f0-a047-dfb56aaf7c4e',
key='vnfd',
value=template)
session.add(vnfd_attr)
session.flush()
return vnfd_attr
def _insert_dummy_vnf(self, status="ACTIVE"):
session = self.context.session
vnf_db = vnfm_db.VNF(

Loading…
Cancel
Save