+ async def test_instantiate_pdu(self):
+ nsr_id = descriptors.test_ids["TEST-A"]["ns"]
+ nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
+ # Modify vnfd/vnfr to change KDU for PDU. Adding keys that NBI will already set
+ self.db.set_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "1"},
+ update_dict={"ip-address": "10.205.1.46",
+ "vdur.0.pdu-id": "53e1ec21-2464-451e-a8dc-6e311d45b2c8",
+ "vdur.0.pdu-type": "PDU-TYPE-1",
+ "vdur.0.ip-address": "10.205.1.46",
+ },
+ unset={"vdur.status": None})
+ self.db.set_one("vnfrs", {"nsr-id-ref": nsr_id, "member-vnf-index-ref": "2"},
+ update_dict={"ip-address": "10.205.1.47",
+ "vdur.0.pdu-id": "53e1ec21-2464-451e-a8dc-6e311d45b2c8",
+ "vdur.0.pdu-type": "PDU-TYPE-1",
+ "vdur.0.ip-address": "10.205.1.47",
+ },
+ unset={"vdur.status": None})
+
+ await self.my_ns.instantiate(nsr_id, nslcmop_id)
+ db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
+ self.assertEqual(db_nsr.get("nsState"), "READY", 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_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")
+ db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
+ for vnfr in db_vnfrs_list:
+ self.assertEqual(vnfr["_admin"].get("nsState"), "NOT_INSTANTIATED", "Not instantiated")
+
+ @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")
+