from osm_common.msgkafka import MsgKafka
from osm_common.fslocal import FsLocal
from osm_lcm.lcm_utils import TaskRegistry
-# from osm_lcm.ROclient import ROClient
+from osm_lcm.ROclient import ROClient
from uuid import uuid4
# from asynctest.mock import patch
self.my_ns.n2vc.GetPrimitiveStatus = asynctest.CoroutineMock(return_value="completed")
self.my_ns.n2vc.GetPrimitiveOutput = asynctest.CoroutineMock(return_value={"result": "ok",
"pubkey": pub_key})
+ self.my_ns.n2vc.delete_execution_environment = asynctest.CoroutineMock(return_value=None)
self.my_ns.n2vc.get_public_key = asynctest.CoroutineMock(
return_value=getenv("OSMLCM_VCA_PUBKEY", "public_key"))
self.my_ns.n2vc.delete_namespace = asynctest.CoroutineMock(return_value=None)
nsd["constituent-vnfd"].append({"vnfd-id-ref": k, "member-vnf-index": k})
n2vc_key_list = ["key"]
- ro_ns_params = self.my_ns.ns_params_2_RO(ns_params, nsd, vnfd_dict, n2vc_key_list)
+ ro_ns_params = self.my_ns._ns_params_2_RO(ns_params, nsd, vnfd_dict, n2vc_key_list)
ro_params_expected = {'wim_account': None, "datacenter": ro_vim_id,
"vnfs": {"5": {"vdus": {"vdu_id": {"mgmt_keys": n2vc_key_list}}}}}
self.assertEqual(ro_ns_params, ro_params_expected)
self.assertEqual(return_value, expected_value)
# print("scale_result: {}".format(self.db.get_one("nslcmops", {"_id": nslcmop_id}).get("detailed-status")))
- # Test _reintent_or_skip_suboperation()
+ # Test _retry_or_skip_suboperation()
# Expected result:
# - if a suboperation's 'operationState' is marked as 'COMPLETED', SUBOPERATION_STATUS_SKIP is expected
# - if marked as anything but 'COMPLETED', the suboperation index is expected
- def test_scale_reintent_or_skip_suboperation(self):
+ def test_scale_retry_or_skip_suboperation(self):
# Load an alternative 'nslcmops' YAML for this test
nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
db_nslcmop = self.db.get_one('nslcmops', {"_id": nslcmop_id})
op_index = 2
# Test when 'operationState' is 'COMPLETED'
db_nslcmop['_admin']['operations'][op_index]['operationState'] = 'COMPLETED'
- return_value = self.my_ns._reintent_or_skip_suboperation(db_nslcmop, op_index)
+ return_value = self.my_ns._retry_or_skip_suboperation(db_nslcmop, op_index)
expected_value = self.my_ns.SUBOPERATION_STATUS_SKIP
self.assertEqual(return_value, expected_value)
# Test when 'operationState' is not 'COMPLETED'
db_nslcmop['_admin']['operations'][op_index]['operationState'] = None
- return_value = self.my_ns._reintent_or_skip_suboperation(db_nslcmop, op_index)
+ return_value = self.my_ns._retry_or_skip_suboperation(db_nslcmop, op_index)
expected_value = op_index
self.assertEqual(return_value, expected_value)
async def test_deploy_kdus(self):
nsr_id = descriptors.test_ids["TEST-KDU"]["ns"]
- # nslcmop_id = descriptors.test_ids["TEST-KDU"]["instantiate"]
+ nslcmop_id = descriptors.test_ids["TEST-KDU"]["instantiate"]
db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
db_vnfr = self.db.get_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "multikdu"})
db_vnfrs = {"multikdu": db_vnfr}
db_vnfd = self.db.get_one("vnfds", {"_id": db_vnfr["vnfd-id"]})
db_vnfds = {db_vnfd["_id"]: db_vnfd}
+ task_register = {}
logging_text = "KDU"
self.my_ns.k8sclusterhelm.install = asynctest.CoroutineMock(return_value="k8s_id")
self.my_ns.k8sclusterhelm.synchronize_repos = asynctest.CoroutineMock(return_value=("", ""))
- await self.my_ns.deploy_kdus(logging_text, nsr_id, db_nsr, db_vnfrs, db_vnfds)
+ await self.my_ns.deploy_kdus(logging_text, nsr_id, nslcmop_id, db_vnfrs, db_vnfds, task_register)
+ await asyncio.wait(list(task_register.keys()), timeout=100)
db_nsr = self.db.get_list("nsrs")[1]
self.assertIn("K8s", db_nsr["_admin"]["deployed"], "K8s entry not created at '_admin.deployed'")
self.assertIsInstance(db_nsr["_admin"]["deployed"]["K8s"], list, "K8s entry is not of type list")
self.assertEqual(len(db_nsr["_admin"]["deployed"]["K8s"]), 2, "K8s entry is not of type list")
- k8s_instace_info = {"kdu-instance": "k8s_id", "k8scluster-uuid": "73d96432-d692-40d2-8440-e0c73aee209c",
- "k8scluster-type": "chart",
- "kdu-name": "ldap", "kdu-model": "stable/openldap:1.2.1"}
+ k8s_instace_info = {"kdu-instance": None, "k8scluster-uuid": "73d96432-d692-40d2-8440-e0c73aee209c",
+ "k8scluster-type": "helm-chart",
+ "kdu-name": "ldap", "kdu-model": "stable/openldap:1.2.1",
+ "member-vnf-index": "multikdu", "namespace": None}
self.assertEqual(db_nsr["_admin"]["deployed"]["K8s"][0], k8s_instace_info)
k8s_instace_info["kdu-name"] = "mongo"
self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
+ @asynctest.fail_on(active_handles=True) # all async tasks must be completed
+ async def test_terminate_without_configuration(self):
+ nsr_id = descriptors.test_ids["TEST-A"]["ns"]
+ nslcmop_id = descriptors.test_ids["TEST-A"]["terminate"]
+ # set instantiation task as completed
+ self.db.set_list("nslcmops", {"nsInstanceId": nsr_id, "_id.ne": nslcmop_id},
+ update_dict={"operationState": "COMPLETED"})
+ self.my_ns.RO.show = asynctest.CoroutineMock(ROClient.show, side_effect=self._ro_show(delete=nslcmop_id))
+ self.db.set_one("nsrs", {"_id": nsr_id},
+ update_dict={"_admin.deployed.VCA.0": None, "_admin.deployed.VCA.1": None})
+
+ await self.my_ns.terminate(nsr_id, nslcmop_id)
+ db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
+ self.assertEqual(db_nslcmop.get("operationState"), 'COMPLETED', db_nslcmop.get("detailed-status"))
+ db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
+ self.assertEqual(db_nsr.get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
+ self.assertEqual(db_nsr["_admin"].get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
+ self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
+ self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
+ self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
+ self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
+
+ @asynctest.fail_on(active_handles=True) # all async tasks must be completed
+ async def test_terminate_primitive(self):
+ nsr_id = descriptors.test_ids["TEST-A"]["ns"]
+ nslcmop_id = descriptors.test_ids["TEST-A"]["terminate"]
+ self.my_ns.RO.show = asynctest.CoroutineMock(ROClient.show, side_effect=self._ro_show(delete=nslcmop_id))
+ # set instantiation task as completed
+ self.db.set_list("nslcmops", {"nsInstanceId": nsr_id, "_id.ne": nslcmop_id},
+ update_dict={"operationState": "COMPLETED"})
+
+ # modify vnfd descriptor to include terminate_primitive
+ terminate_primitive = [{
+ "name": "touch",
+ "parameter": [{"name": "filename", "value": "terminate_filename"}],
+ "seq": '1'
+ }]
+ db_vnfr = self.db.get_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "1"})
+ self.db.set_one("vnfds", {"_id": db_vnfr["vnfd-id"]},
+ {"vnf-configuration.terminate-config-primitive": terminate_primitive})
+
+ await self.my_ns.terminate(nsr_id, nslcmop_id)
+ db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
+ self.assertEqual(db_nslcmop.get("operationState"), 'COMPLETED', db_nslcmop.get("detailed-status"))
+ db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
+ self.assertEqual(db_nsr.get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
+ self.assertEqual(db_nsr["_admin"].get("nsState"), "NOT_INSTANTIATED", str(db_nsr.get("errorDescription ")))
+ self.assertEqual(db_nsr.get("currentOperation"), "IDLE", "currentOperation different than 'IDLE'")
+ self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
+ self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
+ self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
+
if __name__ == '__main__':
asynctest.main()
projects / osm / LCM.git / blobdiff