- 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": "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"
- k8s_instace_info["kdu-model"] = "stable/mongodb"
- self.assertEqual(db_nsr["_admin"]["deployed"]["K8s"][1], k8s_instace_info)
-
- 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")
-
- @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")
-