#######################################################################################
import logging
-from types import ModuleType
import unittest
-from unittest.mock import Mock, patch
+from unittest.mock import MagicMock, patch
from osm_ng_ro.ns_thread import VimInteractionNet
+from osm_ro_plugin.vimconn import VimConnConnectionException, VimConnException
class TestVimInteractionNet(unittest.TestCase):
- @classmethod
- def setUpClass(cls):
+ def setUp(self):
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
+ 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__mgmt_net_id_in_find_params_mgmt_several_vim_nets(self):
"""
- management_network_id in find_params.get('mgmt')
+ mgmt network is set in find_params
+ management_network_id in vim config
More than one network found in the VIM
"""
db = "test_db"
logger = "test_logger"
my_vims = "test-vim"
db_vims = {
- "vim_openstack_2": {
+ 0: {
"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
- ):
+ ), patch.object(instance, "db_vims", db_vims):
ro_task = {
"target_id": 0,
"tasks": {
"task_index_2": {
- "target_id": "vim_openstack_2",
+ "target_id": 0,
"action_id": "123456",
"nsr_id": "654321",
"task_id": "123456:1",
"target_record": "test_target_record",
"target_record_id": "test_target_record_id",
# values coming from extra_dict
- "params": "",
+ "params": {},
"find_params": {
"mgmt": True,
"name": "some_mgmt_name",
def test__mgmt_net_id_in_find_params_mgmt_no_vim_nets(self):
"""
- management_network_id in find_params.get('mgmt')
+ mgmt network is set in find_params
+ management_network_id in vim config
The network could not be found in the VIM
"""
db = "test_db"
logger = "test_logger"
my_vims = "test-vim"
db_vims = {
- "vim_openstack_3": {
+ 0: {
"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
- ):
+ instance, "db_vims", db_vims
+ ), patch.object(instance, "logger", logging):
ro_task = {
"target_id": 0,
"tasks": {
"task_index_3": {
- "target_id": "vim_openstack_3",
+ "target_id": 0,
"action_id": "123456",
"nsr_id": "654321",
"task_id": "123456:1",
"item": "test_item",
"target_record": "test_target_record",
"target_record_id": "test_target_record_id",
- "params": "",
+ "params": {},
# values coming from extra_dict
"find_params": {
"mgmt": True,
self.assertEqual(result[1].get("created"), False)
self.assertEqual(result[1].get("vim_status"), "VIM_ERROR")
+ def test__mgmt_net_in_find_params_no_vim_config_no_vim_nets(self):
+ """
+ mgmt network is set in find_params
+ vim config does not have management_network_id or management_network_id
+ The network could not be found in the VIM
+ """
+ db = "test_db"
+ logger = "test_logger"
+ my_vims = "test-vim"
+ db_vims = {
+ 0: {
+ "config": {},
+ },
+ }
+
+ instance = VimInteractionNet(db, logger, my_vims, db_vims)
+ with patch.object(instance, "my_vims", [self.target_vim]), patch.object(
+ instance, "db_vims", db_vims
+ ), patch.object(instance, "logger", logging):
+ ro_task = {
+ "target_id": 0,
+ "tasks": {
+ "task_index_3": {
+ "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",
+ "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 = []
+ 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.assertDictEqual(
+ result[1].get("created_items"), {"item1": "sample_created_item"}
+ )
+ self.assertEqual(result[1].get("vim_status"), "BUILD")
+
def test__mgmt_net_name_in_find_params_mgmt_several_vim_nets(self):
"""
- management_network_name in find_params.get('mgmt')
+ mgmt network is set in find_params
+ management_network_name in vim config
More than one network found in the VIM
"""
db = "test_db"
logger = "test_logger"
my_vims = "test-vim"
db_vims = {
- "vim_openstack_4": {
+ 0: {
"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
- ):
+ ), patch.object(instance, "db_vims", db_vims):
ro_task = {
"target_id": 0,
"tasks": {
"task_index_4": {
- "target_id": "vim_openstack_4",
+ "target_id": 0,
"action_id": "123456",
"nsr_id": "654321",
"task_id": "123456:1",
"target_record": "test_target_record",
"target_record_id": "test_target_record_id",
# values coming from extra_dict
- "params": "",
+ "params": {},
"find_params": {
"mgmt": True,
"name": "some_mgmt_name",
def test__mgmt_net_name_in_find_params_mgmt_no_vim_nets(self):
"""
- management_network_name in find_params.get('mgmt')
+ mgmt network is set in find_params
+ management_network_name in vim config
The network could not be found in the VIM
"""
db = "test_db"
logger = "test_logger"
my_vims = "test-vim"
db_vims = {
- "vim_openstack_5": {
+ 0: {
"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
- ):
+ ), patch.object(instance, "db_vims", db_vims):
ro_task = {
"target_id": 0,
"tasks": {
"task_index_5": {
- "target_id": "vim_openstack_5",
+ "target_id": 0,
"action_id": "123456",
"nsr_id": "654321",
"task_id": "123456:1",
"target_record": "test_target_record",
"target_record_id": "test_target_record_id",
# values coming from extra_dict
- "params": "",
+ "params": {},
"find_params": {
"mgmt": True,
"name": "some_mgmt_name",
def test__mgmt_net_name_in_find_params_filterdict_several_vim_nets(self):
"""
- management_network_name in find_params.get('filterdict')
+ mgmt network is set in find_params
+ management_network_name in vim config
+ network_name is set 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": {
+ 0: {
"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
- ):
+ ), patch.object(instance, "db_vims", db_vims):
ro_task = {
"target_id": 0,
"tasks": {
"task_index_6": {
- "target_id": "vim_openstack_6",
+ "target_id": 0,
"action_id": "123456",
"nsr_id": "654321",
"task_id": "123456:1",
"target_record": "test_target_record",
"target_record_id": "test_target_record_id",
# values coming from extra_dict
- "params": "",
+ "params": {},
"find_params": {
"filter_dict": {
"name": "some-network-name",
logger = "test_logger"
my_vims = "test-vim"
db_vims = {
- "vim_openstack_7": {
+ 0: {
"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
- ):
+ ), patch.object(instance, "db_vims", db_vims):
ro_task = {
"target_id": 0,
"tasks": {
"task_index_4": {
- "target_id": "vim_openstack_7",
+ "target_id": 0,
"action_id": "123456",
"nsr_id": "654321",
"task_id": "123456:1",
"target_record": "test_target_record",
"target_record_id": "test_target_record_id",
# values coming from extra_dict
- "params": "",
+ "params": {},
"find_params": {"wrong_param": "wrong_value"},
"depends_on": "test_depends_on",
},
logger = "test_logger"
my_vims = "test-vim"
db_vims = {
- "vim_openstack_8": {
+ 0: {
"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
- ):
+ ), patch.object(instance, "db_vims", db_vims):
ro_task = {
"target_id": 0,
"tasks": {
"task_index_8": {
- "target_id": "vim_openstack_8",
+ "target_id": 0,
"action_id": "123456",
"nsr_id": "654321",
"task_id": "123456:1",
"target_record": "test_target_record",
"target_record_id": "test_target_record_id",
# values coming from extra_dict
- "params": "test_params",
+ "params": {
+ "net_name": "test_params",
+ },
"find_params": {
"filter_dict": {
"name": "some-network-name",
def test__mgmt_net_name_in_find_params_filterdict_no_vim_nets(self):
"""
- management_network_name in find_params.get('filterdict')
+ mgmt network is set in find_params
+ management_network_name in vim config
+ network_name is set 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": {
+ 0: {
"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
- ):
+ ), patch.object(instance, "db_vims", db_vims):
ro_task = {
"target_id": 0,
"tasks": {
"task_index_9": {
- "target_id": "vim_openstack_9",
+ "target_id": 0,
"action_id": "123456",
"nsr_id": "654321",
"task_id": "123456:1",
self.assertEqual(result[1].get("created"), False)
self.assertEqual(result[1].get("vim_status"), "VIM_ERROR")
+ def test__mgmt_net_in_find_params_filterdict_no_config_no_vim_nets(self):
+ """
+ mgmt network is set in find_params
+ vim config is empty
+ network_name is set 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 = {
+ 0: {
+ "config": {},
+ },
+ }
+ instance = VimInteractionNet(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_9": {
+ "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": {},
+ "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
+ mgmt network is set in find_params
management_network_name is not in db_vims.get('config')
One network found in the VIM
"""
logger = "test_logger"
my_vims = "test-vim"
db_vims = {
- "vim_openstack_10": {
+ 0: {
"config": {},
},
}
instance = VimInteractionNet(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_2": {
- "target_id": "vim_openstack_10",
+ "target_id": 0,
"action_id": "123456",
"nsr_id": "654321",
"task_id": "123456:1",
"target_record": "test_target_record",
"target_record_id": "test_target_record_id",
# values coming from extra_dict
- "params": "",
+ "params": {},
"find_params": {
"mgmt": True,
"name": "some_mgmt_name",
logger = "test_logger"
my_vims = "test-vim"
db_vims = {
- "vim_openstack_11": {
+ 0: {
"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
- ):
+ ), patch.object(instance, "db_vims", db_vims):
ro_task = {
"target_id": 0,
"tasks": {
"task_index_11": {
- "target_id": "vim_openstack_11",
+ "target_id": 0,
"action_id": "123456",
"nsr_id": "654321",
"task_id": "123456:1",
logger = "test_logger"
my_vims = "test-vim"
db_vims = {
- "vim_openstack_13": {
+ 0: {
"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
- ):
+ ), patch.object(instance, "db_vims", db_vims):
ro_task = {
"target_id": 0,
"tasks": {
"task_index_12": {
- "target_id": "vim_openstack_13",
+ "target_id": 0,
"action_id": "123456",
"nsr_id": "654321",
"task_id": "123456:1",
"target_record": "test_target_record",
"target_record_id": "test_target_record_id",
# values coming from extra_dict
- "params": "",
+ "params": {},
"depends_on": "test_depends_on",
},
},
}
task_index = "task_index_12"
- with self.assertRaises(TypeError):
+ self.target_vim.new_network.side_effect = VimConnConnectionException(
+ "VimConnConnectionException occurred."
+ )
+ with self.assertLogs() as captured:
instance.new(ro_task, task_index, self.task_depends)
+ self.assertEqual(captured.records[0].levelname, "ERROR")
def test__refresh_ro_task_vim_status_active(self):
"""
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):
+ ), patch.object(instance, "logger", logging), patch.object(
+ instance, "db_vims", db_vims
+ ):
ro_task = {
"_id": "122436:1",
"locked_by": None,
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):
+ ), patch.object(instance, "logger", logging), patch.object(
+ instance, "db_vims", db_vims
+ ):
ro_task = {
"_id": "122436:1",
"locked_by": None,
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):
+ ), patch.object(instance, "logger", logging), patch.object(
+ instance, "db_vims", db_vims
+ ):
ro_task = {
"_id": "122436:1",
"locked_by": None,
self.assertEqual(result[0], task_status)
self.assertDictEqual(result[1], ro_vim_item_update)
- def test__refresh_ro_task_VimConnException_occured(self):
+ def test__refresh_ro_task_VimConnException_occurred(self):
"""
vimconn.VimConnException has occured
"""
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):
+ ), patch.object(instance, "logger", logging), patch.object(
+ instance, "db_vims", db_vims
+ ):
ro_task = {
"_id": "122436:1",
"locked_by": None,
"to_check_at": 1637324200.994312,
"tasks": {},
}
- self.target_vim.refresh_nets_status.side_effect = Mock(
- side_effect=self.VimConnException("VimConnException occured")
+ self.target_vim.refresh_nets_status.side_effect = VimConnException(
+ "VimConnException occurred."
)
- with self.assertRaises(TypeError):
+ with self.assertLogs() as captured:
instance.refresh(ro_task)
- self.target_vim.refresh_nets_status.side_effect = None
+ self.assertEqual(captured.records[0].levelname, "ERROR")
def test__refresh_ro_task_vim_status_deleted(self):
"""
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):
+ ), patch.object(instance, "logger", logging), patch.object(
+ instance, "db_vims", db_vims
+ ):
ro_task = {
"_id": "122436:1",
"locked_by": None,
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):
+ ), patch.object(instance, "logger", logging), patch.object(
+ instance, "db_vims", db_vims
+ ):
ro_task = {
"_id": "128436:1",
"locked_by": None,
projects / osm / RO.git / blobdiff