__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 Mock
-# from osm_common import dbbase, fsbase, msgbase
-from osm_nbi.base_topic import BaseTopic, EngineException
+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
+db_vnfd_content = yaml.safe_load(db_vnfds_text)[0]
+db_nsd_content = yaml.safe_load(db_nsds_text)[0]
-class Test_BaseTopic(TestCase):
+class Test_BaseTopic(TestCase):
@classmethod
def setUpClass(cls):
cls.test_name = "test-base-topic"
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)
- ({"a": {"none": None}}, {"a.b.num": "v"}, {"a": {"none": None, "b": {"num": "v"}}}, "create dict"),
- ({"a": {"none": None}}, {"a.none.num": "v"}, {"a": {"none": {"num": "v"}}}, "create dict over none"),
- ({"a": {"b": {"num": 4}}}, {"a.b.num": "v"}, {"a": {"b": {"num": "v"}}}, "replace_number"),
- ({"a": {"b": {"num": 4}}}, {"a.b.num.c.d": "v"}, {"a": {"b": {"num": {"c": {"d": "v"}}}}},
- "create dict over number"),
- ({"a": {"b": {"num": 4}}}, {"a.b": "v"}, {"a": {"b": "v"}}, "replace dict with a string"),
- ({"a": {"b": {"num": 4}}}, {"a.b": None}, {"a": {}}, "replace dict with None"),
- ({"a": [{"b": {"num": 4}}]}, {"a.b.num": "v"}, None, "create dict over list should fail"),
- ({"a": [{"b": {"num": 4}}]}, {"a.0.b.num": "v"}, {"a": [{"b": {"num": "v"}}]}, "set list"),
- ({"a": [{"b": {"num": 4}}]}, {"a.3.b.num": "v"},
- {"a": [{"b": {"num": 4}}, None, None, {"b": {"num": "v"}}]}, "expand list"),
+ (
+ {"a": {"none": None}},
+ {"a.b.num": "v"},
+ {"a": {"none": None, "b": {"num": "v"}}},
+ "create dict",
+ ),
+ (
+ {"a": {"none": None}},
+ {"a.none.num": "v"},
+ {"a": {"none": {"num": "v"}}},
+ "create dict over none",
+ ),
+ (
+ {"a": {"b": {"num": 4}}},
+ {"a.b.num": "v"},
+ {"a": {"b": {"num": "v"}}},
+ "replace_number",
+ ),
+ (
+ {"a": {"b": {"num": 4}}},
+ {"a.b.num.c.d": "v"},
+ {"a": {"b": {"num": {"c": {"d": "v"}}}}},
+ "create dict over number",
+ ),
+ (
+ {"a": {"b": {"num": 4}}},
+ {"a.b": "v"},
+ {"a": {"b": "v"}},
+ "replace dict with a string",
+ ),
+ (
+ {"a": {"b": {"num": 4}}},
+ {"a.b": None},
+ {"a": {}},
+ "replace dict with None",
+ ),
+ (
+ {"a": [{"b": {"num": 4}}]},
+ {"a.b.num": "v"},
+ None,
+ "create dict over list should fail",
+ ),
+ (
+ {"a": [{"b": {"num": 4}}]},
+ {"a.0.b.num": "v"},
+ {"a": [{"b": {"num": "v"}}]},
+ "set list",
+ ),
+ (
+ {"a": [{"b": {"num": 4}}]},
+ {"a.3.b.num": "v"},
+ {"a": [{"b": {"num": 4}}, None, None, {"b": {"num": "v"}}]},
+ "expand list",
+ ),
({"a": [[4]]}, {"a.0.0": "v"}, {"a": [["v"]]}, "set nested list"),
- ({"a": [[4]]}, {"a.0.2": "v"}, {"a": [[4, None, "v"]]}, "expand nested list"),
- ({"a": [[4]]}, {"a.2.2": "v"}, {"a": [[4], None, {"2": "v"}]}, "expand list and add number key"),
+ (
+ {"a": [[4]]},
+ {"a.0.2": "v"},
+ {"a": [[4, None, "v"]]},
+ "expand nested list",
+ ),
+ (
+ {"a": [[4]]},
+ {"a.2.2": "v"},
+ {"a": [[4], None, {"2": "v"}]},
+ "expand list and add number key",
+ ),
({"a": None}, {"b.c": "v"}, {"a": None, "b": {"c": "v"}}, "expand at root"),
)
for desc, kwargs, expected, message in test_set:
if expected is None:
- self.assertRaises(EngineException, BaseTopic._update_input_with_kwargs, desc, kwargs)
+ self.assertRaises(
+ EngineException, BaseTopic._update_input_with_kwargs, desc, kwargs
+ )
else:
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__':
+if __name__ == "__main__":
unittest.main()
projects / osm / NBI.git / blobdiff