VimInteractionMigration,
VimInteractionNet,
VimInteractionResize,
+ VimInteractionSharedVolume,
)
from osm_ro_plugin.vimconn import VimConnConnectionException, VimConnException
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"
projects / osm / RO.git / blobdiff