dst_plan = sdc_run_catalog.parse_nsd(ns_inst.nspackage_id, input_parameters)
logger.debug('tosca plan dest: %s' % dst_plan)
logger.debug('Start query nsd(%s)' % ns_inst.nspackage_id)
+ sdc_run_catalog.modify_nsd_state(ns_inst.nspackage_id)
NSInstModel.objects.filter(id=self.ns_inst_id).update(nsd_model=dst_plan)
params_vnf = []
config.WORKFLOW_OPTION = "buildin"
mock_call_req.side_effect = [
[0, {"model": json.dumps(NSD_MODEL_DICT)}, '200'],
+ [0, None, '202'],
[0, VNFM_LIST_IN_AAI_DICT, '200'],
[0, VNFM_IN_AAI_DICT, '200']
]
[0, json.JSONEncoder().encode(items), "200"],
"/api/wso2bpel/v1/process/instance":
[0, json.JSONEncoder().encode(ret), "200"],
+ "/api/catalog/v1/ns_descriptors/1":
+ [0, None, "202"],
}
def side_effect(*args):
mock_vals = {
"api/workflow/v1/process/instance":
[0, json.JSONEncoder().encode(ret), "200"],
+ "/api/catalog/v1/ns_descriptors/1":
+ [0, None, "202"],
}
def side_effect(*args):
def test_ns_instantiate_when_succeed_to_enter_workflow(self, mock_run, mock_call_req):
config.WORKFLOW_OPTION = "buildin"
mock_call_req.side_effect = [
+ [0, None, '202'],
[0, self.vnfms, '200'],
[0, self.vnfm, '200']
]
logger.error("Status code is %s, detail is %s.", ret[2], ret[1])
raise NSLCMException("Failed to query pnf descriptor(%s) from catalog." % pnfdId)
return json.JSONDecoder().decode(ret[1])
+
+
+def modify_nsd_state(csar_id):
+ req_param = json.JSONEncoder().encode({"usageState": 1})
+ ret = req_by_msb("/api/catalog/v1/ns_descriptors/%s" % csar_id, "PUT", req_param)
+ if ret[0] != 0:
+ logger.error("Status code is %s, detail is %s.", ret[2], ret[1])
+ raise NSLCMException("Failed to modify nsd state of CSAR(%s) from catalog." % csar_id)