def handle_upload_failed(self, pnf_pkg_id):
pnf_pkg = PnfPackageModel.objects.filter(pnfPackageId=pnf_pkg_id)
pnf_pkg.update(onboardingState=PKG_STATUS.CREATED)
+
+ def parse_pnfd(self, csar_id, inputs):
+ ret = None
+ try:
+ pnf_pkg = PnfPackageModel.objects.filter(pnfPackageId=csar_id)
+ if not pnf_pkg:
+ raise CatalogException("PNF CSAR(%s) does not exist." % csar_id)
+ csar_path = pnf_pkg[0].localFilePath
+ ret = {"model": toscaparser.parse_pnfd(csar_path, inputs)}
+ except CatalogException as e:
+ return [1, e.message]
+ except Exception as e:
+ logger.error(e.message)
+ return [1, e.message]
+ return [0, ret]
mock_download.side_effect = TypeError("integer type")
response = self.client.get("/api/nsd/v1/pnf_descriptors/22/pnfd_content")
self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(toscaparser, 'parse_pnfd')
+ def test_pnfd_parse_normal(self, mock_parse_pnfd):
+ PnfPackageModel(pnfPackageId="8", pnfdId="10").save()
+ mock_parse_pnfd.return_value = json.JSONEncoder().encode({"c": "d"})
+ req_data = {"csarId": "8", "inputs": []}
+ resp = self.client.post("/api/catalog/v1/parserpnfd", req_data, format='json')
+ self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
+ self.assertEqual({"model": '{"c": "d"}'}, resp.data)
url(r'^api/catalog/v1/vnfpackages/(?P<csarId>[0-9a-zA-Z\-\_]+)$', catalog_views.nf_rd_csar, name='nfpackage_rd'),
url(r'^api/catalog/v1/parsernsd$', catalog_views.ns_model_parser, name='nsmodelparser_rc'),
url(r'^api/catalog/v1/parservnfd$', catalog_views.vnf_model_parser, name='vnfmodelparser_rc'),
+ url(r'^api/catalog/v1/parserpnfd$', pnf_descriptor_views.pnf_model_parser, name='pnfmodelparser_rc'),
- # NSD
+ # NSPakcage& NSD
url(r'^api/nsd/v1/ns_descriptors$', ns_descriptor_views.ns_descriptors_rc, name='ns_descriptors_rc'),
url(r'^api/nsd/v1/ns_descriptors/(?P<nsdInfoId>[0-9a-zA-Z\-\_]+)$', ns_descriptor_views.ns_info_rd, name='ns_info_rd'),
url(r'^api/nsd/v1/ns_descriptors/(?P<nsdInfoId>[0-9a-zA-Z\-\_]+)/nsd_content$', ns_descriptor_views.nsd_content_ru, name='nsd_content_ru'),
# url(r'^api/nsd/v1/subscriptions', nsd_subscriptions.as_view(), name='subscriptions_rc'),
# url(r'^api/nsd/v1/subscriptions/(?P<subscriptionId>[0-9a-zA-Z\-\_]+)$', nsd_subscription.as_view(), name='subscription_rd'),
- # PNF
+ # PNF Package and PNFD
url(r'^api/nsd/v1/pnf_descriptors$', pnf_descriptor_views.pnf_descriptors_rc, name='pnf_descriptors_rc'),
url(r'^api/nsd/v1/pnf_descriptors/(?P<pnfdInfoId>[0-9a-zA-Z\-\_]+)$', pnf_descriptor_views.pnfd_info_rd, name='pnfd_info_rd'),
url(r'^api/nsd/v1/pnf_descriptors/(?P<pnfdInfoId>[0-9a-zA-Z\-\_]+)/pnfd_content$', pnf_descriptor_views.pnfd_content_ru, name='pnfd_content_ru'),
- # TODO SOL005 & SOL003
+ # VNFD
url(r'^api/vnfpkgm/v1/vnf_packages$', vnf_package_views.vnf_packages_rc, name='vnf_packages_rc'),
url(r'^api/vnfpkgm/v1/vnf_packages/(?P<vnfPkgId>[0-9a-zA-Z\-\_]+)$', vnf_package_views.vnf_package_rd, name='vnf_package_rd'),
url(r'^api/vnfpkgm/v1/vnf_packages/(?P<vnfPkgId>[0-9a-zA-Z\-\_]+)/package_content$', vnf_package_views.package_content_ru, name='package_content_ru'),
from catalog.packages.serializers.pnfd_infos import PnfdInfosSerializer
from catalog.packages.views.common import validate_data
from catalog.pub.exceptions import CatalogException, ResourceNotFoundException
+from catalog.packages.serializers.catalog_serializers import ParseModelRequestSerializer
+from catalog.packages.serializers.catalog_serializers import ParseModelResponseSerializer
+from catalog.packages.serializers.catalog_serializers import InternalErrorRequestSerializer
+from catalog.pub.utils.syscomm import fun_name
+from catalog.pub.utils.values import ignore_case_get
logger = logging.getLogger(__name__)
error_code = status.HTTP_500_INTERNAL_SERVER_ERROR
error_data = {'error': 'Downloading PNFD content failed.'}
return Response(data=error_data, status=error_code)
+
+
+@swagger_auto_schema(
+ method='POST',
+ operation_description="Parse PNF model",
+ request_body=ParseModelRequestSerializer,
+ responses={
+ status.HTTP_202_ACCEPTED: ParseModelResponseSerializer,
+ status.HTTP_500_INTERNAL_SERVER_ERROR: InternalErrorRequestSerializer})
+@api_view(http_method_names=['POST'])
+def pnf_model_parser(request, *args, **kwargs):
+ csar_id = ignore_case_get(request.data, "csarId")
+ inputs = ignore_case_get(request.data, "inputs")
+ logger.debug(
+ "Enter %s, csar_id=%s, inputs=%s",
+ fun_name(),
+ csar_id,
+ inputs)
+ ret = PnfDescriptor().parse_pnfd(csar_id, inputs)
+ logger.info("Leave %s, Return value is %s", fun_name(), ret)
+ if ret[0] != 0:
+ return Response(data={'error': ret[1]}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+ response = validate_data(ret[1], ParseModelResponseSerializer)
+ return Response(data=response.data, status=status.HTTP_202_ACCEPTED)
from catalog.pub.utils.toscaparser.nsdmodel import EtsiNsdInfoModel
from catalog.pub.utils.toscaparser.vnfdmodel import EtsiVnfdInfoModel
+from catalog.pub.utils.toscaparser.pnfmodel import PnfdInfoModel
def parse_nsd(path, input_parameters=[]):
def parse_pnfd(path, input_parameters=[]):
- pass
+ tosca_obj = PnfdInfoModel(path, input_parameters)
+ strResponse = json.dumps(tosca_obj, default=lambda obj: obj.__dict__)
+ strResponse = strResponse.replace(': null', ': ""')
+ return strResponse
class BaseInfoModel(object):
+ def __init__(self, path, params):
+ tosca = self.buildToscaTemplate(path, params)
+ self.parseModel(tosca)
+
+ def parseModel(self, tosca):
+ pass
+
+ def buildInputs(self, top_inputs):
+ ret = {}
+ for tmpinput in top_inputs:
+ tmp = {}
+ tmp['type'] = tmpinput.type
+ tmp['description'] = tmpinput.description
+ tmp['default'] = tmpinput.default
+
+ ret[tmpinput.name] = tmp
+ return ret
+
def buildToscaTemplate(self, path, params):
file_name = None
try:
if tosca.tpl['metadata'].get('UUID', ''):
self.metadata['id'] = tosca.tpl['metadata']['UUID']
+ def buildNode(self, nodeTemplate, tosca):
+ inputs = tosca.inputs
+ parsed_params = tosca.parsed_params
+ ret = {}
+ ret['name'] = nodeTemplate.name
+ ret['nodeType'] = nodeTemplate.type
+ if 'description' in nodeTemplate.entity_tpl:
+ ret['description'] = nodeTemplate.entity_tpl['description']
+ else:
+ ret['description'] = ''
+ if 'metadata' in nodeTemplate.entity_tpl:
+ ret['metadata'] = nodeTemplate.entity_tpl['metadata']
+ else:
+ ret['metadata'] = ''
+ props = self.buildProperties_ex(nodeTemplate, tosca.topology_template)
+ ret['properties'] = self.verify_properties(props, inputs, parsed_params)
+ ret['requirements'] = self.build_requirements(nodeTemplate)
+ self.buildCapabilities(nodeTemplate, inputs, ret)
+ self.buildArtifacts(nodeTemplate, inputs, ret)
+ interfaces = self.build_interfaces(nodeTemplate)
+ if interfaces:
+ ret['interfaces'] = interfaces
+ return ret
+
def buildProperties(self, nodeTemplate, parsed_params):
properties = {}
isMappingParams = parsed_params and len(parsed_params) > 0
--- /dev/null
+# Copyright 2018 ZTE Corporation.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import functools
+import logging
+import os
+from catalog.pub.utils.toscaparser.basemodel import BaseInfoModel
+logger = logging.getLogger(__name__)
+
+
+class PnfdInfoModel(BaseInfoModel):
+
+ def __init__(self, path, params):
+ super(PnfdInfoModel, self).__init__(path, params)
+
+ def parseModel(self, tosca):
+ self.buidMetadata(tosca)
+ if hasattr(tosca, 'topology_template') and hasattr(tosca.topology_template, 'inputs'):
+ self.inputs = self.buildInputs(tosca.topology_template.inputs)
+
+ nodeTemplates = map(functools.partial(self.buildNode, tosca=tosca),
+ tosca.nodetemplates)
+ print nodeTemplates
+ self.basepath = self._get_base_path(tosca)
+ self.pnf = {}
+ self.get_all_cp(nodeTemplates)
+
+ def _get_base_path(self, tosca):
+ fpath, fname = os.path.split(tosca.path)
+ return fpath
+
+ def get_substitution_mappings(self, tosca):
+ pnf_substitution_mappings = tosca.tpl['topology_template']['substitution_mappings']
+ if pnf_substitution_mappings:
+ self.pnf['type'] = pnf_substitution_mappings['node_type']
+ self.pnf['properties'] = pnf_substitution_mappings['properties']
+
+ def get_all_cp(self, nodeTemplates):
+ self.pnf['ExtPorts'] = []
+ for node in nodeTemplates:
+ if self.isPnfExtPort(node):
+ cp = {}
+ cp['id'] = node['name']
+ cp['type'] = node['nodeType']
+ cp['properties'] = node['properties']
+ self.pnf['ExtPorts'].append(cp)
+
+ def isPnfExtPort(self, node):
+ return node['nodeType'].find('tosca.nodes.nfv.PnfExtPort') >= 0
from django.test import TestCase
-from catalog.pub.utils.toscaparser import parse_vnfd
+from catalog.pub.utils.toscaparser import parse_vnfd, parse_pnfd
logger = logging.getLogger(__name__)
metadata = json.loads(vnfd_json).get("metadata")
logger.debug("metadata:%s", metadata)
self.assertEqual(("vCPE_%s" % vcpe_part), metadata.get("template_name", ""))
+
+ def test_pnfd_parse(self):
+ csar_path = os.path.dirname(os.path.abspath(__file__)) + "/testdata/pnf/ran-du.csar"
+ pnfd_json = parse_pnfd(csar_path)
+ print pnfd_json
+ metadata = json.loads(pnfd_json).get("metadata")
+ self.assertEqual("RAN_DU", metadata.get("template_name", ""))
Code Review / vfc / nfvo / catalog.git / commitdiff