+ nsr_result = copy.deepcopy(db_nsr["_admin"]["deployed"]["K8s"][0])
+ nsr_kdu_model_result = nsr_result.pop("kdu-model")
+ expected_kdu_model = "stable/openldap:1.2.1"
+ self.assertEqual(nsr_result, k8s_instace_info)
+ self.assertTrue(
+ nsr_kdu_model_result in expected_kdu_model
+ or expected_kdu_model in nsr_kdu_model_result
+ )
+ nsr_result = copy.deepcopy(db_nsr["_admin"]["deployed"]["K8s"][1])
+ nsr_kdu_model_result = nsr_result.pop("kdu-model")
+ k8s_instace_info["kdu-name"] = "mongo"
+ expected_kdu_model = "stable/mongodb"
+ self.assertEqual(nsr_result, k8s_instace_info)
+ self.assertTrue(
+ nsr_kdu_model_result in expected_kdu_model
+ or expected_kdu_model in nsr_kdu_model_result
+ )
+
+ # 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.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"]
+ # # 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.0.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__":