From: aticig Date: Sat, 1 Jan 2022 08:42:23 +0000 (+0300) Subject: Adding Unit tests for TestVimInteractionNet Class X-Git-Tag: v12.0.0rc1~42 X-Git-Url: https://osm.etsi.org/gitweb/?a=commitdiff_plain;h=refs%2Fchanges%2F02%2F11502%2F13;p=osm%2FRO.git Adding Unit tests for TestVimInteractionNet Class Unit tests are added for TestVIMInteractionNet Class which belongs to osm_ng_ro.ns_thread Change-Id: I66a096083667c0c25f663b15cd5080850b69212b Signed-off-by: aticig --- diff --git a/NG-RO/osm_ng_ro/tests/test_ns_thread.py b/NG-RO/osm_ng_ro/tests/test_ns_thread.py new file mode 100644 index 00000000..16b98e72 --- /dev/null +++ b/NG-RO/osm_ng_ro/tests/test_ns_thread.py @@ -0,0 +1,959 @@ +####################################################################################### +# Copyright ETSI Contributors and Others. +# +# 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 logging +from types import ModuleType +import unittest +from unittest.mock import Mock, patch + +from osm_ng_ro.ns_thread import VimInteractionNet + + +class TestVimInteractionNet(unittest.TestCase): + @classmethod + def setUpClass(cls): + module_name = "osm_ro_plugin" + osm_ro_plugin = ModuleType(module_name) + osm_ro_plugin.vimconn = Mock(name=module_name + ".vimconn") + osm_ro_plugin.vimconn.VimConnector = Mock( + name=module_name + "vimconn.VimConnector" + ) + osm_ro_plugin.vimconn.VimConnException = Mock( + name=module_name + ".vimconn.VimConnException" + ) + cls.target_vim = osm_ro_plugin.vimconn.VimConnector + cls.VimConnException = osm_ro_plugin.vimconn.VimConnException + cls.task_depends = None + + @classmethod + def tearDownClass(cls): + del cls.target_vim + del cls.task_depends + del cls.VimConnException + + def test__mgmt_net_id_in_find_params_mgmt_several_vim_nets(self): + """ + management_network_id in find_params.get('mgmt') + More than one network found in the VIM + """ + db = "test_db" + logger = "test_logger" + my_vims = "test-vim" + db_vims = { + "vim_openstack_2": { + "config": { + "management_network_id": "test_mgmt_id", + }, + }, + } + + instance = VimInteractionNet(db, logger, my_vims, db_vims) + with patch.object(instance, "my_vims", [self.target_vim]), patch.object( + instance, "logger", logging + ): + ro_task = { + "target_id": 0, + "tasks": { + "task_index_2": { + "target_id": "vim_openstack_2", + "action_id": "123456", + "nsr_id": "654321", + "task_id": "123456:1", + "status": "SCHEDULED", + "action": "CREATE", + "item": "test_item", + "target_record": "test_target_record", + "target_record_id": "test_target_record_id", + # values coming from extra_dict + "params": "", + "find_params": { + "mgmt": True, + "name": "some_mgmt_name", + }, + "depends_on": "test_depends_on", + }, + }, + } + + task_index = "task_index_2" + self.target_vim.get_network_list.return_value = [ + {"id": "existing_net_1"}, + {"id": "existing_net_2"}, + ] + with self.assertLogs() as captured: + result = instance.new(ro_task, task_index, self.task_depends) + self.assertEqual(len(captured.records), 1) + self.assertTrue( + "More than one network found with this criteria" + in captured.records[0].getMessage() + ) + self.assertEqual(captured.records[0].levelname, "ERROR") + self.assertEqual(result[0], "FAILED") + self.assertEqual(result[1].get("created"), False) + self.assertEqual(result[1].get("vim_status"), "VIM_ERROR") + + def test__mgmt_net_id_in_find_params_mgmt_no_vim_nets(self): + """ + management_network_id in find_params.get('mgmt') + The network could not be found in the VIM + """ + db = "test_db" + logger = "test_logger" + my_vims = "test-vim" + db_vims = { + "vim_openstack_3": { + "config": { + "management_network_id": "test_mgmt_id", + }, + }, + } + + instance = VimInteractionNet(db, logger, my_vims, db_vims) + with patch.object(instance, "my_vims", [self.target_vim]), patch.object( + instance, "logger", logging + ): + ro_task = { + "target_id": 0, + "tasks": { + "task_index_3": { + "target_id": "vim_openstack_3", + "action_id": "123456", + "nsr_id": "654321", + "task_id": "123456:1", + "status": "SCHEDULED", + "action": "CREATE", + "item": "test_item", + "target_record": "test_target_record", + "target_record_id": "test_target_record_id", + "params": "", + # values coming from extra_dict + "find_params": { + "mgmt": True, + "name": "some_mgmt_name", + }, + "depends_on": "test_depends_on", + }, + }, + } + + task_index = "task_index_3" + self.target_vim.get_network_list.return_value = [] + with self.assertLogs() as captured: + result = instance.new(ro_task, task_index, self.task_depends) + self.assertEqual(len(captured.records), 1) + self.assertTrue( + "Network not found with this criteria" + in captured.records[0].getMessage() + ) + self.assertEqual(captured.records[0].levelname, "ERROR") + self.assertEqual(result[0], "FAILED") + self.assertEqual(result[1].get("created"), False) + self.assertEqual(result[1].get("vim_status"), "VIM_ERROR") + + def test__mgmt_net_name_in_find_params_mgmt_several_vim_nets(self): + """ + management_network_name in find_params.get('mgmt') + More than one network found in the VIM + """ + db = "test_db" + logger = "test_logger" + my_vims = "test-vim" + db_vims = { + "vim_openstack_4": { + "config": { + "management_network_name": "test_mgmt_name", + }, + }, + } + + instance = VimInteractionNet(db, logger, my_vims, db_vims) + with patch.object(instance, "my_vims", [self.target_vim]), patch.object( + instance, "logger", logging + ): + ro_task = { + "target_id": 0, + "tasks": { + "task_index_4": { + "target_id": "vim_openstack_4", + "action_id": "123456", + "nsr_id": "654321", + "task_id": "123456:1", + "status": "SCHEDULED", + "action": "CREATE", + "item": "test_item", + "target_record": "test_target_record", + "target_record_id": "test_target_record_id", + # values coming from extra_dict + "params": "", + "find_params": { + "mgmt": True, + "name": "some_mgmt_name", + }, + "depends_on": "test_depends_on", + }, + }, + } + + task_index = "task_index_4" + self.target_vim.get_network_list.return_value = [ + {"id": "existing_net_1"}, + {"id": "existing_net_2"}, + ] + with self.assertLogs() as captured: + result = instance.new(ro_task, task_index, self.task_depends) + self.assertEqual(len(captured.records), 1) + self.assertTrue( + "More than one network found with this criteria" + in captured.records[0].getMessage() + ) + self.assertEqual(captured.records[0].levelname, "ERROR") + self.assertEqual(result[0], "FAILED") + self.assertEqual(result[1].get("created"), False) + self.assertEqual(result[1].get("vim_status"), "VIM_ERROR") + + def test__mgmt_net_name_in_find_params_mgmt_no_vim_nets(self): + """ + management_network_name in find_params.get('mgmt') + The network could not be found in the VIM + """ + db = "test_db" + logger = "test_logger" + my_vims = "test-vim" + db_vims = { + "vim_openstack_5": { + "config": { + "management_network_name": "test_mgmt_name", + }, + }, + } + + instance = VimInteractionNet(db, logger, my_vims, db_vims) + with patch.object(instance, "my_vims", [self.target_vim]), patch.object( + instance, "logger", logging + ): + ro_task = { + "target_id": 0, + "tasks": { + "task_index_5": { + "target_id": "vim_openstack_5", + "action_id": "123456", + "nsr_id": "654321", + "task_id": "123456:1", + "status": "SCHEDULED", + "action": "CREATE", + "item": "test_item", + "target_record": "test_target_record", + "target_record_id": "test_target_record_id", + # values coming from extra_dict + "params": "", + "find_params": { + "mgmt": True, + "name": "some_mgmt_name", + }, + "depends_on": "test_depends_on", + }, + }, + } + + task_index = "task_index_5" + self.target_vim.get_network_list.return_value = [] + with self.assertLogs() as captured: + result = instance.new(ro_task, task_index, self.task_depends) + self.assertEqual(len(captured.records), 1) + self.assertTrue( + "Network not found with this criteria" + in captured.records[0].getMessage() + ) + self.assertEqual(captured.records[0].levelname, "ERROR") + self.assertEqual(result[0], "FAILED") + self.assertEqual(result[1].get("created"), False) + self.assertEqual(result[1].get("vim_status"), "VIM_ERROR") + + def test__mgmt_net_name_in_find_params_filterdict_several_vim_nets(self): + """ + management_network_name in find_params.get('filterdict') + More than one network found in the VIM + """ + db = "test_db" + logger = "test_logger" + my_vims = "test-vim" + db_vims = { + "vim_openstack_6": { + "config": { + "management_network_name": "test_mgmt_name", + }, + }, + } + instance = VimInteractionNet(db, logger, my_vims, db_vims) + with patch.object(instance, "my_vims", [self.target_vim]), patch.object( + instance, "logger", logging + ): + ro_task = { + "target_id": 0, + "tasks": { + "task_index_6": { + "target_id": "vim_openstack_6", + "action_id": "123456", + "nsr_id": "654321", + "task_id": "123456:1", + "status": "SCHEDULED", + "action": "CREATE", + "item": "test_item", + "target_record": "test_target_record", + "target_record_id": "test_target_record_id", + # values coming from extra_dict + "params": "", + "find_params": { + "filter_dict": { + "name": "some-network-name", + }, + "mgmt": True, + "name": "some_mgmt_name", + }, + "depends_on": "test_depends_on", + }, + }, + } + + task_index = "task_index_6" + self.target_vim.get_network_list.return_value = [ + {"id": "existing_net_1"}, + {"id": "existing_net_2"}, + ] + with self.assertLogs() as captured: + result = instance.new(ro_task, task_index, self.task_depends) + self.assertEqual(len(captured.records), 1) + self.assertTrue( + "More than one network found with this criteria" + in captured.records[0].getMessage() + ) + self.assertEqual(captured.records[0].levelname, "ERROR") + self.assertEqual(result[0], "FAILED") + self.assertEqual(result[1].get("created"), False) + self.assertEqual(result[1].get("vim_status"), "VIM_ERROR") + + def test__mgmt_net_name_in_find_params_no_filterdict_no_mgmt(self): + """ + There is find_params in the task + No mgmt in find_params + No filter_dict in find_params + """ + db = "test_db" + logger = "test_logger" + my_vims = "test-vim" + db_vims = { + "vim_openstack_7": { + "config": { + "management_network_name": "test_mgmt_name", + }, + }, + } + instance = VimInteractionNet(db, logger, my_vims, db_vims) + with patch.object(instance, "my_vims", [self.target_vim]), patch.object( + instance, "logger", logging + ): + ro_task = { + "target_id": 0, + "tasks": { + "task_index_4": { + "target_id": "vim_openstack_7", + "action_id": "123456", + "nsr_id": "654321", + "task_id": "123456:1", + "status": "SCHEDULED", + "action": "CREATE", + "item": "test_item", + "target_record": "test_target_record", + "target_record_id": "test_target_record_id", + # values coming from extra_dict + "params": "", + "find_params": {"wrong_param": "wrong_value"}, + "depends_on": "test_depends_on", + }, + }, + } + + task_index = "task_index_4" + with self.assertLogs() as captured: + result = instance.new(ro_task, task_index, self.task_depends) + self.assertEqual(len(captured.records), 1) + self.assertTrue( + "Invalid find_params for new_net" + in captured.records[0].getMessage() + ) + self.assertEqual(captured.records[0].levelname, "ERROR") + self.assertEqual(result[0], "FAILED") + self.assertEqual(result[1].get("created"), False) + self.assertEqual(result[1].get("vim_status"), "VIM_ERROR") + + def test__mgmt_net_name_in_find_params_filterdict_no_vim_nets_params_in_task(self): + """ + management_network_name in find_params.get('filterdict') + The network could not be found in the VIM + There are items in the task.get(params) + """ + db = "test_db" + logger = "test_logger" + my_vims = "test-vim" + db_vims = { + "vim_openstack_8": { + "config": { + "management_network_name": "test_mgmt_name", + }, + }, + } + instance = VimInteractionNet(db, logger, my_vims, db_vims) + with patch.object(instance, "my_vims", [self.target_vim]), patch.object( + instance, "logger", logging + ): + ro_task = { + "target_id": 0, + "tasks": { + "task_index_8": { + "target_id": "vim_openstack_8", + "action_id": "123456", + "nsr_id": "654321", + "task_id": "123456:1", + "status": "SCHEDULED", + "action": "CREATE", + "item": "test_item", + "target_record": "test_target_record", + "target_record_id": "test_target_record_id", + # values coming from extra_dict + "params": "test_params", + "find_params": { + "filter_dict": { + "name": "some-network-name", + }, + "mgmt": True, + "name": "some_mgmt_name", + }, + "depends_on": "test_depends_on", + }, + }, + } + + task_index = "task_index_8" + self.target_vim.get_network_list.return_value = [] + result = instance.new(ro_task, task_index, self.task_depends) + self.assertEqual(result[0], "BUILD") + self.assertEqual(result[1].get("created"), False) + self.assertEqual(result[1].get("vim_id"), None) + self.assertEqual(result[1].get("created_items"), {}) + self.assertEqual(result[1].get("vim_status"), "BUILD") + + def test__mgmt_net_name_in_find_params_filterdict_no_vim_nets(self): + """ + management_network_name in find_params.get('filterdict') + Any network could not be found in the VIM + """ + db = "test_db" + logger = "test_logger" + my_vims = "test-vim" + db_vims = { + "vim_openstack_9": { + "config": { + "management_network_name": "test_mgmt_name", + }, + }, + } + instance = VimInteractionNet(db, logger, my_vims, db_vims) + with patch.object(instance, "my_vims", [self.target_vim]), patch.object( + instance, "logger", logging + ): + ro_task = { + "target_id": 0, + "tasks": { + "task_index_9": { + "target_id": "vim_openstack_9", + "action_id": "123456", + "nsr_id": "654321", + "task_id": "123456:1", + "status": "SCHEDULED", + "action": "CREATE", + "item": "test_item", + "target_record": "test_target_record", + "target_record_id": "test_target_record_id", + # values coming from extra_dict + "params": "", + "find_params": { + "filter_dict": { + "name": "some-network-name", + }, + "mgmt": True, + "name": "some_mgmt_name", + }, + "depends_on": "test_depends_on", + }, + }, + } + + task_index = "task_index_9" + self.target_vim.get_network_list.return_value = [] + with self.assertLogs() as captured: + result = instance.new(ro_task, task_index, self.task_depends) + self.assertEqual(len(captured.records), 1) + self.assertTrue( + "Network not found with this criteria" + in captured.records[0].getMessage() + ) + self.assertEqual(captured.records[0].levelname, "ERROR") + self.assertEqual(result[0], "FAILED") + self.assertEqual(result[1].get("created"), False) + self.assertEqual(result[1].get("vim_status"), "VIM_ERROR") + + def test__mgmt_net_name_in_find_params_mgmt_no_config_one_vim_net(self): + """ + name in find_params + management_network_name is not in db_vims.get('config') + One network found in the VIM + """ + db = "test_db" + logger = "test_logger" + my_vims = "test-vim" + db_vims = { + "vim_openstack_10": { + "config": {}, + }, + } + instance = VimInteractionNet(db, logger, my_vims, db_vims) + with patch.object(instance, "my_vims", [self.target_vim]), patch.object( + instance, "logger", logging + ): + ro_task = { + "target_id": 0, + "tasks": { + "task_index_2": { + "target_id": "vim_openstack_10", + "action_id": "123456", + "nsr_id": "654321", + "task_id": "123456:1", + "status": "SCHEDULED", + "action": "CREATE", + "item": "test_item", + "target_record": "test_target_record", + "target_record_id": "test_target_record_id", + # values coming from extra_dict + "params": "", + "find_params": { + "mgmt": True, + "name": "some_mgmt_name", + }, + "depends_on": "test_depends_on", + }, + }, + } + + task_index = "task_index_2" + self.target_vim.get_network_list.return_value = [ + {"id": "4d83a7c9-3ef4-4a45-99c8-aca3550490dd"} + ] + result = instance.new(ro_task, task_index, self.task_depends) + self.assertEqual(result[0], "BUILD") + self.assertEqual(result[1].get("created"), False) + self.assertEqual(result[1].get("vim_status"), "BUILD") + + def test__params_in_task_no_find_params(self): + """ + params in the task + find_params does not exist in the task + """ + db = "test_db" + logger = "test_logger" + my_vims = "test-vim" + db_vims = { + "vim_openstack_11": { + "config": { + "management_network_name": "test_mgmt_name", + }, + }, + } + instance = VimInteractionNet(db, logger, my_vims, db_vims) + with patch.object(instance, "my_vims", [self.target_vim]), patch.object( + instance, "logger", logging + ): + ro_task = { + "target_id": 0, + "tasks": { + "task_index_11": { + "target_id": "vim_openstack_11", + "action_id": "123456", + "nsr_id": "654321", + "task_id": "123456:1", + "status": "SCHEDULED", + "action": "CREATE", + "item": "test_item", + "target_record": "test_target_record", + "target_record_id": "test_target_record_id", + # values coming from extra_dict + "params": { + "net_name": "test-network", + "net_type": "vlan", + }, + "depends_on": "test_depends_on", + }, + }, + } + + task_index = "task_index_11" + self.target_vim.new_network.return_value = "sample_net_id", { + "item1": "sample_created_item" + } + result = instance.new(ro_task, task_index, self.task_depends) + self.assertEqual(result[0], "BUILD") + self.assertEqual(result[1].get("vim_id"), "sample_net_id") + self.assertEqual(result[1].get("created"), True) + self.assertEqual( + result[1].get("created_items"), {"item1": "sample_created_item"} + ) + self.assertEqual(result[1].get("vim_status"), "BUILD") + + def test__no_params_in_task_no_find_params(self): + """ + empty params in the task + find_params does not exist in the task + """ + db = "test_db" + logger = "test_logger" + my_vims = "test-vim" + db_vims = { + "vim_openstack_13": { + "config": { + "management_network_name": "test_mgmt_name", + }, + }, + } + instance = VimInteractionNet(db, logger, my_vims, db_vims) + with patch.object(instance, "my_vims", [self.target_vim]), patch.object( + instance, "logger", logging + ): + ro_task = { + "target_id": 0, + "tasks": { + "task_index_12": { + "target_id": "vim_openstack_13", + "action_id": "123456", + "nsr_id": "654321", + "task_id": "123456:1", + "status": "SCHEDULED", + "action": "CREATE", + "item": "test_item", + "target_record": "test_target_record", + "target_record_id": "test_target_record_id", + # values coming from extra_dict + "params": "", + "depends_on": "test_depends_on", + }, + }, + } + + task_index = "task_index_12" + with self.assertRaises(TypeError): + instance.new(ro_task, task_index, self.task_depends) + + def test__refresh_ro_task_vim_status_active(self): + """ + vim_info.get('status') is ACTIVE + """ + db = "test_db" + logger = "test_logger" + my_vims = "test-vim" + db_vims = { + "vim_openstack_1": { + "config": {}, + }, + } + instance = VimInteractionNet(db, logger, my_vims, db_vims) + with patch.object( + instance, "my_vims", {"vim_openstack_1": self.target_vim} + ), patch.object(instance, "logger", logging): + ro_task = { + "_id": "122436:1", + "locked_by": None, + "locked_at": 0.0, + "target_id": "vim_openstack_1", + "vim_info": { + "created": False, + "created_items": None, + "vim_id": "test-vim-id", + "vim_name": "test-vim", + "vim_status": None, + "vim_details": "some-details", + "refresh_at": None, + }, + "modified_at": 1637324200.994312, + "created_at": 1637324200.994312, + "to_check_at": 1637324200.994312, + "tasks": {}, + } + + self.target_vim.refresh_nets_status.return_value = { + "test-vim-id": { + "vim_info": "some-details", + "status": "ACTIVE", + "name": "test-vim", + "error_msg": "", + } + } + task_status = "DONE" + ro_vim_item_update = { + "vim_status": "ACTIVE", + } + result = instance.refresh(ro_task) + self.assertEqual(result[0], task_status) + self.assertDictEqual(result[1], ro_vim_item_update) + + def test__refresh_ro_task_vim_status_build(self): + """ + vim_info.get('status') is BUILD + """ + db = "test_db" + logger = "test_logger" + my_vims = "test-vim" + db_vims = { + "vim_openstack_1": { + "config": {}, + }, + } + instance = VimInteractionNet(db, logger, my_vims, db_vims) + with patch.object( + instance, "my_vims", {"vim_openstack_1": self.target_vim} + ), patch.object(instance, "logger", logging): + ro_task = { + "_id": "122436:1", + "locked_by": None, + "locked_at": 0.0, + "target_id": "vim_openstack_1", + "vim_info": { + "created": False, + "created_items": None, + "vim_id": "test-vim-id", + "vim_name": "test-vim", + "vim_status": "BUILD", + "vim_details": "", + "refresh_at": None, + }, + "modified_at": 1637324200.994312, + "created_at": 1637324200.994312, + "to_check_at": 1637324200.994312, + "tasks": {}, + } + + self.target_vim.refresh_nets_status.return_value = { + "test-vim-id": { + "vim_info": "some-details", + "status": "BUILD", + "name": "other-vim", + "error_msg": "", + } + } + task_status = "BUILD" + ro_vim_item_update = { + "vim_name": "other-vim", + "vim_details": "some-details", + } + result = instance.refresh(ro_task) + self.assertEqual(result[0], task_status) + self.assertDictEqual(result[1], ro_vim_item_update) + + def test__refresh_ro_task_vim_status_error(self): + """ + vim_info.get('status') is ERROR + """ + db = "test_db" + logger = "test_logger" + my_vims = "test-vim" + db_vims = { + "vim_openstack_1": { + "config": {}, + }, + } + instance = VimInteractionNet(db, logger, my_vims, db_vims) + with patch.object( + instance, "my_vims", {"vim_openstack_1": self.target_vim} + ), patch.object(instance, "logger", logging): + ro_task = { + "_id": "122436:1", + "locked_by": None, + "locked_at": 0.0, + "target_id": "vim_openstack_1", + "vim_info": { + "created": False, + "created_items": None, + "vim_id": "test-vim-id", + "vim_name": "test-vim", + "vim_status": "BUILD", + "vim_details": "", + "refresh_at": None, + }, + "modified_at": 1637324200.994312, + "created_at": 1637324200.994312, + "to_check_at": 1637324200.994312, + "tasks": {}, + } + + self.target_vim.refresh_nets_status.return_value = { + "test-vim-id": { + "vim_info": "some-details", + "status": "ERROR", + "name": "test-vim", + "error_msg": "some error message", + } + } + task_status = "FAILED" + ro_vim_item_update = { + "vim_status": "ERROR", + "vim_details": "some error message", + } + result = instance.refresh(ro_task) + self.assertEqual(result[0], task_status) + self.assertDictEqual(result[1], ro_vim_item_update) + + def test__refresh_ro_task_VimConnException_occured(self): + """ + vimconn.VimConnException has occured + """ + db = "test_db" + logger = "test_logger" + my_vims = "test-vim" + db_vims = { + "vim_openstack_1": { + "config": {}, + }, + } + instance = VimInteractionNet(db, logger, my_vims, db_vims) + with patch.object( + instance, "my_vims", {"vim_openstack_1": self.target_vim} + ), patch.object(instance, "logger", logging): + ro_task = { + "_id": "122436:1", + "locked_by": None, + "locked_at": 0.0, + "target_id": "vim_openstack_1", + "vim_info": { + "created": False, + "created_items": None, + "vim_id": "test-vim-id", + "vim_name": "test-vim", + "vim_status": "BUILD", + "vim_details": "", + "refresh_at": None, + }, + "modified_at": 1637324200.994312, + "created_at": 1637324200.994312, + "to_check_at": 1637324200.994312, + "tasks": {}, + } + self.target_vim.refresh_nets_status.side_effect = Mock( + side_effect=self.VimConnException("VimConnException occured") + ) + with self.assertRaises(TypeError): + instance.refresh(ro_task) + self.target_vim.refresh_nets_status.side_effect = None + + def test__refresh_ro_task_vim_status_deleted(self): + """ + vim_info.get('status') is DELETED + """ + db = "test_db" + logger = "test_logger" + my_vims = "test-vim" + db_vims = { + "vim_openstack_1": { + "config": {}, + }, + } + instance = VimInteractionNet(db, logger, my_vims, db_vims) + with patch.object( + instance, "my_vims", {"vim_openstack_1": self.target_vim} + ), patch.object(instance, "logger", logging): + ro_task = { + "_id": "122436:1", + "locked_by": None, + "locked_at": 0.0, + "target_id": "vim_openstack_1", + "vim_info": { + "created": False, + "created_items": None, + "vim_id": "test-vim-id", + "vim_name": "test-vim", + "vim_status": "BUILD", + "vim_details": "", + "refresh_at": None, + }, + "modified_at": 163724200.994312, + "created_at": 1637324200.994312, + "to_check_at": 1637324200.994312, + "tasks": {}, + } + self.target_vim.refresh_nets_status.return_value = { + "test-vim-id": { + "vim_info": "some-details", + "status": "DELETED", + "name": "test-vim", + "error_msg": "some error message", + } + } + task_status = "FAILED" + ro_vim_item_update = { + "vim_status": "DELETED", + "vim_details": "Deleted externally", + "vim_id": None, + } + result = instance.refresh(ro_task) + self.assertEqual(result[0], task_status) + self.assertDictEqual(result[1], ro_vim_item_update) + + def test__refresh_ro_task_empty_vim_dict(self): + """ + vim_dict does not include vim_id key + Raises KeyError + """ + db = "test_db" + logger = "test_logger" + my_vims = "test-vim" + db_vims = { + "vim_openstack_2": { + "config": {}, + }, + } + instance = VimInteractionNet(db, logger, my_vims, db_vims) + with patch.object( + instance, "my_vims", {"vim_openstack_2": self.target_vim} + ), patch.object(instance, "logger", logging): + ro_task = { + "_id": "128436:1", + "locked_by": None, + "locked_at": 0.0, + "target_id": "vim_openstack_2", + "vim_info": { + "created": False, + "created_items": None, + "vim_id": "test-vim-id", + "vim_name": "test-vim", + "vim_status": "BUILD", + "vim_details": "", + "refresh_at": None, + }, + "modified_at": 163724211.994312, + "created_at": 1637324211.994312, + "to_check_at": 1637324211.994312, + "tasks": {}, + } + self.target_vim.refresh_nets_status.return_value = {} + with self.assertRaises(KeyError): + instance.refresh(ro_task) diff --git a/releasenotes/notes/test_vim_interaction_net-d6ba897d64b6ddd0.yaml b/releasenotes/notes/test_vim_interaction_net-d6ba897d64b6ddd0.yaml new file mode 100644 index 00000000..59d79923 --- /dev/null +++ b/releasenotes/notes/test_vim_interaction_net-d6ba897d64b6ddd0.yaml @@ -0,0 +1,25 @@ +####################################################################################### +# Copyright ETSI Contributors and Others. +# +# 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. +####################################################################################### +--- +other: + - | + TestVimInteractionNet is one of the Class in ns_thread.py under NG-RO. + Unit tests are added for the methods of TestVimInteractionNet Class which + checks the general functionality of methods. That will be helpful to improve + the test coverage and code quality. + +