diff --git a/tacker/tests/functional/nfvo/test_nfvo.py b/tacker/tests/functional/nfvo/test_nfvo.py index bb797f81b..ccc7e2c77 100644 --- a/tacker/tests/functional/nfvo/test_nfvo.py +++ b/tacker/tests/functional/nfvo/test_nfvo.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', diff --git a/tacker/tests/unit/api/test_v2_extension.py b/tacker/tests/unit/api/test_v2_extension.py deleted file mode 100644 index 328f8c64f..000000000 --- a/tacker/tests/unit/api/test_v2_extension.py +++ /dev/null @@ -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) diff --git a/tacker/tests/unit/nfvo/test_nfvo_plugin.py b/tacker/tests/unit/nfvo/test_nfvo_plugin.py index 644400cab..634d0dfa9 100644 --- a/tacker/tests/unit/nfvo/test_nfvo_plugin.py +++ b/tacker/tests/unit/nfvo/test_nfvo_plugin.py @@ -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( diff --git a/tacker/tests/unit/vnfm/infra_drivers/openstack/test_openstack.py b/tacker/tests/unit/vnfm/infra_drivers/openstack/test_openstack.py index 8281fc271..6cb4bf04b 100644 --- a/tacker/tests/unit/vnfm/infra_drivers/openstack/test_openstack.py +++ b/tacker/tests/unit/vnfm/infra_drivers/openstack/test_openstack.py @@ -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': diff --git a/tacker/tests/unit/vnfm/infra_drivers/openstack/test_openstack_driver.py b/tacker/tests/unit/vnfm/infra_drivers/openstack/test_openstack_driver.py index 667f1a0c0..3febe14af 100644 --- a/tacker/tests/unit/vnfm/infra_drivers/openstack/test_openstack_driver.py +++ b/tacker/tests/unit/vnfm/infra_drivers/openstack/test_openstack_driver.py @@ -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( diff --git a/tacker/tests/unit/vnfm/test_plugin.py b/tacker/tests/unit/vnfm/test_plugin.py index 666b7e71e..f954668c7 100644 --- a/tacker/tests/unit/vnfm/test_plugin.py +++ b/tacker/tests/unit/vnfm/test_plugin.py @@ -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(