+ # Test scale() for native kdu
+ # this also includes testing _scale_kdu()
+ nsr_id = descriptors.test_ids["TEST-NATIVE-KDU"]["ns"]
+ nslcmop_id = descriptors.test_ids["TEST-NATIVE-KDU"]["instantiate"]
+
+ self.my_ns.k8sclusterjuju.scale = asynctest.mock.CoroutineMock()
+ self.my_ns.k8sclusterjuju.exec_primitive = asynctest.mock.CoroutineMock()
+ self.my_ns.k8sclusterjuju.get_scale_count = asynctest.mock.CoroutineMock(
+ return_value=1
+ )
+ await self.my_ns.scale(nsr_id, nslcmop_id)
+ expected_value = "COMPLETED"
+ return_value = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
+ "operationState"
+ )
+ self.assertEqual(return_value, expected_value)
+ self.my_ns.k8sclusterjuju.scale.assert_called_once()
+
+ # Test scale() for native kdu with 2 resource
+ nsr_id = descriptors.test_ids["TEST-NATIVE-KDU-2"]["ns"]
+ nslcmop_id = descriptors.test_ids["TEST-NATIVE-KDU-2"]["instantiate"]
+
+ self.my_ns.k8sclusterjuju.get_scale_count.return_value = 2
+ await self.my_ns.scale(nsr_id, nslcmop_id)
+ expected_value = "COMPLETED"
+ return_value = self.db.get_one("nslcmops", {"_id": nslcmop_id}).get(
+ "operationState"
+ )
+ self.assertEqual(return_value, expected_value)
+ self.my_ns.k8sclusterjuju.scale.assert_called()
+
+ async def test_vca_status_refresh(self):
+ nsr_id = descriptors.test_ids["TEST-A"]["ns"]
+ nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
+ await self.my_ns.vca_status_refresh(nsr_id, nslcmop_id)
+ expected_value = dict()
+ return_value = dict()
+ vnf_descriptors = self.db.get_list("vnfds")
+ for i, _ in enumerate(vnf_descriptors):
+ for j, value in enumerate(vnf_descriptors[i]["df"]):
+ if "lcm-operations-configuration" in vnf_descriptors[i]["df"][j]:
+ if (
+ "day1-2"
+ in value["lcm-operations-configuration"][
+ "operate-vnf-op-config"
+ ]
+ ):
+ for k, v in enumerate(
+ value["lcm-operations-configuration"][
+ "operate-vnf-op-config"
+ ]["day1-2"]
+ ):
+ if (
+ v.get("execution-environment-list")
+ and "juju" in v["execution-environment-list"][k]
+ ):
+ expected_value = self.db.get_list("nsrs")[i][
+ "vcaStatus"
+ ]
+ await self.my_ns._on_update_n2vc_db(
+ "nsrs", {"_id": nsr_id}, "_admin.deployed.VCA.0", {}
+ )
+ return_value = self.db.get_list("nsrs")[i]["vcaStatus"]
+ self.assertEqual(return_value, expected_value)
+