import logging
from lcm.ns.const import OWNER_TYPE
+from lcm.pub.utils import restcall
from lcm.pub.database.models import NSInstModel, NfInstModel, VLInstModel, CPInstModel, VNFFGInstModel
logger = logging.getLogger(__name__)
'nsdId': ns_inst.nsd_id,
'nsdInvariantId': ns_inst.nsd_invariant_id,
'vnfInfo': self.get_vnf_infos(ns_inst.id),
+ 'pnfInfo': self.get_pnf_infos(ns_inst.id),
'vlInfo': self.get_vl_infos(ns_inst.id),
'vnffgInfo': self.get_vnffg_infos(ns_inst.id, ns_inst.nsd_model),
'nsState': ns_inst.status}
return [{
'vnffgInstanceId': vnffg.vnffginstid,
'vnfId': self.convert_string_to_list(vnffg.vnflist),
- 'pnfId': self.get_pnf_infos(nsd_model),
+ 'pnfId': self.get_pnf_ids(nsd_model),
'virtualLinkId': self.convert_string_to_list(vnffg.vllist),
'cpId': self.convert_string_to_list(vnffg.cplist),
'nfp': self.convert_string_to_list(vnffg.fplist)} for vnffg in vnffgs]
@staticmethod
- def get_pnf_infos(nsd_model):
+ def get_pnf_ids(nsd_model):
context = json.loads(nsd_model)
pnfs = context['pnfs']
return [pnf['pnf_id'] for pnf in pnfs]
if not detail_id_string:
return None
return detail_id_string.split(',')
+
+ @staticmethod
+ def get_pnf_infos(ns_instance_id):
+ uri = "api/nslcm/v1/pnfs?nsInstanceId=%s" % ns_instance_id
+ ret = restcall.req_by_msb(uri, "GET")
+ if ret[0] == 0:
+ return json.loads(ret[1])
+ else:
+ return []
# limitations under the License.
from rest_framework import serializers
+from lcm.ns_pnfs.serializers.pnf_serializer import PnfInstanceSerializer
class ContextSerializer(serializers.Serializer):
description = serializers.CharField(help_text="Description of NS instance", required=False, allow_null=True)
nsdId = serializers.CharField(help_text="ID of NSD", required=True)
vnfInfo = VnfInstSerializer(help_text="VNF instances", many=True, required=False, allow_null=True)
+ pnfInfo = PnfInstanceSerializer(help_text="PNF instances", many=True, required=False, allow_null=True)
vlInfo = VlInstSerializer(help_text="VL instances", many=True, required=False, allow_null=True)
vnffgInfo = VnffgInstSerializer(help_text="VNFFG instances", many=True, required=False, allow_null=True)
nsState = serializers.CharField(help_text="State of NS instance", required=False, allow_null=True)
).save()
response = self.client.get("/api/nslcm/v1/pnfs/%s" % pnfId)
self.assertEqual(status.HTTP_200_OK, response.status_code)
+
+ def test_get_fileter_nsinstance_restapi(self):
+ pnfId = str(uuid.uuid4())
+ nsInstanceId = str(uuid.uuid4())
+ PNFInstModel(pnfId=pnfId,
+ pnfName="Test PNF",
+ pnfdId=str(uuid.uuid4()),
+ pnfdInfoId=str(uuid.uuid4()),
+ pnfProfileId=str(uuid.uuid4()),
+ cpInfo=[{
+ "cpInstanceId": str(uuid.uuid4()),
+ "cpdId": "pnf_ext_cp01",
+ "cpProtocolData": []
+ }],
+ emsId=str(uuid.uuid4()),
+ nsInstances=nsInstanceId
+ ).save()
+ response = self.client.get("/api/nslcm/v1/pnfs?nsInstanceId=%s" % nsInstanceId)
+ self.assertEqual(status.HTTP_200_OK, response.status_code)
+ self.assertEqual(1, len(response.data))
+ self.assertEqual(pnfId, response.data[0]['pnfId'])
def get(self, request):
try:
logger.debug("PnfView::get")
- pnfInstDataSet = GetPnf().do_biz()
+ nsInstanceId = request.query_params.get('nsInstanceId', None)
+ if nsInstanceId is not None:
+ filter = {"nsInstanceId": nsInstanceId}
+ pnfInstDataSet = GetPnf(filter).do_biz()
+ else:
+ pnfInstDataSet = GetPnf().do_biz()
logger.debug("PnfView::get::ret=%s", pnfInstDataSet)
resp_serializer = PnfInstancesSerializer(data=[pnfInstData.__dict__ for pnfInstData in pnfInstDataSet])
if not resp_serializer.is_valid():