"id": "persistent-root-volume",
"type-of-storage": "persistent-storage:persistent-storage",
"size-of-storage": "10",
+ "vdu-storage-requirements": [
+ {"key": "keep-volume", "value": "true"},
+ ],
},
{
"id": "ephemeral-volume",
"id": "persistent-root-volume",
"size-of-storage": "10",
"type-of-storage": "persistent-storage:persistent-storage",
+ "vdu-storage-requirements": [
+ {"key": "keep-volume", "value": "true"},
+ ],
},
{
"id": "ephemeral-volume",
self.ns = Ns()
self.logger = CopyingMock(autospec=True)
- def test_find_persistent_root_volumes_empty_instantiation_vol_list(self):
+ @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
+ def test_find_persistent_root_volumes_empty_instantiation_vol_list(
+ self, mock_volume_keeping_required
+ ):
"""Find persistent root volume, instantiation_vol_list is empty."""
vnfd = deepcopy(vnfd_wth_persistent_storage)
target_vdu = target_vdu_wth_persistent_storage
vdu_instantiation_volumes_list = []
disk_list = []
+ mock_volume_keeping_required.return_value = True
+ expected_root_disk = {
+ "id": "persistent-root-volume",
+ "type-of-storage": "persistent-storage:persistent-storage",
+ "size-of-storage": "10",
+ "vdu-storage-requirements": [{"key": "keep-volume", "value": "true"}],
+ }
expected_persist_root_disk = {
"persistent-root-volume": {
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": True,
}
}
expected_disk_list = [
{
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": True,
},
]
persist_root_disk = self.ns.find_persistent_root_volumes(
vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list
)
self.assertEqual(persist_root_disk, expected_persist_root_disk)
+ mock_volume_keeping_required.assert_called_once_with(expected_root_disk)
self.assertEqual(disk_list, expected_disk_list)
self.assertEqual(len(disk_list), 1)
- def test_find_persistent_root_volumes_always_selects_first_vsd_as_root(self):
+ @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
+ def test_find_persistent_root_volumes_always_selects_first_vsd_as_root(
+ self, mock_volume_keeping_required
+ ):
"""Find persistent root volume, always selects the first vsd as root volume."""
vnfd = deepcopy(vnfd_wth_persistent_storage)
vnfd["vdu"][0]["virtual-storage-desc"] = [
target_vdu = target_vdu_wth_persistent_storage
vdu_instantiation_volumes_list = []
disk_list = []
+ mock_volume_keeping_required.return_value = True
+ expected_root_disk = {
+ "id": "persistent-volume2",
+ "type-of-storage": "persistent-storage:persistent-storage",
+ "size-of-storage": "10",
+ }
expected_persist_root_disk = {
"persistent-volume2": {
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": True,
}
}
expected_disk_list = [
{
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": True,
},
]
persist_root_disk = self.ns.find_persistent_root_volumes(
vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list
)
self.assertEqual(persist_root_disk, expected_persist_root_disk)
+ mock_volume_keeping_required.assert_called_once_with(expected_root_disk)
self.assertEqual(disk_list, expected_disk_list)
self.assertEqual(len(disk_list), 1)
- def test_find_persistent_root_volumes_empty_size_of_storage(self):
+ @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
+ def test_find_persistent_root_volumes_empty_size_of_storage(
+ self, mock_volume_keeping_required
+ ):
"""Find persistent root volume, size of storage is empty."""
vnfd = deepcopy(vnfd_wth_persistent_storage)
vnfd["virtual-storage-desc"][0]["size-of-storage"] = ""
vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list
)
self.assertEqual(persist_root_disk, None)
+ mock_volume_keeping_required.assert_not_called()
self.assertEqual(disk_list, [])
- def test_find_persistent_root_empty_disk_list(self):
- """Find persistent root volume, empty disk list."""
+ @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
+ def test_find_persistent_root_volumes_keeping_is_not_required(
+ self, mock_volume_keeping_required
+ ):
+ """Find persistent root volume, volume keeping is not required."""
vnfd = deepcopy(vnfd_wth_persistent_storage)
+ vnfd["virtual-storage-desc"][1]["vdu-storage-requirements"] = [
+ {"key": "keep-volume", "value": "false"},
+ ]
target_vdu = target_vdu_wth_persistent_storage
vdu_instantiation_volumes_list = []
disk_list = []
+ mock_volume_keeping_required.return_value = False
+ expected_root_disk = {
+ "id": "persistent-root-volume",
+ "type-of-storage": "persistent-storage:persistent-storage",
+ "size-of-storage": "10",
+ "vdu-storage-requirements": [{"key": "keep-volume", "value": "false"}],
+ }
expected_persist_root_disk = {
"persistent-root-volume": {
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": False,
}
}
expected_disk_list = [
{
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": False,
},
]
persist_root_disk = self.ns.find_persistent_root_volumes(
vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list
)
self.assertEqual(persist_root_disk, expected_persist_root_disk)
+ mock_volume_keeping_required.assert_called_once_with(expected_root_disk)
self.assertEqual(disk_list, expected_disk_list)
self.assertEqual(len(disk_list), 1)
- def test_find_persistent_root_volumes_target_vdu_mismatch(self):
+ @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
+ def test_find_persistent_root_volumes_target_vdu_mismatch(
+ self, mock_volume_keeping_required
+ ):
"""Find persistent root volume, target vdu name is not matching."""
vnfd = deepcopy(vnfd_wth_persistent_storage)
vnfd["vdu"][0]["name"] = "Several_Volumes-VM"
vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list
)
self.assertEqual(result, None)
+ mock_volume_keeping_required.assert_not_called()
self.assertEqual(disk_list, [])
self.assertEqual(len(disk_list), 0)
- def test_find_persistent_root_volumes_with_instantiation_vol_list(self):
+ @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
+ def test_find_persistent_root_volumes_with_instantiation_vol_list(
+ self, mock_volume_keeping_required
+ ):
"""Find persistent root volume, existing volume needs to be used."""
vnfd = deepcopy(vnfd_wth_persistent_storage)
target_vdu = target_vdu_wth_persistent_storage
vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list
)
self.assertEqual(persist_root_disk, expected_persist_root_disk)
+ mock_volume_keeping_required.assert_not_called()
self.assertEqual(disk_list, expected_disk_list)
self.assertEqual(len(disk_list), 1)
- def test_find_persistent_root_volumes_invalid_instantiation_params(self):
+ @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
+ def test_find_persistent_root_volumes_invalid_instantiation_params(
+ self, mock_volume_keeping_required
+ ):
"""Find persistent root volume, existing volume id keyword is invalid."""
vnfd = deepcopy(vnfd_wth_persistent_storage)
target_vdu = target_vdu_wth_persistent_storage
self.ns.find_persistent_root_volumes(
vnfd, target_vdu, vdu_instantiation_volumes_list, disk_list
)
-
+ mock_volume_keeping_required.assert_not_called()
self.assertEqual(disk_list, [])
self.assertEqual(len(disk_list), 0)
+ @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
def test_find_persistent_volumes_vdu_wth_persistent_root_disk_wthout_inst_vol_list(
- self,
+ self, mock_volume_keeping_required
):
"""Find persistent ordinary volume, there is persistent root disk and instatiation volume list is empty."""
persistent_root_disk = {
"persistent-root-volume": {
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": False,
}
}
-
+ mock_volume_keeping_required.return_value = False
target_vdu = target_vdu_wth_persistent_storage
vdu_instantiation_volumes_list = []
disk_list = [
{
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": False,
},
]
-
+ expected_disk = {
+ "id": "persistent-volume2",
+ "size-of-storage": "10",
+ "type-of-storage": "persistent-storage:persistent-storage",
+ }
expected_disk_list = [
{
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": False,
},
{
"size": "10",
+ "keep": False,
},
]
self.ns.find_persistent_volumes(
persistent_root_disk, target_vdu, vdu_instantiation_volumes_list, disk_list
)
self.assertEqual(disk_list, expected_disk_list)
+ mock_volume_keeping_required.assert_called_once_with(expected_disk)
- def test_find_persistent_volumes_vdu_wth_inst_vol_list(self):
+ @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
+ def test_find_persistent_volumes_vdu_wth_inst_vol_list(
+ self, mock_volume_keeping_required
+ ):
"""Find persistent ordinary volume, vim-volume-id is given as instantiation parameter."""
persistent_root_disk = {
"persistent-root-volume": {
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": False,
}
}
vdu_instantiation_volumes_list = [
{
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": False,
},
]
expected_disk_list = [
{
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": False,
},
{
"vim_volume_id": vim_volume_id,
persistent_root_disk, target_vdu, vdu_instantiation_volumes_list, disk_list
)
self.assertEqual(disk_list, expected_disk_list)
+ mock_volume_keeping_required.assert_not_called()
- def test_find_persistent_volumes_vdu_wthout_persistent_storage(self):
+ @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
+ def test_find_persistent_volumes_vdu_wthout_persistent_storage(
+ self, mock_volume_keeping_required
+ ):
"""Find persistent ordinary volume, there is not any persistent disk."""
persistent_root_disk = {}
vdu_instantiation_volumes_list = []
+ mock_volume_keeping_required.return_value = False
target_vdu = target_vdu_wthout_persistent_storage
disk_list = []
self.ns.find_persistent_volumes(
persistent_root_disk, target_vdu, vdu_instantiation_volumes_list, disk_list
)
self.assertEqual(disk_list, disk_list)
+ mock_volume_keeping_required.assert_not_called()
+ @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
def test_find_persistent_volumes_vdu_wth_persistent_root_disk_wthout_ordinary_disk(
- self,
+ self, mock_volume_keeping_required
):
"""There is persistent root disk, but there is not ordinary persistent disk."""
persistent_root_disk = {
"persistent-root-volume": {
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": False,
}
}
vdu_instantiation_volumes_list = []
+ mock_volume_keeping_required.return_value = False
target_vdu = deepcopy(target_vdu_wth_persistent_storage)
target_vdu["virtual-storages"] = [
{
"id": "persistent-root-volume",
"size-of-storage": "10",
"type-of-storage": "persistent-storage:persistent-storage",
+ "vdu-storage-requirements": [
+ {"key": "keep-volume", "value": "true"},
+ ],
},
{
"id": "ephemeral-volume",
{
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": False,
},
]
self.ns.find_persistent_volumes(
persistent_root_disk, target_vdu, vdu_instantiation_volumes_list, disk_list
)
self.assertEqual(disk_list, disk_list)
+ mock_volume_keeping_required.assert_not_called()
- def test_find_persistent_volumes_wth_inst_vol_list_disk_id_mismatch(self):
+ @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
+ def test_find_persistent_volumes_wth_inst_vol_list_disk_id_mismatch(
+ self, mock_volume_keeping_required
+ ):
"""Find persistent ordinary volume, volume id is not persistent_root_disk dict,
vim-volume-id is given as instantiation parameter but disk id is not matching."""
+ mock_volume_keeping_required.return_value = True
vdu_instantiation_volumes_list = [
{
"vim-volume-id": vim_volume_id,
"persistent-root-volume": {
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": False,
}
}
disk_list = [
{
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": False,
},
]
expected_disk_list = [
{
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": False,
},
{
"size": "10",
+ "keep": True,
},
]
-
+ expected_disk = {
+ "id": "persistent-volume2",
+ "size-of-storage": "10",
+ "type-of-storage": "persistent-storage:persistent-storage",
+ }
target_vdu = target_vdu_wth_persistent_storage
self.ns.find_persistent_volumes(
persistent_root_disk, target_vdu, vdu_instantiation_volumes_list, disk_list
)
self.assertEqual(disk_list, expected_disk_list)
+ mock_volume_keeping_required.assert_called_once_with(expected_disk)
+
+ def test_is_volume_keeping_required_true(self):
+ """Volume keeping is required."""
+ virtual_storage_descriptor = {
+ "id": "persistent-root-volume",
+ "type-of-storage": "persistent-storage:persistent-storage",
+ "size-of-storage": "10",
+ "vdu-storage-requirements": [
+ {"key": "keep-volume", "value": "true"},
+ ],
+ }
+ result = self.ns.is_volume_keeping_required(virtual_storage_descriptor)
+ self.assertEqual(result, True)
+
+ def test_is_volume_keeping_required_false(self):
+ """Volume keeping is not required."""
+ virtual_storage_descriptor = {
+ "id": "persistent-root-volume",
+ "type-of-storage": "persistent-storage:persistent-storage",
+ "size-of-storage": "10",
+ "vdu-storage-requirements": [
+ {"key": "keep-volume", "value": "false"},
+ ],
+ }
+ result = self.ns.is_volume_keeping_required(virtual_storage_descriptor)
+ self.assertEqual(result, False)
+
+ def test_is_volume_keeping_required_wthout_vdu_storage_reqirement(self):
+ """Volume keeping is not required, vdu-storage-requirements key does not exist."""
+ virtual_storage_descriptor = {
+ "id": "persistent-root-volume",
+ "type-of-storage": "persistent-storage:persistent-storage",
+ "size-of-storage": "10",
+ }
+ result = self.ns.is_volume_keeping_required(virtual_storage_descriptor)
+ self.assertEqual(result, False)
+
+ def test_is_volume_keeping_required_wrong_keyword(self):
+ """vdu-storage-requirements key to indicate keeping-volume is wrong."""
+ virtual_storage_descriptor = {
+ "id": "persistent-root-volume",
+ "type-of-storage": "persistent-storage:persistent-storage",
+ "size-of-storage": "10",
+ "vdu-storage-requirements": [
+ {"key": "hold-volume", "value": "true"},
+ ],
+ }
+ result = self.ns.is_volume_keeping_required(virtual_storage_descriptor)
+ self.assertEqual(result, False)
def test_sort_vdu_interfaces_position_all_wth_positions(self):
"""Interfaces are sorted according to position, all have positions."""
self.assertDictEqual(cloud_config, expected_cloud_config)
@patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
- def test_add_persistent_root_disk_to_disk_list(
- self, mock_select_persistent_root_disk
+ @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
+ def test_add_persistent_root_disk_to_disk_list_keep_false(
+ self, mock_volume_keeping_required, mock_select_persistent_root_disk
):
- """Add persistent root disk to disk_list"""
+ """Add persistent root disk to disk_list, keep volume set to False."""
root_disk = {
"id": "persistent-root-volume",
"type-of-storage": "persistent-storage:persistent-storage",
target_vdu = deepcopy(target_vdu_wth_persistent_storage)
persistent_root_disk = {}
disk_list = []
+ mock_volume_keeping_required.return_value = False
expected_disk_list = [
{
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": False,
}
]
self.ns._add_persistent_root_disk_to_disk_list(
)
self.assertEqual(disk_list, expected_disk_list)
mock_select_persistent_root_disk.assert_called_once()
+ mock_volume_keeping_required.assert_called_once()
@patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
+ @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
def test_add_persistent_root_disk_to_disk_list_select_persistent_root_disk_raises(
- self, mock_select_persistent_root_disk
+ self, mock_volume_keeping_required, mock_select_persistent_root_disk
):
"""Add persistent root disk to disk_list"""
root_disk = {
)
self.assertEqual(disk_list, [])
mock_select_persistent_root_disk.assert_called_once()
+ mock_volume_keeping_required.assert_not_called()
- def test_add_persistent_ordinary_disk_to_disk_list(self):
- """Add persistent ordinary disk to disk_list"""
+ @patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
+ @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
+ def test_add_persistent_root_disk_to_disk_list_keep_true(
+ self, mock_volume_keeping_required, mock_select_persistent_root_disk
+ ):
+ """Add persistent root disk, keeo volume set to True."""
+ vnfd = deepcopy(vnfd_wth_persistent_storage)
target_vdu = deepcopy(target_vdu_wth_persistent_storage)
+ mock_volume_keeping_required.return_value = True
+ root_disk = {
+ "id": "persistent-root-volume",
+ "type-of-storage": "persistent-storage:persistent-storage",
+ "size-of-storage": "10",
+ "vdu-storage-requirements": [
+ {"key": "keep-volume", "value": "true"},
+ ],
+ }
+ mock_select_persistent_root_disk.return_value = root_disk
+ persistent_root_disk = {}
+ disk_list = []
+ expected_disk_list = [
+ {
+ "image_id": "ubuntu20.04",
+ "size": "10",
+ "keep": True,
+ }
+ ]
+ self.ns._add_persistent_root_disk_to_disk_list(
+ vnfd, target_vdu, persistent_root_disk, disk_list
+ )
+ self.assertEqual(disk_list, expected_disk_list)
+ mock_volume_keeping_required.assert_called_once_with(root_disk)
+
+ @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
+ def test_add_persistent_ordinary_disk_to_disk_list(
+ self, mock_volume_keeping_required
+ ):
+ """Add persistent ordinary disk, keeo volume set to True."""
+ target_vdu = deepcopy(target_vdu_wth_persistent_storage)
+ mock_volume_keeping_required.return_value = False
persistent_root_disk = {
"persistent-root-volume": {
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": True,
}
}
+ ordinary_disk = {
+ "id": "persistent-volume2",
+ "type-of-storage": "persistent-storage:persistent-storage",
+ "size-of-storage": "10",
+ }
persistent_ordinary_disk = {}
disk_list = []
expected_disk_list = [
{
"size": "10",
+ "keep": False,
}
]
self.ns._add_persistent_ordinary_disks_to_disk_list(
target_vdu, persistent_root_disk, persistent_ordinary_disk, disk_list
)
self.assertEqual(disk_list, expected_disk_list)
+ mock_volume_keeping_required.assert_called_once_with(ordinary_disk)
- def test_add_persistent_ordinary_disk_to_disk_list_vsd_id_in_root_disk_dict(self):
+ @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
+ def test_add_persistent_ordinary_disk_to_disk_list_vsd_id_in_root_disk_dict(
+ self, mock_volume_keeping_required
+ ):
"""Add persistent ordinary disk, vsd id is in root_disk dict."""
target_vdu = deepcopy(target_vdu_wth_persistent_storage)
+ mock_volume_keeping_required.return_value = False
persistent_root_disk = {
"persistent-root-volume": {
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": True,
},
"persistent-volume2": {
"size": "10",
target_vdu, persistent_root_disk, persistent_ordinary_disk, disk_list
)
self.assertEqual(disk_list, [])
+ mock_volume_keeping_required.assert_not_called()
@patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
+ @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
def test_add_persistent_root_disk_to_disk_list_vnfd_wthout_persistent_storage(
- self, mock_select_persistent_root_disk
+ self, mock_volume_keeping_required, mock_select_persistent_root_disk
):
"""VNFD does not have persistent storage."""
vnfd = deepcopy(vnfd_wthout_persistent_storage)
)
self.assertEqual(disk_list, [])
self.assertEqual(mock_select_persistent_root_disk.call_count, 2)
+ mock_volume_keeping_required.assert_not_called()
@patch("osm_ng_ro.ns.Ns._select_persistent_root_disk")
+ @patch("osm_ng_ro.ns.Ns.is_volume_keeping_required")
def test_add_persistent_root_disk_to_disk_list_wthout_persistent_root_disk(
- self, mock_select_persistent_root_disk
+ self, mock_volume_keeping_required, mock_select_persistent_root_disk
):
"""Persistent_root_disk dict is empty."""
vnfd = deepcopy(vnfd_wthout_persistent_storage)
)
self.assertEqual(disk_list, [])
self.assertEqual(mock_select_persistent_root_disk.call_count, 2)
+ mock_volume_keeping_required.assert_not_called()
def test_prepare_vdu_affinity_group_list_invalid_extra_dict(self):
"""Invalid extra dict."""
"persistent-root-volume": {
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": True,
}
}
mock_find_persistent_root_volumes.return_value = persistent_root_disk
"persistent-root-volume": {
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": True,
}
}
mock_find_persistent_root_volumes.return_value = persistent_root_disk
"persistent-root-volume": {
"image_id": "ubuntu20.04",
"size": "10",
+ "keep": True,
}
}
mock_find_persistent_root_volumes.return_value = persistent_root_disk
self.assertEqual(existing_vim_volumes, expected_existing_vim_volumes)
self.vimconn.cinder.volumes.create.assert_not_called()
- def test_prepare_persistent_non_root_volumes_vim_using_volume_id(self):
+ @patch.object(vimconnector, "update_block_device_mapping")
+ def test_prepare_persistent_non_root_volumes_vim_using_volume_id(
+ self, mock_update_block_device_mapping
+ ):
"""Existing persistent non root volume with vim_volume_id."""
vm_av_zone = ["nova"]
base_disk_index = ord("b")
self.assertDictEqual(block_device_mapping, expected_block_device_mapping)
self.assertEqual(existing_vim_volumes, expected_existing_vim_volumes)
self.vimconn.cinder.volumes.create.assert_not_called()
+ mock_update_block_device_mapping.assert_not_called()
- def test_prepare_persistent_root_volumes_using_vim_id(self):
+ @patch.object(vimconnector, "update_block_device_mapping")
+ def test_prepare_persistent_root_volumes_using_vim_id(
+ self, mock_update_block_device_mapping
+ ):
"""Existing persistent root volume with vim_id."""
vm_av_zone = ["nova"]
base_disk_index = ord("a")
self.assertDictEqual(block_device_mapping, expected_block_device_mapping)
self.assertEqual(existing_vim_volumes, expected_existing_vim_volumes)
self.vimconn.cinder.volumes.create.assert_not_called()
+ mock_update_block_device_mapping.assert_not_called()
- def test_prepare_persistent_non_root_volumes_using_vim_id(self):
+ @patch.object(vimconnector, "update_block_device_mapping")
+ def test_prepare_persistent_non_root_volumes_using_vim_id(
+ self, mock_update_block_device_mapping
+ ):
"""Existing persistent root volume with vim_id."""
vm_av_zone = ["nova"]
base_disk_index = ord("b")
self.assertDictEqual(block_device_mapping, expected_block_device_mapping)
self.assertEqual(existing_vim_volumes, expected_existing_vim_volumes)
self.vimconn.cinder.volumes.create.assert_not_called()
+ mock_update_block_device_mapping.assert_not_called()
- def test_prepare_persistent_root_volumes_create(self):
+ @patch.object(vimconnector, "update_block_device_mapping")
+ def test_prepare_persistent_root_volumes_create(
+ self, mock_update_block_device_mapping
+ ):
"""Create persistent root volume."""
self.vimconn.cinder.volumes.create.return_value.id = volume_id2
vm_av_zone = ["nova"]
existing_vim_volumes = []
created_items = {}
expected_boot_vol_id = volume_id2
- expected_block_device_mapping = {"vda": volume_id2}
+ boot_volume_id = self.vimconn._prepare_persistent_root_volumes(
+ name,
+ vm_av_zone,
+ disk,
+ base_disk_index,
+ block_device_mapping,
+ existing_vim_volumes,
+ created_items,
+ )
+ self.assertEqual(boot_volume_id, expected_boot_vol_id)
+ self.vimconn.cinder.volumes.create.assert_called_once_with(
+ size=10,
+ name="basicvmvda",
+ imageRef=image_id,
+ availability_zone=["nova"],
+ )
+ mock_update_block_device_mapping.assert_called_once()
+ _call_mock_update_block_device_mapping = (
+ mock_update_block_device_mapping.call_args_list
+ )
+ self.assertEqual(
+ _call_mock_update_block_device_mapping[0].kwargs["block_device_mapping"],
+ block_device_mapping,
+ )
+ self.assertEqual(
+ _call_mock_update_block_device_mapping[0].kwargs["base_disk_index"], 97
+ )
+ self.assertEqual(_call_mock_update_block_device_mapping[0].kwargs["disk"], disk)
+ self.assertEqual(
+ _call_mock_update_block_device_mapping[0].kwargs["created_items"], {}
+ )
+
+ @patch.object(vimconnector, "update_block_device_mapping")
+ def test_prepare_persistent_root_volumes_create_with_keep(
+ self, mock_update_block_device_mapping
+ ):
+ """Create persistent root volume, disk has keep parameter."""
+ self.vimconn.cinder.volumes.create.return_value.id = volume_id2
+ vm_av_zone = ["nova"]
+ base_disk_index = ord("a")
+ disk = {"size": 10, "image_id": image_id, "keep": True}
+ block_device_mapping = {}
+ existing_vim_volumes = []
+ created_items = {}
+ expected_boot_vol_id = volume_id2
expected_existing_vim_volumes = []
boot_volume_id = self.vimconn._prepare_persistent_root_volumes(
name,
created_items,
)
self.assertEqual(boot_volume_id, expected_boot_vol_id)
- self.assertDictEqual(block_device_mapping, expected_block_device_mapping)
self.assertEqual(existing_vim_volumes, expected_existing_vim_volumes)
self.vimconn.cinder.volumes.create.assert_called_once_with(
size=10,
imageRef=image_id,
availability_zone=["nova"],
)
- self.assertEqual(created_items, {f"volume:{volume_id2}": True})
+ mock_update_block_device_mapping.assert_called_once()
+ _call_mock_update_block_device_mapping = (
+ mock_update_block_device_mapping.call_args_list
+ )
+ self.assertEqual(
+ _call_mock_update_block_device_mapping[0].kwargs["block_device_mapping"],
+ block_device_mapping,
+ )
+ self.assertEqual(
+ _call_mock_update_block_device_mapping[0].kwargs["base_disk_index"], 97
+ )
+ self.assertEqual(_call_mock_update_block_device_mapping[0].kwargs["disk"], disk)
+ self.assertEqual(
+ _call_mock_update_block_device_mapping[0].kwargs["created_items"], {}
+ )
- def test_prepare_persistent_non_root_volumes_create(self):
+ @patch.object(vimconnector, "update_block_device_mapping")
+ def test_prepare_persistent_non_root_volumes_create(
+ self, mock_update_block_device_mapping
+ ):
"""Create persistent non-root volume."""
self.vimconn.cinder = CopyingMock()
self.vimconn.cinder.volumes.create.return_value.id = volume_id2
block_device_mapping = {}
existing_vim_volumes = []
created_items = {}
- expected_block_device_mapping = {"vda": volume_id2}
expected_existing_vim_volumes = []
self.vimconn._prepare_non_root_persistent_volumes(
name,
created_items,
)
- self.assertDictEqual(block_device_mapping, expected_block_device_mapping)
self.assertEqual(existing_vim_volumes, expected_existing_vim_volumes)
self.vimconn.cinder.volumes.create.assert_called_once_with(
size=10, name="basicvmvda", availability_zone=["nova"]
)
- self.assertEqual(created_items, {f"volume:{volume_id2}": True})
+ mock_update_block_device_mapping.assert_called_once()
+ _call_mock_update_block_device_mapping = (
+ mock_update_block_device_mapping.call_args_list
+ )
+ self.assertEqual(
+ _call_mock_update_block_device_mapping[0].kwargs["block_device_mapping"],
+ block_device_mapping,
+ )
+ self.assertEqual(
+ _call_mock_update_block_device_mapping[0].kwargs["base_disk_index"], 97
+ )
+ self.assertEqual(_call_mock_update_block_device_mapping[0].kwargs["disk"], disk)
+ self.assertEqual(
+ _call_mock_update_block_device_mapping[0].kwargs["created_items"], {}
+ )
+
+ @patch.object(vimconnector, "update_block_device_mapping")
+ def test_prepare_persistent_non_root_volumes_create_with_keep(
+ self, mock_update_block_device_mapping
+ ):
+ """Create persistent non-root volume."""
+ self.vimconn.cinder = CopyingMock()
+ self.vimconn.cinder.volumes.create.return_value.id = volume_id2
+ vm_av_zone = ["nova"]
+ base_disk_index = ord("a")
+ disk = {"size": 10, "keep": True}
+ block_device_mapping = {}
+ existing_vim_volumes = []
+ created_items = {}
+ expected_existing_vim_volumes = []
+ self.vimconn._prepare_non_root_persistent_volumes(
+ name,
+ disk,
+ vm_av_zone,
+ block_device_mapping,
+ base_disk_index,
+ existing_vim_volumes,
+ created_items,
+ )
+
+ self.assertEqual(existing_vim_volumes, expected_existing_vim_volumes)
+ self.vimconn.cinder.volumes.create.assert_called_once_with(
+ size=10, name="basicvmvda", availability_zone=["nova"]
+ )
+ mock_update_block_device_mapping.assert_called_once()
+ _call_mock_update_block_device_mapping = (
+ mock_update_block_device_mapping.call_args_list
+ )
+ self.assertEqual(
+ _call_mock_update_block_device_mapping[0].kwargs["block_device_mapping"],
+ block_device_mapping,
+ )
+ self.assertEqual(
+ _call_mock_update_block_device_mapping[0].kwargs["base_disk_index"], 97
+ )
+ self.assertEqual(_call_mock_update_block_device_mapping[0].kwargs["disk"], disk)
+ self.assertEqual(
+ _call_mock_update_block_device_mapping[0].kwargs["created_items"], {}
+ )
- def test_prepare_persistent_root_volumes_create_raise_exception(self):
+ @patch.object(vimconnector, "update_block_device_mapping")
+ def test_prepare_persistent_root_volumes_create_raise_exception(
+ self, mock_update_block_device_mapping
+ ):
"""Create persistent root volume raise exception."""
self.vimconn.cinder.volumes.create.side_effect = Exception
vm_av_zone = ["nova"]
self.assertEqual(existing_vim_volumes, [])
self.assertEqual(block_device_mapping, {})
self.assertEqual(created_items, {})
+ mock_update_block_device_mapping.assert_not_called()
- def test_prepare_persistent_non_root_volumes_create_raise_exception(self):
+ @patch.object(vimconnector, "update_block_device_mapping")
+ def test_prepare_persistent_non_root_volumes_create_raise_exception(
+ self, mock_update_block_device_mapping
+ ):
"""Create persistent non-root volume raise exception."""
self.vimconn.cinder.volumes.create.side_effect = Exception
vm_av_zone = ["nova"]
self.assertEqual(existing_vim_volumes, [])
self.assertEqual(block_device_mapping, {})
self.assertEqual(created_items, {})
+ mock_update_block_device_mapping.assert_not_called()
@patch("time.sleep")
def test_wait_for_created_volumes_availability_volume_status_available(
mock_created_vol_availability.return_value = 10
mock_existing_vol_availability.return_value = 15
self.vimconn.cinder = CopyingMock()
-
self.vimconn._prepare_disk_for_vminstance(
name,
existing_vim_volumes,
self.vimconn.neutron.update_port.assert_not_called()
@patch("time.time")
+ @patch.object(vimconnector, "remove_keep_tag_from_persistent_volumes")
@patch.object(vimconnector, "_reload_connection")
@patch.object(vimconnector, "_prepare_network_for_vminstance")
@patch.object(vimconnector, "_create_user_data")
mock_create_user_data,
mock_prepare_network_for_vm_instance,
mock_reload_connection,
+ mock_remove_keep_flag_from_persistent_volumes,
mock_time,
):
"""New VM instance creation is successful."""
created_items={},
vm_start_time=time_return_value,
)
+ mock_remove_keep_flag_from_persistent_volumes.assert_not_called()
mock_delete_vm_instance.assert_not_called()
mock_format_exception.assert_not_called()
@patch("time.time")
+ @patch.object(vimconnector, "remove_keep_tag_from_persistent_volumes")
@patch.object(vimconnector, "_reload_connection")
@patch.object(vimconnector, "_prepare_network_for_vminstance")
@patch.object(vimconnector, "_create_user_data")
mock_create_user_data,
mock_prepare_network_for_vm_instance,
mock_reload_connection,
+ mock_remove_keep_flag_from_persistent_volumes,
mock_time,
):
"""New VM instance creation failed because of user data creation failure."""
mock_get_vm_availability_zone.return_value = "nova"
+ mock_remove_keep_flag_from_persistent_volumes.return_value = {}
+
self.vimconn.nova.servers.create.return_value = self.server
mock_time.return_value = time_return_value
mock_time.assert_not_called()
mock_update_port_security.assert_not_called()
mock_prepare_external_network.assert_not_called()
+ mock_remove_keep_flag_from_persistent_volumes.assert_called_once_with({})
mock_delete_vm_instance.assert_called_once_with(None, {})
mock_format_exception.assert_called_once()
arg = mock_format_exception.call_args[0][0]
self.assertEqual(str(arg), "User data could not be retrieved.")
@patch("time.time")
+ @patch.object(vimconnector, "remove_keep_tag_from_persistent_volumes")
@patch.object(vimconnector, "_reload_connection")
@patch.object(vimconnector, "_prepare_network_for_vminstance")
@patch.object(vimconnector, "_create_user_data")
mock_create_user_data,
mock_prepare_network_for_vm_instance,
mock_reload_connection,
+ mock_remove_keep_flag_from_persistent_volumes,
mock_time,
):
"""New VM instance creation, external network connection has failed as floating
mock_time.return_value = time_return_value
+ mock_remove_keep_flag_from_persistent_volumes.return_value = {}
+
mock_prepare_external_network.side_effect = VimConnException(
"Can not create floating ip."
)
created_items={},
vm_start_time=time_return_value,
)
+ mock_remove_keep_flag_from_persistent_volumes.assert_called_once_with({})
mock_delete_vm_instance.assert_called_once_with(self.server.id, {})
mock_format_exception.assert_called_once()
arg = mock_format_exception.call_args[0][0]
self.assertEqual(str(arg), "Can not create floating ip.")
@patch("time.time")
+ @patch.object(vimconnector, "remove_keep_tag_from_persistent_volumes")
@patch.object(vimconnector, "_reload_connection")
@patch.object(vimconnector, "_prepare_network_for_vminstance")
@patch.object(vimconnector, "_create_user_data")
mock_create_user_data,
mock_prepare_network_for_vm_instance,
mock_reload_connection,
+ mock_remove_keep_flag_from_persistent_volumes,
mock_time,
):
"""New VM creation with affinity group."""
created_items={},
vm_start_time=time_return_value,
)
+ mock_remove_keep_flag_from_persistent_volumes.assert_not_called()
mock_delete_vm_instance.assert_not_called()
mock_format_exception.assert_not_called()
@patch("time.time")
+ @patch.object(vimconnector, "remove_keep_tag_from_persistent_volumes")
@patch.object(vimconnector, "_reload_connection")
@patch.object(vimconnector, "_prepare_network_for_vminstance")
@patch.object(vimconnector, "_create_user_data")
mock_create_user_data,
mock_prepare_network_for_vm_instance,
mock_reload_connection,
+ mock_remove_keep_flag_from_persistent_volumes,
mock_time,
):
"""New VM(server) creation failed."""
mock_time.return_value = time_return_value
+ mock_remove_keep_flag_from_persistent_volumes.return_value = {}
+
self.vimconn.new_vminstance(
name,
description,
mock_time.assert_not_called()
mock_update_port_security.assert_not_called()
mock_prepare_external_network.assert_not_called()
+ mock_remove_keep_flag_from_persistent_volumes.assert_called_once_with({})
mock_delete_vm_instance.assert_called_once_with(None, {})
mock_format_exception.assert_called_once()
arg = mock_format_exception.call_args[0][0]
self.assertEqual(str(arg), "Server could not be created.")
@patch("time.time")
+ @patch.object(vimconnector, "remove_keep_tag_from_persistent_volumes")
@patch.object(vimconnector, "_reload_connection")
@patch.object(vimconnector, "_prepare_network_for_vminstance")
@patch.object(vimconnector, "_create_user_data")
mock_create_user_data,
mock_prepare_network_for_vm_instance,
mock_reload_connection,
+ mock_remove_keep_flag_from_persistent_volumes,
mock_time,
):
"""Connection to Cloud API has failed."""
mock_get_vm_availability_zone.return_value = "nova"
self.vimconn.nova.servers.create.return_value = self.server
mock_time.return_value = time_return_value
+ mock_remove_keep_flag_from_persistent_volumes.return_value = {}
self.vimconn.new_vminstance(
name,
mock_time.assert_not_called()
mock_update_port_security.assert_not_called()
mock_prepare_external_network.assert_not_called()
+ mock_remove_keep_flag_from_persistent_volumes.assert_called_once_with({})
mock_delete_vm_instance.assert_called_once_with(None, {})
@patch.object(vimconnector, "_delete_ports_by_id_wth_neutron")
self.vimconn.logger.error.assert_not_called()
@patch("time.sleep")
+ @patch.object(vimconnector, "_extract_items_wth_keep_flag_from_created_items")
@patch.object(vimconnector, "_format_exception")
@patch.object(vimconnector, "_reload_connection")
@patch.object(vimconnector, "_delete_vm_ports_attached_to_network")
mock_delete_vm_ports_attached_to_network,
mock_reload_connection,
mock_format_exception,
+ mock_extract_items_wth_keep_flag_from_created_items,
mock_sleep,
):
vm_id = f"{virtual_mac_id}"
created_items = deepcopy(created_items_all_true)
+ mock_extract_items_wth_keep_flag_from_created_items.return_value = created_items
volumes_to_hold = [f"{volume_id}", f"{volume_id2}"]
mock_delete_created_items.return_value = False
self.vimconn.delete_vminstance(vm_id, created_items, volumes_to_hold)
)
mock_sleep.assert_not_called()
mock_format_exception.assert_not_called()
+ mock_extract_items_wth_keep_flag_from_created_items.assert_called_once_with(
+ created_items
+ )
+
+ @patch("time.sleep")
+ @patch.object(vimconnector, "_extract_items_wth_keep_flag_from_created_items")
+ @patch.object(vimconnector, "_format_exception")
+ @patch.object(vimconnector, "_reload_connection")
+ @patch.object(vimconnector, "_delete_vm_ports_attached_to_network")
+ @patch.object(vimconnector, "_delete_created_items")
+ def test_delete_vminstance_created_items_has_keep_flag(
+ self,
+ mock_delete_created_items,
+ mock_delete_vm_ports_attached_to_network,
+ mock_reload_connection,
+ mock_format_exception,
+ mock_extract_items_wth_keep_flag_from_created_items,
+ mock_sleep,
+ ):
+ """Created_items includes items which has keep flag."""
+ vm_id = f"{virtual_mac_id}"
+ initial_created_items = {
+ f"port{port_id}": True,
+ f"floating_ip{floating_network_vim_id}": None,
+ f"volume{volume_id}keep": True,
+ f"volume{volume_id2}keep": True,
+ }
+ created_items = {
+ f"port{port_id}": True,
+ f"floating_ip{floating_network_vim_id}": None,
+ }
+ mock_extract_items_wth_keep_flag_from_created_items.return_value = created_items
+ volumes_to_hold = []
+ mock_delete_created_items.return_value = False
+ self.vimconn.delete_vminstance(vm_id, initial_created_items, volumes_to_hold)
+ mock_reload_connection.assert_called_once()
+ mock_delete_vm_ports_attached_to_network.assert_called_once_with(created_items)
+ self.vimconn.nova.servers.delete.assert_called_once_with(vm_id)
+ mock_delete_created_items.assert_called_once_with(
+ created_items, volumes_to_hold, False
+ )
+ mock_sleep.assert_not_called()
+ mock_format_exception.assert_not_called()
+ mock_extract_items_wth_keep_flag_from_created_items.assert_called_once_with(
+ initial_created_items
+ )
@patch("time.sleep")
+ @patch.object(vimconnector, "_extract_items_wth_keep_flag_from_created_items")
+ @patch.object(vimconnector, "_format_exception")
+ @patch.object(vimconnector, "_reload_connection")
+ @patch.object(vimconnector, "_delete_vm_ports_attached_to_network")
+ @patch.object(vimconnector, "_delete_created_items")
+ def test_delete_vminstance_extract_items_wth_keep_raises(
+ self,
+ mock_delete_created_items,
+ mock_delete_vm_ports_attached_to_network,
+ mock_reload_connection,
+ mock_format_exception,
+ mock_extract_items_wth_keep_flag_from_created_items,
+ mock_sleep,
+ ):
+ """extract_items_wth_keep_flag_from_created_items raises AttributeError."""
+ vm_id = f"{virtual_mac_id}"
+ initial_created_items = {
+ f"port{port_id}": True,
+ f"floating_ip{floating_network_vim_id}": None,
+ f"volume{volume_id}keep": True,
+ f"volume{volume_id2}keep": True,
+ }
+
+ mock_extract_items_wth_keep_flag_from_created_items.side_effect = AttributeError
+ volumes_to_hold = []
+ mock_delete_created_items.return_value = False
+ with self.assertRaises(AttributeError):
+ self.vimconn.delete_vminstance(
+ vm_id, initial_created_items, volumes_to_hold
+ )
+ mock_reload_connection.assert_not_called()
+ mock_delete_vm_ports_attached_to_network.assert_not_called()
+ self.vimconn.nova.servers.delete.assert_not_called()
+ mock_delete_created_items.assert_not_called()
+ mock_sleep.assert_not_called()
+ mock_format_exception.assert_not_called()
+ mock_extract_items_wth_keep_flag_from_created_items.assert_called_once_with(
+ initial_created_items
+ )
+
+ @patch("time.sleep")
+ @patch.object(vimconnector, "_extract_items_wth_keep_flag_from_created_items")
@patch.object(vimconnector, "_format_exception")
@patch.object(vimconnector, "_reload_connection")
@patch.object(vimconnector, "_delete_vm_ports_attached_to_network")
mock_delete_vm_ports_attached_to_network,
mock_reload_connection,
mock_format_exception,
+ mock_extract_items_wth_keep_flag_from_created_items,
mock_sleep,
):
"""Delete creted items raises exception."""
vm_id = f"{virtual_mac_id}"
created_items = deepcopy(created_items_all_true)
+ mock_extract_items_wth_keep_flag_from_created_items.return_value = created_items
mock_sleep = MagicMock()
volumes_to_hold = []
err = ConnectionError("ClientException occured.")
self.vimconn.nova.servers.delete.assert_called_once_with(vm_id)
mock_delete_created_items.assert_called_once()
mock_sleep.assert_not_called()
+ mock_extract_items_wth_keep_flag_from_created_items.assert_called_once_with(
+ created_items
+ )
@patch("time.sleep")
+ @patch.object(vimconnector, "_extract_items_wth_keep_flag_from_created_items")
@patch.object(vimconnector, "_format_exception")
@patch.object(vimconnector, "_reload_connection")
@patch.object(vimconnector, "_delete_vm_ports_attached_to_network")
mock_delete_vm_ports_attached_to_network,
mock_reload_connection,
mock_format_exception,
+ mock_extract_items_wth_keep_flag_from_created_items,
mock_sleep,
):
"""Delete vm ports raises exception."""
vm_id = f"{virtual_mac_id}"
created_items = deepcopy(created_items_all_true)
+ mock_extract_items_wth_keep_flag_from_created_items.return_value = created_items
volumes_to_hold = [f"{volume_id}", f"{volume_id2}"]
err = ConnectionError("ClientException occured.")
mock_delete_vm_ports_attached_to_network.side_effect = err
self.vimconn.nova.servers.delete.assert_not_called()
mock_delete_created_items.assert_not_called()
mock_sleep.assert_not_called()
+ mock_extract_items_wth_keep_flag_from_created_items.assert_called_once_with(
+ created_items
+ )
@patch("time.sleep")
+ @patch.object(vimconnector, "_extract_items_wth_keep_flag_from_created_items")
@patch.object(vimconnector, "_format_exception")
@patch.object(vimconnector, "_reload_connection")
@patch.object(vimconnector, "_delete_vm_ports_attached_to_network")
mock_delete_vm_ports_attached_to_network,
mock_reload_connection,
mock_format_exception,
+ mock_extract_items_wth_keep_flag_from_created_items,
mock_sleep,
):
"""Nova server delete raises exception."""
vm_id = f"{virtual_mac_id}"
created_items = deepcopy(created_items_all_true)
+ mock_extract_items_wth_keep_flag_from_created_items.return_value = created_items
volumes_to_hold = [f"{volume_id}", f"{volume_id2}"]
err = VimConnConnectionException("ClientException occured.")
self.vimconn.nova.servers.delete.side_effect = err
self.vimconn.nova.servers.delete.assert_called_once_with(vm_id)
mock_delete_created_items.assert_not_called()
mock_sleep.assert_not_called()
+ mock_extract_items_wth_keep_flag_from_created_items.assert_called_once_with(
+ created_items
+ )
@patch("time.sleep")
+ @patch.object(vimconnector, "_extract_items_wth_keep_flag_from_created_items")
@patch.object(vimconnector, "_format_exception")
@patch.object(vimconnector, "_reload_connection")
@patch.object(vimconnector, "_delete_vm_ports_attached_to_network")
mock_delete_vm_ports_attached_to_network,
mock_reload_connection,
mock_format_exception,
+ mock_extract_items_wth_keep_flag_from_created_items,
mock_sleep,
):
"""Reload connection raises exception."""
vm_id = f"{virtual_mac_id}"
created_items = deepcopy(created_items_all_true)
+ mock_extract_items_wth_keep_flag_from_created_items.return_value = created_items
mock_sleep = MagicMock()
volumes_to_hold = [f"{volume_id}", f"{volume_id2}"]
err = ConnectionError("ClientException occured.")
self.vimconn.nova.servers.delete.assert_not_called()
mock_delete_created_items.assert_not_called()
mock_sleep.assert_not_called()
+ mock_extract_items_wth_keep_flag_from_created_items.assert_called_once_with(
+ created_items
+ )
@patch("time.sleep")
+ @patch.object(vimconnector, "_extract_items_wth_keep_flag_from_created_items")
@patch.object(vimconnector, "_format_exception")
@patch.object(vimconnector, "_reload_connection")
@patch.object(vimconnector, "_delete_vm_ports_attached_to_network")
mock_delete_vm_ports_attached_to_network,
mock_reload_connection,
mock_format_exception,
+ mock_extract_items_wth_keep_flag_from_created_items,
mock_sleep,
):
"""created_items and volumes_to_hold are None."""
vm_id = f"{virtual_mac_id}"
created_items = None
volumes_to_hold = None
+ mock_extract_items_wth_keep_flag_from_created_items.return_value = {}
mock_delete_created_items.return_value = False
self.vimconn.delete_vminstance(vm_id, created_items, volumes_to_hold)
mock_reload_connection.assert_called_once()
mock_delete_created_items.assert_called_once_with({}, [], False)
mock_sleep.assert_not_called()
mock_format_exception.assert_not_called()
+ mock_extract_items_wth_keep_flag_from_created_items.assert_called_once_with({})
@patch("time.sleep")
+ @patch.object(vimconnector, "_extract_items_wth_keep_flag_from_created_items")
@patch.object(vimconnector, "_format_exception")
@patch.object(vimconnector, "_reload_connection")
@patch.object(vimconnector, "_delete_vm_ports_attached_to_network")
mock_delete_vm_ports_attached_to_network,
mock_reload_connection,
mock_format_exception,
+ mock_extract_items_wth_keep_flag_from_created_items,
mock_sleep,
):
"""vm_id is None."""
vm_id = None
created_items = deepcopy(created_items_all_true)
+ mock_extract_items_wth_keep_flag_from_created_items.return_value = created_items
volumes_to_hold = [f"{volume_id}", f"{volume_id2}"]
mock_delete_created_items.side_effect = [True, True, False]
self.vimconn.delete_vminstance(vm_id, created_items, volumes_to_hold)
self.assertEqual(mock_delete_created_items.call_count, 3)
self.assertEqual(mock_sleep.call_count, 2)
mock_format_exception.assert_not_called()
+ mock_extract_items_wth_keep_flag_from_created_items.assert_called_once_with(
+ created_items
+ )
@patch("time.sleep")
+ @patch.object(vimconnector, "_extract_items_wth_keep_flag_from_created_items")
@patch.object(vimconnector, "_format_exception")
@patch.object(vimconnector, "_reload_connection")
@patch.object(vimconnector, "_delete_vm_ports_attached_to_network")
mock_delete_vm_ports_attached_to_network,
mock_reload_connection,
mock_format_exception,
+ mock_extract_items_wth_keep_flag_from_created_items,
mock_sleep,
):
"""Delete created items always return True."""
vm_id = None
created_items = deepcopy(created_items_all_true)
+ mock_extract_items_wth_keep_flag_from_created_items.return_value = created_items
volumes_to_hold = [f"{volume_id}", f"{volume_id2}"]
mock_delete_created_items.side_effect = [True] * 1800
self.vimconn.delete_vminstance(vm_id, created_items, volumes_to_hold)
self.assertEqual(mock_delete_created_items.call_count, 1800)
self.assertEqual(mock_sleep.call_count, 1800)
mock_format_exception.assert_not_called()
+ mock_extract_items_wth_keep_flag_from_created_items.assert_called_once_with(
+ created_items
+ )
+
+ def test_remove_keep_tag_from_persistent_volumes_keep_flag_exists(self):
+ """Keep flag exists in created items."""
+ created_items = {
+ f"port:{port_id}": True,
+ f"floating_ip:{floating_network_vim_id}": True,
+ f"volume:{volume_id}:keep": True,
+ f"volume:{volume_id2}:keep": True,
+ }
+ expected_result = {
+ f"port:{port_id}": True,
+ f"floating_ip:{floating_network_vim_id}": True,
+ f"volume:{volume_id}": True,
+ f"volume:{volume_id2}": True,
+ }
+ result = self.vimconn.remove_keep_tag_from_persistent_volumes(created_items)
+ self.assertDictEqual(result, expected_result)
+
+ def test_remove_keep_tag_from_persistent_volumes_without_keep_flag(self):
+ """Keep flag does not exist in created items."""
+ created_items = {
+ f"port:{port_id}": True,
+ f"floating_ip:{floating_network_vim_id}": True,
+ f"volume:{volume_id}": True,
+ f"volume:{volume_id2}": True,
+ }
+ result = self.vimconn.remove_keep_tag_from_persistent_volumes(created_items)
+ self.assertDictEqual(result, created_items)
+
+ def test_update_block_device_mapping_empty_volume(self):
+ """"""
+ volume = ""
+ block_device_mapping = {}
+ base_disk_index = 100
+ disk = {}
+ created_items = {}
+ with self.assertRaises(VimConnException) as err:
+ self.vimconn.update_block_device_mapping(
+ volume, block_device_mapping, base_disk_index, disk, created_items
+ )
+ self.assertEqual(str(err), "Volume is empty.")
+ self.assertEqual(block_device_mapping, {})
+ self.assertEqual(created_items, {})
+
+ def test_update_block_device_mapping_invalid_volume(self):
+ """"""
+ volume = "Volume-A"
+ block_device_mapping = {}
+ base_disk_index = 100
+ disk = {}
+ created_items = {}
+ with self.assertRaises(VimConnException) as err:
+ self.vimconn.update_block_device_mapping(
+ volume, block_device_mapping, base_disk_index, disk, created_items
+ )
+ self.assertEqual(
+ str(err), "Created volume is not valid, does not have id attribute."
+ )
+ self.assertEqual(block_device_mapping, {})
+ self.assertEqual(created_items, {})
+
+ def test_update_block_device_mapping(self):
+ """"""
+ volume = MagicMock(autospec=True)
+ volume.id = volume_id
+ block_device_mapping = {}
+ base_disk_index = 100
+ disk = {}
+ created_items = {}
+ self.vimconn.update_block_device_mapping(
+ volume, block_device_mapping, base_disk_index, disk, created_items
+ )
+ self.assertEqual(
+ block_device_mapping, {"vdd": "ac408b73-b9cc-4a6a-a270-82cc4811bd4a"}
+ )
+ self.assertEqual(
+ created_items, {"volume:ac408b73-b9cc-4a6a-a270-82cc4811bd4a": True}
+ )
+
+ def test_update_block_device_mapping_with_keep_flag(self):
+ """"""
+ volume = MagicMock(autospec=True)
+ volume.id = volume_id
+ block_device_mapping = {}
+ base_disk_index = 100
+ disk = {"size": 10, "keep": True}
+ created_items = {}
+ self.vimconn.update_block_device_mapping(
+ volume, block_device_mapping, base_disk_index, disk, created_items
+ )
+ self.assertEqual(
+ block_device_mapping, {"vdd": "ac408b73-b9cc-4a6a-a270-82cc4811bd4a"}
+ )
+ self.assertEqual(
+ created_items, {"volume:ac408b73-b9cc-4a6a-a270-82cc4811bd4a:keep": True}
+ )
+
+ def test_extract_items_with_keep_flag_item_has_keep_flag(self):
+ created_items = deepcopy(created_items_all_true)
+ created_items[f"volume:{volume_id2}:keep"] = True
+ result = self.vimconn._extract_items_wth_keep_flag_from_created_items(
+ created_items
+ )
+ self.assertEqual(result, deepcopy(created_items_all_true))
+
+ def test_extract_items_with_keep_flag_no_item_wth_keep_flag(self):
+ created_items = deepcopy(created_items_all_true)
+ result = self.vimconn._extract_items_wth_keep_flag_from_created_items(
+ created_items
+ )
+ self.assertEqual(result, deepcopy(created_items_all_true))
+
+ def test_extract_items_with_keep_flag_all_items_are_already_deleted(self):
+ created_items = {
+ f"port:{port_id}": None,
+ f"floating_ip:{floating_network_vim_id}": None,
+ f"volume:{volume_id}:keep": None,
+ f"volume:{volume_id2}:keep": None,
+ }
+ expected_result = {
+ f"port:{port_id}": None,
+ f"floating_ip:{floating_network_vim_id}": None,
+ }
+ result = self.vimconn._extract_items_wth_keep_flag_from_created_items(
+ created_items
+ )
+ self.assertEqual(result, expected_result)
+
+ def test_extract_items_with_keep_flag_without_semicolon(self):
+ created_items = {
+ f"port{port_id}": True,
+ f"floating_ip{floating_network_vim_id}": None,
+ f"volume{volume_id}keep": True,
+ f"volume{volume_id2}keep": True,
+ }
+ result = self.vimconn._extract_items_wth_keep_flag_from_created_items(
+ created_items
+ )
+ self.assertEqual(result, {})
+
+ def test_extract_items_with_keep_flag_invalid_type_created_items(self):
+ created_items = [{f"port{port_id}": True}, {f"volume{volume_id2}keep": True}]
+ with self.assertRaises(AttributeError):
+ self.vimconn._extract_items_wth_keep_flag_from_created_items(created_items)
if __name__ == "__main__":
projects / osm / RO.git / commitdiff