logger = logging.getLogger(__name__)
-def create_vnf_pkg(data):
- user_defined_data = ignore_case_get(data, "userDefinedData")
- vnf_pkg_id = str(uuid.uuid4())
- VnfPackageModel.objects.create(
- vnfPackageId=vnf_pkg_id,
- onboardingState=PKG_STATUS.CREATED,
- operationalState=PKG_STATUS.DISABLED,
- usageState=PKG_STATUS.NOT_IN_USE,
- userDefinedData=user_defined_data
- )
- data = {
- "id": vnf_pkg_id,
- "onboardingState": PKG_STATUS.CREATED,
- "operationalState": PKG_STATUS.DISABLED,
- "usageState": PKG_STATUS.NOT_IN_USE,
- "userDefinedData": user_defined_data,
- "_links": None
- }
- return data
-
-
-def query_multiple():
- pkgs_info = []
- nf_pkgs = VnfPackageModel.objects.filter()
- for nf_pkg in nf_pkgs:
- ret = fill_response_data(nf_pkg)
- pkgs_info.append(ret)
- return pkgs_info
-
-
-def query_single(vnf_pkg_id):
- nf_pkg = VnfPackageModel.objects.filter(vnfPackageId=vnf_pkg_id)
- if not nf_pkg.exists():
- raise ResourceNotFoundException('VNF package(%s) does not exist.' % vnf_pkg_id)
- return fill_response_data(nf_pkg[0])
-
-
-def delete_vnf_pkg(vnf_pkg_id):
- vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId=vnf_pkg_id)
- if not vnf_pkg.exists():
- logger.debug('VNF package(%s) is deleted.' % vnf_pkg_id)
- return
- '''
- if vnf_pkg[0].operationalState != PKG_STATUS.DISABLED:
- raise CatalogException("The VNF package (%s) is not disabled" % vnf_pkg_id)
- if vnf_pkg[0].usageState != PKG_STATUS.NOT_IN_USE:
- raise CatalogException("The VNF package (%s) is in use" % vnf_pkg_id)
- '''
- vnf_pkg.delete()
- vnf_pkg_path = os.path.join(CATALOG_ROOT_PATH, vnf_pkg_id)
- fileutil.delete_dirs(vnf_pkg_path)
-
-
-def parse_vnfd_and_save(vnf_pkg_id, vnf_pkg_path):
- vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId=vnf_pkg_id)
- vnf_pkg.update(onboardingState=PKG_STATUS.PROCESSING)
- vnfd_json = toscaparser.parse_vnfd(vnf_pkg_path)
- vnfd = json.JSONDecoder().decode(vnfd_json)
-
- vnfd_id = vnfd["metadata"]["id"]
- if VnfPackageModel.objects.filter(vnfdId=vnfd_id):
- raise CatalogException("VNFD(%s) already exists." % vnfd_id)
-
- vnfd_ver = vnfd["metadata"].get("vnfd_version")
- if not vnfd_ver:
- vnfd_ver = vnfd["metadata"].get("vnfdVersion", "undefined")
- vnf_pkg.update(
- vnfPackageId=vnf_pkg_id,
- vnfdId=vnfd_id,
- vnfVendor=vnfd["metadata"].get("vendor", "undefined"),
- vnfdVersion=vnfd_ver,
- vnfSoftwareVersion=vnfd["metadata"].get("version", "undefined"),
- vnfdModel=vnfd_json,
- onboardingState=PKG_STATUS.ONBOARDED,
- operationalState=PKG_STATUS.ENABLED,
- usageState=PKG_STATUS.NOT_IN_USE,
- localFilePath=vnf_pkg_path
- )
+class VnfPackage(object):
+ def create_vnf_pkg(self, data):
+ user_defined_data = ignore_case_get(data, "userDefinedData")
+ vnf_pkg_id = str(uuid.uuid4())
+ VnfPackageModel.objects.create(
+ vnfPackageId=vnf_pkg_id,
+ onboardingState=PKG_STATUS.CREATED,
+ operationalState=PKG_STATUS.DISABLED,
+ usageState=PKG_STATUS.NOT_IN_USE,
+ userDefinedData=user_defined_data
+ )
+ data = {
+ "id": vnf_pkg_id,
+ "onboardingState": PKG_STATUS.CREATED,
+ "operationalState": PKG_STATUS.DISABLED,
+ "usageState": PKG_STATUS.NOT_IN_USE,
+ "userDefinedData": user_defined_data,
+ "_links": None
+ }
+ return data
+
+ def query_multiple(self):
+ pkgs_info = []
+ nf_pkgs = VnfPackageModel.objects.filter()
+ for nf_pkg in nf_pkgs:
+ ret = fill_response_data(nf_pkg)
+ pkgs_info.append(ret)
+ return pkgs_info
+
+ def query_single(self, vnf_pkg_id):
+ nf_pkg = VnfPackageModel.objects.filter(vnfPackageId=vnf_pkg_id)
+ if not nf_pkg.exists():
+ raise ResourceNotFoundException('VNF package(%s) does not exist.' % vnf_pkg_id)
+ return fill_response_data(nf_pkg[0])
+
+ def delete_vnf_pkg(self, vnf_pkg_id):
+ vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId=vnf_pkg_id)
+ if not vnf_pkg.exists():
+ logger.debug('VNF package(%s) is deleted.' % vnf_pkg_id)
+ return
+ '''
+ if vnf_pkg[0].operationalState != PKG_STATUS.DISABLED:
+ raise CatalogException("The VNF package (%s) is not disabled" % vnf_pkg_id)
+ if vnf_pkg[0].usageState != PKG_STATUS.NOT_IN_USE:
+ raise CatalogException("The VNF package (%s) is in use" % vnf_pkg_id)
+ '''
+ vnf_pkg.delete()
+ vnf_pkg_path = os.path.join(CATALOG_ROOT_PATH, vnf_pkg_id)
+ fileutil.delete_dirs(vnf_pkg_path)
+
+ def fetch_vnf_pkg(self, request, vnf_pkg_id):
+ nf_pkg = VnfPackageModel.objects.filter(vnfPackageId=vnf_pkg_id)
+ if not nf_pkg.exists():
+ raise ResourceNotFoundException('VNF package(%s) does not exist.' % vnf_pkg_id)
+ if nf_pkg[0].onboardingState != PKG_STATUS.ONBOARDED:
+ raise CatalogException("VNF package (%s) is not on-boarded" % vnf_pkg_id)
+ file_path = nf_pkg[0].localFilePath
+ file_name = file_path.split('/')[-1]
+ file_name = file_name.split('\\')[-1]
+ file_range = request.META.get('RANGE')
+ if file_range:
+ start_end = file_range.split('-')
+ start = int(start_end[0])
+ end = int(start_end[1])
+ f = open(file_path, "rb")
+ f.seek(start, 0)
+ fs = f.read(end - start + 1)
+ response = StreamingHttpResponse(fs, status=status.HTTP_200_OK)
+ response['Content-Range'] = file_range
+ else:
+ response = StreamingHttpResponse(open(file_path, 'rb'), status=status.HTTP_200_OK)
+ response['Content-Type'] = 'application/octet-stream'
+ response['Content-Disposition'] = 'attachment; filename=%s' % file_name.encode('utf-8')
+ return response
class VnfPkgUploadThread(threading.Thread):
return pkg_info
-def fetch_vnf_pkg(request, vnf_pkg_id):
- nf_pkg = VnfPackageModel.objects.filter(vnfPackageId=vnf_pkg_id)
- if not nf_pkg.exists():
- raise ResourceNotFoundException('VNF package(%s) does not exist.' % vnf_pkg_id)
- if nf_pkg[0].onboardingState != PKG_STATUS.ONBOARDED:
- raise CatalogException("VNF package (%s) is not on-boarded" % vnf_pkg_id)
- file_path = nf_pkg[0].localFilePath
- file_name = file_path.split('/')[-1]
- file_name = file_name.split('\\')[-1]
- file_range = request.META.get('RANGE')
- if file_range:
- start_end = file_range.split('-')
- start = int(start_end[0])
- end = int(start_end[1])
- f = open(file_path, "rb")
- f.seek(start, 0)
- fs = f.read(end - start + 1)
- response = StreamingHttpResponse(fs, status=status.HTTP_200_OK)
- response['Content-Range'] = file_range
- else:
- response = StreamingHttpResponse(open(file_path, 'rb'), status=status.HTTP_200_OK)
- response['Content-Type'] = 'application/octet-stream'
- response['Content-Disposition'] = 'attachment; filename=%s' % file_name.encode('utf-8')
- return response
+def parse_vnfd_and_save(vnf_pkg_id, vnf_pkg_path):
+ vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId=vnf_pkg_id)
+ vnf_pkg.update(onboardingState=PKG_STATUS.PROCESSING)
+ vnfd_json = toscaparser.parse_vnfd(vnf_pkg_path)
+ vnfd = json.JSONDecoder().decode(vnfd_json)
+
+ vnfd_id = vnfd["metadata"]["id"]
+ if VnfPackageModel.objects.filter(vnfdId=vnfd_id):
+ raise CatalogException("VNFD(%s) already exists." % vnfd_id)
+
+ vnfd_ver = vnfd["metadata"].get("vnfd_version")
+ if not vnfd_ver:
+ vnfd_ver = vnfd["metadata"].get("vnfdVersion", "undefined")
+ vnf_pkg.update(
+ vnfPackageId=vnf_pkg_id,
+ vnfdId=vnfd_id,
+ vnfVendor=vnfd["metadata"].get("vendor", "undefined"),
+ vnfdVersion=vnfd_ver,
+ vnfSoftwareVersion=vnfd["metadata"].get("version", "undefined"),
+ vnfdModel=vnfd_json,
+ onboardingState=PKG_STATUS.ONBOARDED,
+ operationalState=PKG_STATUS.ENABLED,
+ usageState=PKG_STATUS.NOT_IN_USE,
+ localFilePath=vnf_pkg_path
+ )
def handle_upload_failed(vnf_pkg_id):
from catalog.pub.utils import toscaparser
from catalog.packages.const import PKG_STATUS
from catalog.packages.tests.const import vnfd_data
+from catalog.packages.biz.vnf_package import VnfPackage
class MockReq():
os.remove(vnf_pkg[0].localFilePath)
os.removedirs(os.path.join(CATALOG_ROOT_PATH, vnf_pkg[0].vnfPackageId))
+ def test_upload_vnf_pkg_failed(self):
+ data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "empty.txt"), "rb")}
+ VnfPackageModel.objects.create(
+ vnfPackageId="222",
+ )
+ response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
@mock.patch.object(toscaparser, 'parse_vnfd')
@mock.patch.object(urllib2, 'urlopen')
def test_upload_nf_pkg_from_uri(self, mock_urlopen, mock_parse_vnfd):
self.assertEqual(response.data, expect_data)
self.assertEqual(response.status_code, status.HTTP_200_OK)
+ def test_query_single_vnf_failed(self):
+ response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222")
+ self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
def test_query_multiple_vnf(self):
VnfPackageModel.objects.create(
vnfPackageId="111",
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual('BBBB', partial_file_content)
os.remove("vnfPackage.csar")
+
+ def test_fetch_vnf_pkg_when_pkg_not_exist(self):
+ response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222/package_content")
+ self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+ @mock.patch.object(VnfPackage, "create_vnf_pkg")
+ def test_create_vnf_pkg_when_catch_exception(self, mock_create_vnf_pkg):
+ mock_create_vnf_pkg.side_effect = TypeError('integer type')
+ req_data = {
+ "userDefinedData": {"a": "A"}
+ }
+ response = self.client.post("/api/vnfpkgm/v1/vnf_packages", data=req_data, format="json")
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(VnfPackage, "delete_vnf_pkg")
+ def test_delete_single_when_catch_exception(self, mock_delete_vnf_pkg):
+ mock_delete_vnf_pkg.side_effect = TypeError("integer type")
+ response = self.client.delete("/api/vnfpkgm/v1/vnf_packages/222")
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(VnfPackage, "query_single")
+ def test_query_single_when_catch_exception(self, mock_query_single):
+ mock_query_single.side_effect = TypeError("integer type")
+ response = self.client.get("/api/vnfpkgm/v1/vnf_packages/222")
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(VnfPackage, "query_multiple")
+ def test_query_multiple_when_catch_exception(self, mock_query_muitiple):
+ mock_query_muitiple.side_effect = TypeError("integer type")
+ response = self.client.get("/api/vnfpkgm/v1/vnf_packages")
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(toscaparser, 'parse_vnfd')
+ def test_upload_when_catch_exception(self, mock_parse_vnfd):
+ data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "empty.txt"), "rb")}
+ VnfPackageModel.objects.create(
+ vnfPackageId="222",
+ onboardingState="CREATED"
+ )
+ mock_parse_vnfd.side_effect = TypeError("integer type")
+ response = self.client.put("/api/vnfpkgm/v1/vnf_packages/222/package_content", data=data)
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
from catalog.packages.serializers.create_vnf_pkg_info_req import CreateVnfPkgInfoRequestSerializer
from catalog.packages.serializers.vnf_pkg_info import VnfPkgInfoSerializer
from catalog.packages.serializers.vnf_pkg_infos import VnfPkgInfosSerializer
-from catalog.packages.biz.vnf_package import create_vnf_pkg, query_multiple, VnfPkgUploadThread, \
- query_single, delete_vnf_pkg, parse_vnfd_and_save, fetch_vnf_pkg, handle_upload_failed
+from catalog.packages.biz.vnf_package import VnfPackage, VnfPkgUploadThread, parse_vnfd_and_save, handle_upload_failed
from catalog.pub.database.models import VnfPackageModel
from catalog.packages.views.ns_descriptor_views import validate_data
from catalog.packages.const import PKG_STATUS
if request.method == 'GET':
logger.debug("Query VNF packages> %s" % request.data)
try:
- res = query_multiple()
+ res = VnfPackage().query_multiple()
query_serializer = validate_data(res, VnfPkgInfosSerializer)
return Response(data=query_serializer.data, status=status.HTTP_200_OK)
except CatalogException as e:
logger.debug("Create VNF package> %s" % request.data)
try:
req_serializer = validate_data(request.data, CreateVnfPkgInfoRequestSerializer)
- res = create_vnf_pkg(req_serializer.data)
+ res = VnfPackage().create_vnf_pkg(req_serializer.data)
create_vnf_pkg_resp_serializer = validate_data(res, VnfPkgInfoSerializer)
return Response(data=create_vnf_pkg_resp_serializer.data, status=status.HTTP_201_CREATED)
except CatalogException as e:
if request.method == "GET":
try:
- response = fetch_vnf_pkg(request, vnfPkgId)
+ response = VnfPackage().fetch_vnf_pkg(request, vnfPkgId)
return response
except ResourceNotFoundException as e:
logger.error(e.message)
if request.method == 'GET':
logger.debug("Query an individual VNF package> %s" % request.data)
try:
- res = query_single(vnfPkgId)
+ res = VnfPackage().query_single(vnfPkgId)
query_serializer = validate_data(res, VnfPkgInfoSerializer)
return Response(data=query_serializer.data, status=status.HTTP_200_OK)
except ResourceNotFoundException as e:
if request.method == 'DELETE':
logger.debug("Delete an individual VNF package> %s" % request.data)
try:
- delete_vnf_pkg(vnfPkgId)
+ VnfPackage().delete_vnf_pkg(vnfPkgId)
return Response(data=None, status=status.HTTP_204_NO_CONTENT)
except CatalogException as e:
logger.error(e.message)
Code Review / vfc / nfvo / catalog.git / commitdiff