X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_nbi%2Ftests%2Ftest_base_topic.py;h=d5863a64c4175d4b5e3093f8783b54f2a1da85a1;hb=514b546acfeedf03541574394f934e06335f16ee;hp=5107680064a9422ecf7b1465a53cb3760708e23f;hpb=4568a372eb5a204e04d917213de03ec51f9110c1;p=osm%2FNBI.git diff --git a/osm_nbi/tests/test_base_topic.py b/osm_nbi/tests/test_base_topic.py index 5107680..d5863a6 100755 --- a/osm_nbi/tests/test_base_topic.py +++ b/osm_nbi/tests/test_base_topic.py @@ -17,12 +17,23 @@ __author__ = "Alfonso Tierno, alfonso.tiernosepulveda@telefonica.com" __date__ = "2020-06-17" +from copy import deepcopy import unittest from unittest import TestCase +from unittest.mock import patch, Mock +from osm_nbi.base_topic import ( + BaseTopic, + EngineException, + NBIBadArgumentsException, + detect_descriptor_usage, + update_descriptor_usage_state, +) +from osm_common import dbbase +from osm_nbi.tests.test_pkg_descriptors import db_vnfds_text, db_nsds_text +import yaml -# from unittest.mock import Mock -# from osm_common import dbbase, fsbase, msgbase -from osm_nbi.base_topic import BaseTopic, EngineException +db_vnfd_content = yaml.safe_load(db_vnfds_text)[0] +db_nsd_content = yaml.safe_load(db_nsds_text)[0] class Test_BaseTopic(TestCase): @@ -35,14 +46,9 @@ class Test_BaseTopic(TestCase): pass def setUp(self): - pass - # self.db = Mock(dbbase.DbBase()) - # self.fs = Mock(fsbase.FsBase()) - # self.msg = Mock(msgbase.MsgBase()) - # self.auth = Mock(authconn.Authconn(None, None, None)) + self.db = Mock(dbbase.DbBase()) def test_update_input_with_kwargs(self): - test_set = ( # (descriptor content, kwargs, expected descriptor (None=fails), message) ( @@ -123,6 +129,110 @@ class Test_BaseTopic(TestCase): BaseTopic._update_input_with_kwargs(desc, kwargs) self.assertEqual(desc, expected, message) + def test_detect_descriptor_usage_empty_descriptor(self): + descriptor = {} + db_collection = "vnfds" + with self.assertRaises(EngineException) as error: + detect_descriptor_usage(descriptor, db_collection, self.db) + self.assertIn( + "Argument is mandatory and can not be empty, Bad arguments: descriptor", + error, + "Error message is wrong.", + ) + self.db.get_list.assert_not_called() + + def test_detect_descriptor_usage_empty_db_argument(self): + descriptor = deepcopy(db_vnfd_content) + db_collection = "vnfds" + db = None + with self.assertRaises(EngineException) as error: + detect_descriptor_usage(descriptor, db_collection, db) + self.assertIn( + "A valid DB object should be provided, Bad arguments: db", + error, + "Error message is wrong.", + ) + self.db.get_list.assert_not_called() + + def test_detect_descriptor_usage_which_is_in_use(self): + descriptor = deepcopy(db_vnfd_content) + db_collection = "vnfds" + self.db.get_list.side_effect = [deepcopy(db_vnfd_content)] + expected = True + result = detect_descriptor_usage(descriptor, db_collection, self.db) + self.assertEqual(result, expected, "wrong result") + self.db.get_list.assert_called_once_with( + "vnfrs", {"vnfd-id": descriptor["_id"]} + ) + + def test_detect_descriptor_usage_which_is_not_in_use(self): + descriptor = deepcopy(db_nsd_content) + self.db.get_list.return_value = [] + db_collection = "nsds" + expected = None + result = detect_descriptor_usage(descriptor, db_collection, self.db) + self.assertEqual(result, expected, "wrong result") + self.db.get_list.assert_called_once_with("nsrs", {"nsd-id": descriptor["_id"]}) + + def test_detect_descriptor_usage_wrong_desc_format(self): + descriptor = deepcopy(db_nsd_content) + descriptor.pop("_id") + db_collection = "nsds" + with self.assertRaises(EngineException) as error: + detect_descriptor_usage(descriptor, db_collection, self.db) + self.assertIn("KeyError", error, "wrong error type") + self.db.get_list.assert_not_called() + + def test_detect_descriptor_usage_wrong_db_collection(self): + descriptor = deepcopy(db_vnfd_content) + descriptor.pop("_id") + db_collection = "vnf" + with self.assertRaises(EngineException) as error: + detect_descriptor_usage(descriptor, db_collection, self.db) + self.assertIn( + "db_collection should be equal to vnfds or nsds, db_collection", + error, + "wrong error type", + ) + + self.db.get_list.assert_not_called() + + @patch("osm_nbi.base_topic.detect_descriptor_usage") + def test_update_descriptor_usage_state_to_in_use(self, mock_descriptor_usage): + db_collection = "vnfds" + descriptor = deepcopy(db_vnfd_content) + mock_descriptor_usage.return_value = True + descriptor_update = {"_admin.usageState": "IN_USE"} + update_descriptor_usage_state(descriptor, db_collection, self.db) + self.db.set_one.assert_called_once_with( + db_collection, {"_id": descriptor["_id"]}, update_dict=descriptor_update + ) + + @patch("osm_nbi.base_topic.detect_descriptor_usage") + def test_update_descriptor_usage_state_to_not_in_use(self, mock_descriptor_usage): + db_collection = "nsds" + descriptor = deepcopy(db_nsd_content) + mock_descriptor_usage.return_value = False + descriptor_update = {"_admin.usageState": "NOT_IN_USE"} + update_descriptor_usage_state(descriptor, db_collection, self.db) + self.db.set_one.assert_called_once_with( + db_collection, {"_id": descriptor["_id"]}, update_dict=descriptor_update + ) + + @patch("osm_nbi.base_topic.detect_descriptor_usage") + def test_update_descriptor_usage_state_db_exception(self, mock_descriptor_usage): + db_collection = "nsd" + descriptor = deepcopy(db_nsd_content) + mock_descriptor_usage.side_effect = NBIBadArgumentsException + with self.assertRaises(EngineException) as error: + update_descriptor_usage_state(descriptor, db_collection, self.db) + self.assertIn( + "db_collection should be equal to vnfds or nsds, db_collection", + error, + "wrong error type", + ) + self.db.set_one.assert_not_called() + if __name__ == "__main__": unittest.main()