X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=NG-RO%2Fosm_ng_ro%2Ftests%2Ftest_ns_thread.py;fp=NG-RO%2Fosm_ng_ro%2Ftests%2Ftest_ns_thread.py;h=4e42b4f3b8f2c509eb746a3503c4d952c99fdbbc;hb=210ec7ca0dc0df57a9302486eedea70f9afafb18;hp=0914a0664543d25a6d5333694a63120ffabd4a94;hpb=ce61568d0ed0ed91ff684761b3e97102dd4e8883;p=osm%2FRO.git diff --git a/NG-RO/osm_ng_ro/tests/test_ns_thread.py b/NG-RO/osm_ng_ro/tests/test_ns_thread.py index 0914a066..4e42b4f3 100644 --- a/NG-RO/osm_ng_ro/tests/test_ns_thread.py +++ b/NG-RO/osm_ng_ro/tests/test_ns_thread.py @@ -27,6 +27,7 @@ from osm_ng_ro.ns_thread import ( VimInteractionMigration, VimInteractionNet, VimInteractionResize, + VimInteractionSharedVolume, ) from osm_ro_plugin.vimconn import VimConnConnectionException, VimConnException @@ -1450,6 +1451,218 @@ class TestVimInteractionNet(unittest.TestCase): instance.refresh(ro_task) +class TestVimInteractionSharedVolume(unittest.TestCase): + def setUp(self): + module_name = "osm_ro_plugin" + self.target_vim = MagicMock(name=f"{module_name}.vimconn.VimConnector") + self.task_depends = None + + patches = [patch(f"{module_name}.vimconn.VimConnector", self.target_vim)] + + # Enabling mocks and add cleanups + for mock in patches: + mock.start() + self.addCleanup(mock.stop) + + def test__new_shared_volume_ok(self): + """ + create a shared volume with attributes set in params + """ + db = "test_db" + logger = "test_logger" + my_vims = "test-vim" + db_vims = { + 0: { + "config": {}, + }, + } + + instance = VimInteractionSharedVolume(db, logger, my_vims, db_vims) + with patch.object(instance, "my_vims", [self.target_vim]), patch.object( + instance, "logger", logging + ), patch.object(instance, "db_vims", db_vims): + ro_task = { + "target_id": 0, + "tasks": { + "task_index_1": { + "target_id": 0, + "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": { + "shared_volume_data": { + "size": "10", + "name": "shared-volume", + "type": "multiattach", + } + }, + "find_params": {}, + "depends_on": "test_depends_on", + }, + }, + } + task_index = "task_index_1" + self.target_vim.new_shared_volumes.return_value = ("", "shared-volume") + result = instance.new(ro_task, task_index, self.task_depends) + self.assertEqual(result[0], "DONE") + self.assertEqual(result[1].get("vim_id"), "shared-volume") + self.assertEqual(result[1].get("created"), True) + self.assertEqual(result[1].get("vim_status"), "DONE") + + def test__new_shared_volume_failed(self): + """ + create a shared volume with attributes set in params failed + """ + db = "test_db" + logger = "test_logger" + my_vims = "test-vim" + db_vims = { + 0: { + "config": {}, + }, + } + + instance = VimInteractionSharedVolume(db, logger, my_vims, db_vims) + with patch.object(instance, "my_vims", [self.target_vim]), patch.object( + instance, "logger", logging + ), patch.object(instance, "db_vims", db_vims): + ro_task = { + "target_id": 0, + "tasks": { + "task_index_1": { + "target_id": 0, + "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": { + "shared_volume_data": { + "size": "10", + "name": "shared-volume", + "type": "multiattach", + } + }, + "find_params": {}, + "depends_on": "test_depends_on", + }, + }, + } + task_index = "task_index_1" + self.target_vim.new_shared_volumes.side_effect = VimConnException( + "Connection failed." + ) + result = instance.new(ro_task, task_index, self.task_depends) + self.assertEqual(result[0], "FAILED") + self.assertEqual(result[1].get("vim_message"), "Connection failed.") + self.assertEqual(result[1].get("created"), False) + self.assertEqual(result[1].get("vim_status"), "VIM_ERROR") + + def test__delete_shared_volume_ok(self): + """ + Delete a shared volume with attributes set in params + """ + db = "test_db" + logger = "test_logger" + my_vims = "test-vim" + db_vims = { + 0: { + "config": {}, + }, + } + + instance = VimInteractionSharedVolume(db, logger, my_vims, db_vims) + with patch.object(instance, "my_vims", [self.target_vim]), patch.object( + instance, "logger", logging + ), patch.object(instance, "db_vims", db_vims): + ro_task = { + "target_id": 0, + "tasks": { + "task_index_3": { + "target_id": 0, + "task_id": "123456:1", + }, + }, + "vim_info": { + "created": False, + "created_items": None, + "vim_id": "sample_shared_volume_id_3", + "vim_name": "sample_shared_volume_3", + "vim_status": None, + "vim_details": "some-details", + "vim_message": None, + "refresh_at": None, + }, + } + + task_index = "task_index_3" + self.target_vim.delete_shared_volumes.return_value = True + result = instance.delete(ro_task, task_index) + self.assertEqual(result[0], "DONE") + self.assertEqual(result[1].get("vim_id"), None) + self.assertEqual(result[1].get("created"), False) + self.assertEqual(result[1].get("vim_status"), "DELETED") + + def test__delete_shared_volume_failed(self): + """ + Delete a shared volume with attributes set in params failed + """ + db = "test_db" + logger = "test_logger" + my_vims = "test-vim" + db_vims = { + 0: { + "config": {}, + }, + } + + instance = VimInteractionSharedVolume(db, logger, my_vims, db_vims) + with patch.object(instance, "my_vims", [self.target_vim]), patch.object( + instance, "logger", logging + ), patch.object(instance, "db_vims", db_vims): + ro_task = { + "_id": "122436:1", + "target_id": 0, + "tasks": { + "task_index_3": { + "target_id": 0, + "task_id": "123456:1", + }, + }, + "vim_info": { + "created": False, + "created_items": None, + "vim_id": "sample_shared_volume_id_3", + "vim_name": "sample_shared_volume_3", + "vim_status": None, + "vim_details": "some-details", + "vim_message": None, + "refresh_at": None, + }, + } + + task_index = "task_index_3" + self.target_vim.delete_shared_volumes.side_effect = VimConnException( + "Connection failed." + ) + result = instance.delete(ro_task, task_index) + self.assertEqual(result[0], "FAILED") + self.assertEqual( + result[1].get("vim_message"), "Error while deleting: Connection failed." + ) + self.assertEqual(result[1].get("vim_status"), "VIM_ERROR") + + class TestVimInteractionAffinityGroup(unittest.TestCase): def setUp(self): module_name = "osm_ro_plugin"