target
htmlcov
.coverage
-static/*/*
test-reports/*
--- /dev/null
+# Copyright (C) 2019 Verizon. All Rights Reserved
+#
+# 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 logging
+
+from catalog.pub.database.models import VnfPackageModel
+from catalog.pub.exceptions import ResourceNotFoundException, ArtifactNotFoundException
+from catalog.pub.utils import fileutil
+
+logger = logging.getLogger(__name__)
+
+
+class FetchVnfPkgArtifact(object):
+ def fetch(self, vnfPkgId, artifactPath):
+ logger.debug("FetchVnfPkgArtifact--get--single--artifact--biz::>"
+ "ID: %s path: %s" % (vnfPkgId, artifactPath))
+ vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId=vnfPkgId).get()
+ if not vnf_pkg:
+ err_msg = "NF Package (%s) doesn't exists." % vnfPkgId
+ return ResourceNotFoundException(err_msg)
+ local_path = vnf_pkg.localFilePath
+ if local_path.endswith(".csar") or local_path.endswith(".zip"):
+ vnf_extract_path = fileutil.unzip_csar_to_tmp(local_path)
+ artifact_path = fileutil.get_artifact_path(vnf_extract_path, artifactPath)
+ if not artifact_path:
+ raise ArtifactNotFoundException("Couldn't artifact %s" % artifactPath)
+ file_content = open(artifact_path, 'rb').read()
+ return file_content
mock_download.side_effect = TypeError("integer type")
response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(toscaparser, 'parse_vnfd')
+ def test_fetch_vnf_artifact(self, mock_parse_vnfd):
+ data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
+ VnfPackageModel.objects.create(
+ vnfPackageId="222",
+ onboardingState="CREATED"
+ )
+ mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
+ response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
+ self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
+ response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/artifacts/image")
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual(response.getvalue(), "ubuntu_16.04\n")
+
+ @mock.patch.object(toscaparser, 'parse_vnfd')
+ def test_fetch_vnf_artifact_not_exists(self, mock_parse_vnfd):
+ data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
+ VnfPackageModel.objects.create(
+ vnfPackageId="222",
+ onboardingState="CREATED"
+ )
+ mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
+ response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
+ self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
+ response = self.client.get("/api/vnfpkgm/v1/vnf_packages/2224/artifacts/image")
+ self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+ @mock.patch.object(toscaparser, 'parse_vnfd')
+ def test_fetch_vnf_artifact_vnf_not_exists(self, mock_parse_vnfd):
+ data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "resource_test.csar"), "rb")}
+ VnfPackageModel.objects.create(
+ vnfPackageId="222",
+ onboardingState="CREATED"
+ )
+ mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
+ response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
+ self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
+ response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/artifacts/image1")
+ self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
from catalog.packages.views import vnf_package_views
from catalog.packages.views.vnf_package_subscription_views import CreateQuerySubscriptionView, QueryTerminateSubscriptionView
+from catalog.packages.views.vnf_package_artifact_views import FetchVnfPkgmArtifactsView
from catalog.packages.views import catalog_views, ns_descriptor_views, pnf_descriptor_views, nsdm_subscription_views
# ETSI SOL 005 VNF Package Management Subscription APIs
url(r'^api/vnfpkgm/v1/subscriptions$', CreateQuerySubscriptionView.as_view(), name='subscriptions_create_query'),
url(r'^api/vnfpkgm/v1/subscriptions/(?P<subscriptionId>[0-9a-zA-Z\-\_]+)$', QueryTerminateSubscriptionView.as_view(), name='subscriptions_query_terminate'),
+ url(r'^api/vnfpkgm/v1/vnf_packages/(?P<vnfPkgId>[0-9a-zA-Z\-\_]+)/artifacts/(?P<artifactPath>[0-9a-zA-Z\-\_]+)$', FetchVnfPkgmArtifactsView.as_view(), name="fetch_vnf_artifacts"),
# url(r'^api/vnfpkgm/v1/subscriptions/(?P<vnfPkgId>[0-9a-zA-Z\-\_]+)$', vnf_package_subscription_views.vnf_package_subscriptions_rc, name='subscriptions_rc'),
# url(r'^api/vnfpkgm/v1/vnf_packages/(?P<vnfPkgId>[0-9a-zA-Z\-\_]+)/vnfd$', vnfd.as_view(), name='vnfd_r'),# url(r'^api/vnfpkgm/v1/vnf_packages/(?P<vnfPkgId>[0-9a-zA-Z\-\_]+)/artifacts/artifactPath$', artifacts.as_view(), name='artifacts_r'),
--- /dev/null
+# Copyright (C) 2019 Verizon. All Rights Reserved
+#
+# 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 traceback
+import logging
+
+from drf_yasg.utils import swagger_auto_schema
+from rest_framework import status
+from rest_framework.views import APIView
+from rest_framework.response import Response
+from django.http import FileResponse
+
+from catalog.packages.serializers.response import ProblemDetailsSerializer
+from catalog.packages.biz.vnf_pkg_artifacts import FetchVnfPkgArtifact
+from catalog.pub.exceptions import ResourceNotFoundException, ArtifactNotFoundException
+
+logger = logging.getLogger(__name__)
+VALID_FILTERS = ["callbackUri", "notificationTypes", "vnfdId", "vnfPkgId", "operationalState", "usageState"]
+
+
+def get_problem_details_serializer(status_code, error_message):
+ problem_details = {
+ "status": status_code,
+ "detail": error_message
+ }
+ problem_details_serializer = ProblemDetailsSerializer(data=problem_details)
+ problem_details_serializer.is_valid()
+ return problem_details_serializer
+
+
+class FetchVnfPkgmArtifactsView(APIView):
+
+ @swagger_auto_schema(
+ responses={
+ status.HTTP_200_OK: None,
+ status.HTTP_404_NOT_FOUND: ProblemDetailsSerializer(),
+ status.HTTP_500_INTERNAL_SERVER_ERROR: ProblemDetailsSerializer()
+ }
+ )
+ def get(self, request, vnfPkgId, artifactPath):
+ logger.debug("FetchVnfPkgmArtifactsView--get::> ")
+ try:
+
+ resp_data = FetchVnfPkgArtifact().fetch(vnfPkgId, artifactPath)
+ response = FileResponse(resp_data)
+
+ return response
+ except ResourceNotFoundException as e:
+ logger.error(e.message)
+ logger.error(traceback.format_exc())
+ problem_details_serializer = get_problem_details_serializer(status.HTTP_404_NOT_FOUND,
+ traceback.format_exc())
+ return Response(data=problem_details_serializer.data, status=status.HTTP_404_NOT_FOUND)
+ except ArtifactNotFoundException as e:
+ logger.error(e.message)
+ logger.error(traceback.format_exc())
+ problem_details_serializer = get_problem_details_serializer(status.HTTP_404_NOT_FOUND,
+ traceback.format_exc())
+ return Response(data=problem_details_serializer.data, status=status.HTTP_404_NOT_FOUND)
+ except Exception as e:
+ logger.error(e.message)
+ logger.error(traceback.format_exc())
+ problem_details_serializer = get_problem_details_serializer(status.HTTP_404_NOT_FOUND,
+ traceback.format_exc())
+ return Response(data=problem_details_serializer.data, status=status.HTTP_404_NOT_FOUND)
class NsdmDuplicateSubscriptionException(CatalogException):
pass
+
+
+class ArtifactNotFoundException(CatalogException):
+ pass
import os
import shutil
import logging
+import tempfile
import traceback
import urllib2
import zipfile
return os.path.join(dst_dir, csar_path)
else:
return ""
+
+
+def unzip_csar_to_tmp(zip_src):
+ dirpath = tempfile.mkdtemp()
+ zip_ref = zipfile.ZipFile(zip_src, 'r')
+ zip_ref.extractall(dirpath)
+ return dirpath
+
+
+def get_artifact_path(vnf_path, artifact_file):
+ for root, dirs, files in os.walk(vnf_path):
+ if artifact_file in files:
+ return os.path.join(root, artifact_file)
+ return None
Code Review / vfc / nfvo / catalog.git / commitdiff