# 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 json
-from rest_framework import status
-from test_data import nsd_model
+# import json
+# from rest_framework import status
+# from test_data import nsd_model
from django.test import Client
from django.test import TestCase
from lcm.pub.database.models import FPInstModel, VNFFGInstModel
VNFFGInstModel.objects.all().delete()
FPInstModel.objects.all().delete()
- def test_sfc_instance_success(self):
- data = {
- "nsInstanceId": "ns_inst_1",
- "context": json.dumps(nsd_model),
- "fpindex": "fpd_1",
- "sdnControllerId": "sdnControllerId_1"
- }
- resp = self.client.post("/api/nslcm/v1/ns/sfc_instance", data, format='json')
+ # def test_sfc_instance_success(self):
+ # data = {
+ # "nsInstanceId": "ns_inst_1",
+ # "context": json.dumps(nsd_model),
+ # "fpindex": "fpd_1",
+ # "sdnControllerId": "sdnControllerId_1"
+ # }
+ # resp = self.client.post("/api/nslcm/v1/ns/sfc_instance", data, format='json')
- self.assertEqual(resp.status_code, status.HTTP_200_OK, resp.data)
- vnffg = VNFFGInstModel.objects.get(vnffginstid="vnffg_inst_1")
- ret = FPInstModel.objects.get(fpinstid=resp.data["fpinstid"])
- self.assertEqual(vnffg.fplist, resp.data["fpinstid"])
- self.assertIsNotNone(ret)
+ # self.assertEqual(resp.status_code, status.HTTP_200_OK, resp.data)
+ # vnffg = VNFFGInstModel.objects.get(vnffginstid="vnffg_inst_1")
+ # ret = FPInstModel.objects.get(fpinstid=resp.data["fpinstid"])
+ # self.assertEqual(vnffg.fplist, resp.data["fpinstid"])
+ # self.assertIsNotNone(ret)
import uuid
import os
import mock
+import random
from django.test import Client
from django.test import TestCase
from rest_framework import status
from lcm.pub.utils.jobutil import JobUtil, JOB_TYPE, JOB_MODEL_STATUS
from lcm.pub.msapi import catalog
from lcm.pub.utils.scaleaspect import get_json_data
+from lcm.pub.utils import idutil
SCALING_JSON = {
"NS", JOB_TYPE.MANUAL_SCALE_VNF, self.ns_inst_id)
self.package_id = "7"
self.client = Client()
+ idutil.get_auto_id = mock.Mock()
+ idutil.get_auto_id.return_value = random.random()
NSInstModel(
id=self.ns_inst_id,
name="abc",
# See the License for the specific language governing permissions and
# limitations under the License.
-import json
import logging
import traceback
import sys
-from lcm.pub.database.models import NSDModel, NSInstModel, NfPackageModel
+from lcm.pub.database.models import NSDModel, NSInstModel
from lcm.pub.utils.values import ignore_case_get
-from lcm.pub.msapi.catalog import STATUS_ONBOARDED, P_STATUS_ENABLED
from lcm.pub.msapi.catalog import query_csar_from_catalog, set_csar_state
-from lcm.pub.msapi.catalog import query_rawdata_from_catalog, delete_csar_from_catalog
+from lcm.pub.msapi.catalog import delete_csar_from_catalog
from lcm.pub.exceptions import NSLCMException
-from lcm.pub.utils import toscautil
logger = logging.getLogger(__name__)
return fmt_ns_pkg_rsp(STATUS_SUCCESS, ret[1], "")
-def ns_on_boarding(csar_id):
- return ns_common_call(NsPackage().on_boarding, csar_id)
-
-
-def ns_delete_csar(csar_id):
- return ns_common_call(NsPackage().delete_csar, csar_id)
-
-
def ns_delete_pending_csar(csar_id):
return ns_common_call(NsPackage().delete_pending_csar, csar_id)
return ns_common_call(NsPackage().set_state_csar, csar_id, operation)
-def ns_get_csar(csar_id):
- ret = None
- try:
- ret = NsPackage().get_csar(csar_id)
- except NSLCMException as e:
- return [1, e.message]
- except:
- logger.error(traceback.format_exc())
- return [1, str(sys.exc_info())]
- return ret
-
-
###############################################################################################################
def __init__(self):
pass
- def on_boarding(self, csar_id):
- csar = query_csar_from_catalog(csar_id)
- if ignore_case_get(csar, "onBoardState") == STATUS_ONBOARDED:
- raise NSLCMException("CSAR(%s) already onBoarded." % csar_id)
-
- raw_data = query_rawdata_from_catalog(csar_id)
- nsd = toscautil.convert_nsd_model(raw_data["rawData"]) # convert to inner json
- nsd = json.JSONDecoder().decode(nsd)
- nsd_id = nsd["metadata"]["id"]
- if NSDModel.objects.filter(nsd_id=nsd_id):
- raise NSLCMException("NSD(%s) already exists." % nsd_id)
-
- for vnf in nsd["vnfs"]:
- vnfd_id = vnf["properties"]["id"]
- pkg = NfPackageModel.objects.filter(vnfdid=vnfd_id)
- if not pkg:
- raise NSLCMException("VNF package(%s) is not onBoarded." % vnfd_id)
- pkg_id = pkg[0].nfpackageid
- if query_csar_from_catalog(pkg_id, "onBoardState") != STATUS_ONBOARDED:
- raise NSLCMException("VNF package(%s) is not onBoarded on catalog." % pkg_id)
-
- NSDModel(
- id=csar_id,
- nsd_id=nsd_id,
- name=nsd["metadata"].get("name", nsd_id),
- vendor=nsd["metadata"].get("vendor", "undefined"),
- description=nsd["metadata"].get("description", ""),
- version=nsd["metadata"].get("version", "undefined"),
- nsd_model=json.JSONEncoder().encode(nsd)).save()
-
- set_csar_state(csar_id, "onBoardState", STATUS_ONBOARDED)
- set_csar_state(csar_id, "operationalState", P_STATUS_ENABLED)
-
- return [0, "CSAR(%s) onBoarded successfully." % csar_id]
-
def delete_csar(self, csar_id):
if not NSDModel.objects.filter(id=csar_id):
return delete_csar_from_catalog(csar_id)
NSDModel.objects.filter(id=csar_id).delete()
return ret
- def get_csar(self, csar_id):
- package_info = {}
-
- csar_in_catalog = query_csar_from_catalog(csar_id)
- props_of_catalog = [
- "name", "provider", "version", "operationalState", "usageState",
- "onBoardState", "processState", "deletionPending", "downloadUri",
- "createTime", "modifyTime", "format", "size"]
- for prop in props_of_catalog:
- package_info[prop] = ignore_case_get(csar_in_catalog, prop)
-
- csars = NSDModel.objects.filter(id=csar_id)
- if csars:
- package_info["nsdId"] = csars[0].nsd_id
- package_info["nsdProvider"] = csars[0].vendor
- package_info["nsdVersion"] = csars[0].version
-
- nss = NSInstModel.objects.filter(nspackage_id=csar_id)
- ns_instance_info = [{"nsInstanceId": ns.id, "nsInstanceName": ns.name} for ns in nss]
-
- return [0, {"csarId": csar_id, "packageInfo": package_info, "nsInstanceInfo": ns_instance_info}]
-
def set_state_csar(self, csar_id, operation):
if not NSDModel.objects.filter(id=csar_id):
raise NSLCMException("CSAR(%s) does not exist." % csar_id)
+++ /dev/null
-# Copyright 2017 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 json
-import logging
-import os
-import sys
-import traceback
-
-from lcm.pub.config.config import CATALOG_ROOT_PATH
-from lcm.pub.database.models import NSDModel, NSInstModel, NfPackageModel
-from lcm.pub.exceptions import NSLCMException
-from lcm.pub.msapi import sdc
-from lcm.pub.utils import fileutil
-from lcm.pub.utils import toscaparser
-
-logger = logging.getLogger(__name__)
-
-STATUS_SUCCESS, STATUS_FAILED = "success", "failed"
-
-
-def fmt_ns_pkg_rsp(status, desc, error_code="500"):
- return [0, {"status": status, "statusDescription": desc, "errorCode": error_code}]
-
-
-def ns_on_distribute(csar_id):
- ret = None
- try:
- ret = SdcNsPackage().on_distribute(csar_id)
- except NSLCMException as e:
- SdcNsPackage().delete_catalog(csar_id)
- return fmt_ns_pkg_rsp(STATUS_FAILED, e.message)
- except:
- logger.error(traceback.format_exc())
- SdcNsPackage().delete_catalog(csar_id)
- return fmt_ns_pkg_rsp(STATUS_FAILED, str(sys.exc_info()))
- return fmt_ns_pkg_rsp(STATUS_SUCCESS, ret[1], "")
-
-
-def ns_delete_csar(csar_id, force_delete):
- ret = None
- try:
- ret = SdcNsPackage().delete_csar(csar_id, force_delete)
- except NSLCMException as e:
- return fmt_ns_pkg_rsp(STATUS_FAILED, e.message)
- except:
- logger.error(traceback.format_exc())
- return fmt_ns_pkg_rsp(STATUS_FAILED, str(sys.exc_info()))
- return fmt_ns_pkg_rsp(STATUS_SUCCESS, ret[1], "")
-
-
-def ns_get_csars():
- ret = None
- try:
- ret = SdcNsPackage().get_csars()
- except NSLCMException as e:
- return [1, e.message]
- except:
- logger.error(traceback.format_exc())
- return [1, str(sys.exc_info())]
- return ret
-
-
-def ns_get_csar(csar_id):
- ret = None
- try:
- ret = SdcNsPackage().get_csar(csar_id)
- except NSLCMException as e:
- return [1, e.message]
- except:
- logger.error(traceback.format_exc())
- return [1, str(sys.exc_info())]
- return ret
-
-
-###############################################################################################################
-
-class SdcNsPackage(object):
- """
- Actions for sdc ns package.
- """
-
- def __init__(self):
- pass
-
- def on_distribute(self, csar_id):
- if NSDModel.objects.filter(id=csar_id):
- raise NSLCMException("NS CSAR(%s) already exists." % csar_id)
-
- artifact = sdc.get_artifact(sdc.ASSETTYPE_SERVICES, csar_id)
- local_path = os.path.join(CATALOG_ROOT_PATH, csar_id)
- local_file_name = sdc.download_artifacts(artifact["toscaModelURL"],
- local_path,
- "%s.csar" % artifact.get("name", csar_id))
-
- nsd_json = toscaparser.parse_nsd(local_file_name)
- nsd = json.JSONDecoder().decode(nsd_json)
-
- nsd_id = nsd["metadata"]["id"]
- if NSDModel.objects.filter(nsd_id=nsd_id):
- raise NSLCMException("NSD(%s) already exists." % nsd_id)
-
- for vnf in nsd["vnfs"]:
- vnfd_id = vnf["properties"]["id"]
- pkg = NfPackageModel.objects.filter(vnfdid=vnfd_id)
- if not pkg:
- raise NSLCMException("VNF package(%s) is not distributed." % vnfd_id)
-
- NSDModel(
- id=csar_id,
- nsd_id=nsd_id,
- name=nsd["metadata"].get("name", nsd_id),
- vendor=nsd["metadata"].get("vendor", "undefined"),
- description=nsd["metadata"].get("description", ""),
- version=nsd["metadata"].get("version", "undefined"),
- nsd_path=local_file_name,
- nsd_model=nsd_json).save()
-
- return [0, "CSAR(%s) distributed successfully." % csar_id]
-
- def delete_csar(self, csar_id, force_delete):
- if force_delete:
- NSInstModel.objects.filter(nspackage_id=csar_id).delete()
- else:
- if NSInstModel.objects.filter(nspackage_id=csar_id):
- raise NSLCMException("CSAR(%s) is in using, cannot be deleted." % csar_id)
- NSDModel.objects.filter(id=csar_id).delete()
- return [0, "Delete CSAR(%s) successfully." % csar_id]
-
- def get_csars(self):
- csars = {"csars": []}
- nss = NSDModel.objects.filter()
- for ns in nss:
- csars["csars"].append({
- "csarId": ns.id,
- "nsdId": ns.nsd_id
- })
- return [0, csars]
-
- def get_csar(self, csar_id):
- package_info = {}
- csars = NSDModel.objects.filter(id=csar_id)
- if csars:
- package_info["nsdId"] = csars[0].nsd_id
- package_info["nsdProvider"] = csars[0].vendor
- package_info["nsdVersion"] = csars[0].version
-
- nss = NSInstModel.objects.filter(nspackage_id=csar_id)
- ns_instance_info = [{
- "nsInstanceId": ns.id,
- "nsInstanceName": ns.name} for ns in nss]
-
- return [0, {"csarId": csar_id, "packageInfo": package_info, "nsInstanceInfo": ns_instance_info}]
-
- def delete_catalog(self, csar_id):
- local_path = os.path.join(CATALOG_ROOT_PATH, csar_id)
- fileutil.delete_dirs(local_path)
import json
import mock
+import random
from django.test import Client
from django.test import TestCase
from rest_framework import status
from lcm.pub.nfvi.vim.vimadaptor import VimAdaptor
from lcm.pub.utils import fileutil
from lcm.pub.utils import restcall
+from lcm.pub.utils import idutil
class TestNfPackage(TestCase):
NfInstModel.objects.filter().delete()
JobModel.objects.filter().delete()
JobStatusModel.objects.filter().delete()
+ idutil.get_auto_id = mock.Mock()
+ idutil.get_auto_id.return_value = random.random()
self.vnfd_raw_data = {
"rawData": {
"instance": {
def set_nsd_vnf_id(self, val):
self.nsd_raw_data["rawData"]["instance"]["nodes"][0]["properties"]["id"]["value"] = val
- @mock.patch.object(restcall, 'call_req')
- def test_ns_pkg_on_boarding_when_on_boarded(self, mock_call_req):
- mock_call_req.return_value = [0, json.JSONEncoder().encode({"onBoardState": "onBoarded"}), '200']
- resp = self.client.post("/api/nslcm/v0/nspackage", {"csarId": "1"}, format='json')
- self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
- self.assertEqual("failed", resp.data["status"])
- self.assertEqual("CSAR(1) already onBoarded.", resp.data["statusDescription"])
-
- @mock.patch.object(restcall, 'call_req')
- def test_ns_pkg_on_boarding_when_nsd_already_exists(self, mock_call_req):
- self.set_nsd_metadata(key="id", val="2")
- mock_vals = {
- "/api/catalog/v1/csars/2":
- [0, json.JSONEncoder().encode({"onBoardState": "non-onBoarded"}), '200'],
- "/api/catalog/v1/servicetemplates/queryingrawdata":
- [0, json.JSONEncoder().encode(self.nsd_raw_data), '200']}
-
- def side_effect(*args):
- return mock_vals[args[4]]
- mock_call_req.side_effect = side_effect
-
- NSDModel(id="1", nsd_id="2").save()
- resp = self.client.post("/api/nslcm/v0/nspackage", {"csarId": "2"}, format='json')
- self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
- self.assertEqual("failed", resp.data["status"])
- self.assertEqual("NSD(2) already exists.", resp.data["statusDescription"])
-
- @mock.patch.object(restcall, 'call_req')
- def test_ns_pkg_on_boarding_when_vnf_pkg_not_on_boarded(self, mock_call_req):
- self.set_nsd_metadata(key="id", val="2")
- self.set_nsd_vnf_id(val="3")
- mock_vals = {
- "/api/catalog/v1/csars/3":
- [0, json.JSONEncoder().encode({"onBoardState": "non-onBoarded"}), '200'],
- "/api/catalog/v1/servicetemplates/queryingrawdata":
- [0, json.JSONEncoder().encode(self.nsd_raw_data), '200']}
-
- def side_effect(*args):
- return mock_vals[args[4]]
- mock_call_req.side_effect = side_effect
-
- resp = self.client.post("/api/nslcm/v0/nspackage", {"csarId": "3"}, format='json')
- self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
- self.assertEqual("failed", resp.data["status"])
- self.assertEqual("VNF package(3) is not onBoarded.", resp.data["statusDescription"])
-
- @mock.patch.object(restcall, 'call_req')
- def test_ns_pkg_on_boarding_when_vnf_pkg_not_on_boarded_on_catalog(self, mock_call_req):
- self.set_nsd_metadata(key="id", val="2")
- self.set_nsd_vnf_id(val="6")
- mock_vals = {
- "/api/catalog/v1/csars/4":
- [0, json.JSONEncoder().encode({"onBoardState": "non-onBoarded"}), '200'],
- "/api/catalog/v1/servicetemplates/queryingrawdata":
- [0, json.JSONEncoder().encode(self.nsd_raw_data), '200'],
- "/api/catalog/v1/csars/5":
- [0, json.JSONEncoder().encode({"onBoardState": "non-onBoarded"}), '200'], }
-
- def side_effect(*args):
- return mock_vals[args[4]]
- mock_call_req.side_effect = side_effect
-
- NfPackageModel(uuid="1", nfpackageid="5", vnfdid="6").save()
- resp = self.client.post("/api/nslcm/v0/nspackage", {"csarId": "4"}, format='json')
- self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
- self.assertEqual("failed", resp.data["status"])
- self.assertEqual("VNF package(5) is not onBoarded on catalog.", resp.data["statusDescription"])
-
- @mock.patch.object(restcall, 'call_req')
- def test_ns_pkg_on_boarding_when_on_board_success(self, mock_call_req):
- self.set_nsd_metadata(key="id", val="2")
- self.set_nsd_metadata(key="name", val="3")
- self.set_nsd_metadata(key="vendor", val="4")
- self.set_nsd_metadata(key="description", val="5")
- self.set_nsd_metadata(key="version", val="6")
- self.set_nsd_vnf_id(val="7")
- mock_vals = {
- "/api/catalog/v1/csars/5":
- [0, json.JSONEncoder().encode({
- "onBoardState": "non-onBoarded",
- "createTime": "2016-05-15 12:30:34",
- "modifyTime": "2016-05-15 12:30:34"}), '200'],
- "/api/catalog/v1/servicetemplates/queryingrawdata":
- [0, json.JSONEncoder().encode(self.nsd_raw_data), '200'],
- "/api/catalog/v1/csars/6":
- [0, json.JSONEncoder().encode({"onBoardState": "onBoarded"}), '200'],
- "/api/catalog/v1/csars/5?operationalState=Enabled": [0, '{}', 200],
- "/api/catalog/v1/csars/5?onBoardState=onBoarded": [0, "OK", '200']}
-
- def side_effect(*args):
- return mock_vals[args[4]]
-
- mock_call_req.side_effect = side_effect
-
- NfPackageModel(uuid="1", nfpackageid="6", vnfdid="7").save()
- resp = self.client.post("/api/nslcm/v0/nspackage", {"csarId": "5"}, format='json')
- self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
- self.assertEqual("success", resp.data["status"])
- self.assertEqual("CSAR(5) onBoarded successfully.", resp.data["statusDescription"])
- nsds = NSDModel.objects.filter(id="5")
- self.assertEqual(1, len(nsds))
- self.assertEqual("2", nsds[0].nsd_id)
-
- ###############################################################################################################
- @mock.patch.object(restcall, 'call_req')
- def test_delete_csar_when_id_not_exist(self, mock_call_req):
- mock_call_req.return_value = [0, "", '204']
- resp = self.client.delete("/api/nslcm/v0/nspackage/6")
- self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
- self.assertEqual("success", resp.data["status"])
- self.assertEqual("Delete CSAR(6) successfully.", resp.data["statusDescription"])
-
- @mock.patch.object(restcall, 'call_req')
- def test_delete_csar_when_ref_by_ns_inst(self, mock_call_req):
- mock_call_req.return_value = [0, "OK", '200']
-
- NSDModel(id="7", nsd_id="2").save()
- NSInstModel(id="1", nspackage_id="7").save()
-
- resp = self.client.delete("/api/nslcm/v0/nspackage/7")
- self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
- self.assertEqual("success", resp.data["status"])
- self.assertEqual("Set deletionPending to True of CSAR(7) successfully.",
- resp.data["statusDescription"])
-
- @mock.patch.object(restcall, 'call_req')
- def test_delete_csar_when_delete_success(self, mock_call_req):
- mock_call_req.return_value = [0, "OK", '204']
-
- NSDModel(id="8", nsd_id="2").save()
-
- resp = self.client.delete("/api/nslcm/v0/nspackage/8")
- self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
- self.assertEqual("success", resp.data["status"])
- self.assertEqual("Delete CSAR(8) successfully.", resp.data["statusDescription"])
-
###############################################################################################################
def test_delete_pending_csar_when_id_not_exist(self):
resp = self.client.delete("/api/nslcm/v0/nspackage/9/deletionpending")
self.assertEqual("success", resp.data["status"])
self.assertEqual("Delete CSAR(12) successfully.", resp.data["statusDescription"])
- ###############################################################################################################
- @mock.patch.object(restcall, 'call_req')
- def test_get_csar_successfully(self, mock_call_req):
- mock_call_req.return_value = [0, json.JSONEncoder().encode({
- "name": "1",
- "provider": "2",
- "version": "3",
- "operationalState": "4",
- "usageState": "5",
- "onBoardState": "6",
- "processState": "7",
- "deletionPending": "8",
- "downloadUri": "9",
- "createTime": "10",
- "modifyTime": "11",
- "format": "12",
- "size": "13"
- }), '200']
-
- NSDModel(id="13", nsd_id="2", vendor="3", version="4").save()
- NSInstModel(id="1", nspackage_id="13", name="11").save()
- NSInstModel(id="2", nspackage_id="13", name="22").save()
-
- resp = self.client.get("/api/nslcm/v0/nspackage/13")
- self.assertEqual(resp.status_code, status.HTTP_200_OK)
- expect_data = {"nsInstanceInfo": [{"nsInstanceId": "1", "nsInstanceName": "11"},
- {"nsInstanceId": "2", "nsInstanceName": "22"}], "csarId": "13",
- "packageInfo": {"nsdProvider": "3", "usageState": "5",
- "onBoardState": "6", "name": "1", "format": "12",
- "modifyTime": "11", "nsdId": "2", "nsdVersion": "4",
- "deletionPending": "8", "version": "3", "downloadUri": "9",
- "processState": "7", "provider": "2", "operationalState": "4",
- "createTime": "10", "size": "13"}}
- self.assertEqual(expect_data, resp.data)
-
###############################################################################################################
def test_disable_csar_when_id_not_exist_table(self):
resp = self.client.put("/api/nslcm/v0/nspackage/14/disabled")
import json
import mock
+import random
from django.test import Client
from django.test import TestCase
from rest_framework import status
from lcm.pub.database.models import JobStatusModel, JobModel
from lcm.pub.database.models import NfPackageModel, NfInstModel
from lcm.pub.msapi import sdc
-from lcm.pub.utils import restcall, toscaparser
+from lcm.pub.utils import restcall, toscaparser, idutil
class TestNfPackage(TestCase):
NfPackageModel.objects.filter().delete()
NfInstModel.objects.filter().delete()
JobModel.objects.filter().delete()
+ idutil.get_auto_id = mock.Mock()
+ idutil.get_auto_id.return_value = random.random()
JobStatusModel.objects.filter().delete()
self.vnfd_data = {
"volume_storages": [
+++ /dev/null
-# Copyright 2017 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 json
-import mock
-from rest_framework import status
-from django.test import TestCase
-from django.test import Client
-
-from lcm.pub.utils import restcall, toscaparser
-from lcm.pub.database.models import NSDModel, NSInstModel, NfPackageModel
-from lcm.pub.msapi import sdc
-
-
-class TestSdcNsPackage(TestCase):
- def setUp(self):
- self.client = Client()
- NSDModel.objects.filter().delete()
- NSInstModel.objects.filter().delete()
- NfPackageModel.objects.filter().delete()
- self.nsd_data = {
- "vnffgs": [
- {
- "vnffg_id": "vnffg1",
- "description": "",
- "members": [
- "path1",
- "path2"
- ],
- "properties": {
- "vendor": "zte",
- "connection_point": [
- "m6000_data_in",
- "m600_tunnel_cp",
- "m6000_data_out"
- ],
- "version": "1.0",
- "constituent_vnfs": [
- "VFW",
- "VNAT"
- ],
- "number_of_endpoints": 3,
- "dependent_virtual_link": [
- "sfc_data_network",
- "ext_datanet_net",
- "ext_mnet_net"
- ]
- }
- }
- ],
- "inputs": {
- "sfc_data_network": {
- "type": "string",
- "value": "sfc_data_network"
- },
- "externalDataNetworkName": {
- "type": "string",
- "value": "vlan_4004_tunnel_net"
- },
- "externalManageNetworkName": {
- "type": "string",
- "value": "vlan_4008_mng_net"
- },
- "NatIpRange": {
- "type": "string",
- "value": "192.167.0.10-192.168.0.20"
- },
- "externalPluginManageNetworkName": {
- "type": "string",
- "value": "vlan_4007_plugin_net"
- }
- },
- "pnfs": [
- {
- "pnf_id": "m6000_s",
- "cps": [],
- "description": "",
- "properties": {
- "vendor": "zte",
- "request_reclassification": False,
- "pnf_type": "m6000s",
- "version": "1.0",
- "management_address": "111111",
- "id": "m6000_s",
- "nsh_aware": False
- }
- }
- ],
- "fps": [
- {
- "properties": {
- "symmetric": False,
- "policy": {
- "type": "ACL",
- "criteria": {
- "dest_port_range": "1-100",
- "ip_protocol": "tcp",
- "source_ip_range": [
- "119.1.1.1-119.1.1.10"
- ],
- "dest_ip_range": [
- {
- "get_input": "NatIpRange"
- }
- ],
- "dscp": 0,
- "source_port_range": "1-100"
- }
- }
- },
- "forwarder_list": [
- {
- "capability": "",
- "type": "cp",
- "node_name": "m6000_data_out"
- },
- {
- "capability": "",
- "type": "cp",
- "node_name": "m600_tunnel_cp"
- },
- {
- "capability": "vnat_fw_inout",
- "type": "vnf",
- "node_name": "VNAT"
- }
- ],
- "description": "",
- "fp_id": "path2"
- },
- {
- "properties": {
- "symmetric": True,
- "policy": {
- "type": "ACL",
- "criteria": {
- "dest_port_range": "1-100",
- "ip_protocol": "tcp",
- "source_ip_range": [
- "1-100"
- ],
- "dest_ip_range": [
- "1-100"
- ],
- "dscp": 4,
- "source_port_range": "1-100"
- }
- }
- },
- "forwarder_list": [
- {
- "capability": "",
- "type": "cp",
- "node_name": "m6000_data_in"
- },
- {
- "capability": "",
- "type": "cp",
- "node_name": "m600_tunnel_cp"
- },
- {
- "capability": "vfw_fw_inout",
- "type": "vnf",
- "node_name": "VFW"
- },
- {
- "capability": "vnat_fw_inout",
- "type": "vnf",
- "node_name": "VNAT"
- },
- {
- "capability": "",
- "type": "cp",
- "node_name": "m600_tunnel_cp"
- },
- {
- "capability": "",
- "type": "cp",
- "node_name": "m6000_data_out"
- }
- ],
- "description": "",
- "fp_id": "path1"
- }
- ],
- "routers": [],
- "vnfs": [
- {
- "vnf_id": "VFW",
- "description": "",
- "properties": {
- "plugin_info": "vbrasplugin_1.0",
- "vendor": "zte",
- "is_shared": False,
- "adjust_vnf_capacity": True,
- "name": "VFW",
- "vnf_extend_type": "driver",
- "csarVersion": "v1.0",
- "csarType": "NFAR",
- "csarProvider": "ZTE",
- "version": "1.0",
- "nsh_aware": True,
- "cross_dc": False,
- "vnf_type": "VFW",
- "vmnumber_overquota_alarm": True,
- "vnfd_version": "1.0.0",
- "externalPluginManageNetworkName": "vlan_4007_plugin_net",
- "id": "vcpe_vfw_zte_1_0",
- "request_reclassification": False
- },
- "dependencies": [
- {
- "key_name": "vfw_ctrl_by_manager_cp",
- "vl_id": "ext_mnet_net"
- },
- {
- "key_name": "vfw_data_cp",
- "vl_id": "sfc_data_network"
- }
- ],
- "type": "tosca.nodes.nfv.ext.zte.VNF.VFW",
- "networks": []
- }
- ],
- "ns_exposed": {
- "external_cps": [],
- "forward_cps": []
- },
- "policies": [
- {
- "file_url": "policies/abc.drl",
- "name": "aaa"
- }
- ],
- "vls": [
- {
- "route_id": "",
- "vl_id": "ext_mnet_net",
- "route_external": False,
- "description": "",
- "properties": {
- "name": "vlan_4008_mng_net",
- "mtu": 1500,
- "location_info": {
- "tenant": "admin",
- "vimid": 2,
- "availability_zone": "nova"
- },
- "ip_version": 4,
- "dhcp_enabled": True,
- "network_name": "vlan_4008_mng_net",
- "network_type": "vlan"
- }
- },
- {
- "route_id": "",
- "vl_id": "ext_datanet_net",
- "route_external": False,
- "description": "",
- "properties": {
- "name": "vlan_4004_tunnel_net",
- "mtu": 1500,
- "location_info": {
- "tenant": "admin",
- "vimid": 2,
- "availability_zone": "nova"
- },
- "ip_version": 4,
- "dhcp_enabled": True,
- "network_name": "vlan_4004_tunnel_net",
- "network_type": "vlan"
- }
- },
- {
- "route_id": "",
- "vl_id": "sfc_data_network",
- "route_external": False,
- "description": "",
- "properties": {
- "name": "sfc_data_network",
- "dhcp_enabled": True,
- "is_predefined": False,
- "location_info": {
- "tenant": "admin",
- "vimid": 2,
- "availability_zone": "nova"
- },
- "ip_version": 4,
- "mtu": 1500,
- "network_name": "sfc_data_network",
- "network_type": "vlan"
- }
- }
- ],
- "cps": [
- {
- "pnf_id": "m6000_s",
- "vl_id": "path2",
- "description": "",
- "cp_id": "m6000_data_out",
- "properties": {
- "direction": "bidirectional",
- "vnic_type": "normal",
- "bandwidth": 0,
- "mac_address": "11-22-33-22-11-44",
- "interface_name": "xgei-0/4/1/5",
- "ip_address": "176.1.1.2",
- "order": 0,
- "sfc_encapsulation": "mac"
- }
- },
- {
- "pnf_id": "m6000_s",
- "vl_id": "ext_datanet_net",
- "description": "",
- "cp_id": "m600_tunnel_cp",
- "properties": {
- "direction": "bidirectional",
- "vnic_type": "normal",
- "bandwidth": 0,
- "mac_address": "00-11-00-22-33-00",
- "interface_name": "gei-0/4/0/13",
- "ip_address": "191.167.100.5",
- "order": 0,
- "sfc_encapsulation": "mac"
- }
- },
- {
- "pnf_id": "m6000_s",
- "vl_id": "path2",
- "description": "",
- "cp_id": "m6000_data_in",
- "properties": {
- "direction": "bidirectional",
- "vnic_type": "normal",
- "bandwidth": 0,
- "mac_address": "11-22-33-22-11-41",
- "interface_name": "gei-0/4/0/7",
- "ip_address": "1.1.1.1",
- "order": 0,
- "sfc_encapsulation": "mac",
- "bond": "none"
- }
- },
- {
- "pnf_id": "m6000_s",
- "vl_id": "ext_mnet_net",
- "description": "",
- "cp_id": "m600_mnt_cp",
- "properties": {
- "direction": "bidirectional",
- "vnic_type": "normal",
- "bandwidth": 0,
- "mac_address": "00-11-00-22-33-11",
- "interface_name": "gei-0/4/0/1",
- "ip_address": "10.46.244.51",
- "order": 0,
- "sfc_encapsulation": "mac",
- "bond": "none"
- }
- }
- ],
- "metadata": {
- "invariant_id": "vcpe_ns_sff_1",
- "name": "VCPE_NS",
- "csarVersion": "v1.0",
- "csarType": "NSAR",
- "csarProvider": "ZTE",
- "version": 1,
- "vendor": "ZTE",
- "id": "VCPE_NS",
- "description": "vcpe_ns"
- }
- }
-
- def tearDown(self):
- pass
-
- def test_ns_pkg_distribute_when_ns_exists(self):
- NSDModel(id="1", nsd_id="2").save()
- resp = self.client.post("/api/nslcm/v1/nspackage", {"csarId": "1"}, format='json')
- self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
- self.assertEqual("failed", resp.data["status"])
- self.assertEqual("NS CSAR(1) already exists.", resp.data["statusDescription"])
-
- @mock.patch.object(restcall, 'call_req')
- def test_ns_pkg_distribute_when_csar_not_exist(self, mock_call_req):
- mock_call_req.return_value = [0, "[]", '200']
- resp = self.client.post("/api/nslcm/v1/nspackage", {"csarId": "1"}, format='json')
- self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
- self.assertEqual("failed", resp.data["status"])
- self.assertEqual("Failed to query artifact(services,1) from sdc.", resp.data["statusDescription"])
-
- @mock.patch.object(restcall, 'call_req')
- @mock.patch.object(sdc, 'download_artifacts')
- @mock.patch.object(toscaparser, 'parse_nsd')
- def test_ns_pkg_distribute_when_nsd_already_exists(self, mock_parse_nsd, mock_download_artifacts,
- mock_call_req):
- mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data)
- mock_download_artifacts.return_value = "/home/vcpe.csar"
- mock_call_req.return_value = [0, json.JSONEncoder().encode([{
- "uuid": "1",
- "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar"
- }]), '200']
- NSDModel(id="2", nsd_id="VCPE_NS").save()
- resp = self.client.post("/api/nslcm/v1/nspackage", {"csarId": "1"}, format='json')
- self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
- self.assertEqual("failed", resp.data["status"])
- self.assertEqual("NSD(VCPE_NS) already exists.", resp.data["statusDescription"])
-
- @mock.patch.object(restcall, 'call_req')
- @mock.patch.object(sdc, 'download_artifacts')
- @mock.patch.object(toscaparser, 'parse_nsd')
- def test_ns_pkg_distribute_when_nf_not_distributed(self, mock_parse_nsd, mock_download_artifacts,
- mock_call_req):
- mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data)
- mock_download_artifacts.return_value = "/home/vcpe.csar"
- mock_call_req.return_value = [0, json.JSONEncoder().encode([{
- "uuid": "1",
- "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar"
- }]), '200']
- resp = self.client.post("/api/nslcm/v1/nspackage", {"csarId": "1"}, format='json')
- self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
- self.assertEqual("failed", resp.data["status"])
- self.assertEqual("VNF package(vcpe_vfw_zte_1_0) is not distributed.", resp.data["statusDescription"])
-
- @mock.patch.object(restcall, 'call_req')
- @mock.patch.object(sdc, 'download_artifacts')
- @mock.patch.object(toscaparser, 'parse_nsd')
- def test_ns_pkg_distribute_when_successfully(self, mock_parse_nsd, mock_download_artifacts, mock_call_req):
- mock_parse_nsd.return_value = json.JSONEncoder().encode(self.nsd_data)
- mock_download_artifacts.return_value = "/home/vcpe.csar"
- mock_call_req.return_value = [0, json.JSONEncoder().encode([{
- "uuid": "1",
- "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar"
- }]), '200']
- NfPackageModel(uuid="1", nfpackageid="1", vnfdid="vcpe_vfw_zte_1_0").save()
- resp = self.client.post("/api/nslcm/v1/nspackage", {"csarId": "1"}, format='json')
- self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
- self.assertEqual("success", resp.data["status"])
- self.assertEqual("CSAR(1) distributed successfully.", resp.data["statusDescription"])
-
- ###############################################################################################################
-
- def test_ns_pkg_normal_delete(self):
- NSDModel(id="8", nsd_id="2").save()
- resp = self.client.delete("/api/nslcm/v1/nspackage/8")
- self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
- self.assertEqual("success", resp.data["status"])
- self.assertEqual("Delete CSAR(8) successfully.", resp.data["statusDescription"])
-
- def test_ns_pkg_force_delete(self):
- NSInstModel(id="1", nspackage_id="8").save()
- NSDModel(id="8", nsd_id="2").save()
- resp = self.client.delete("/api/nslcm/v1/nspackage/8force")
- self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
- self.assertEqual("success", resp.data["status"])
- self.assertEqual("Delete CSAR(8) successfully.", resp.data["statusDescription"])
-
- def test_ns_pkg_delete_when_pkg_in_using(self):
- NSInstModel(id="1", nspackage_id="8").save()
- NSDModel(id="8", nsd_id="2").save()
- resp = self.client.delete("/api/nslcm/v1/nspackage/8")
- self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
- self.assertEqual("failed", resp.data["status"])
- self.assertEqual("CSAR(8) is in using, cannot be deleted.", resp.data["statusDescription"])
-
- def test_ns_pkg_get_all(self):
- NSDModel(id="13", nsd_id="2", vendor="3", version="4").save()
-
- resp = self.client.get("/api/nslcm/v1/nspackage")
- self.assertEqual(resp.status_code, status.HTTP_200_OK)
- self.assertEqual({"csars": [{"csarId": "13", "nsdId": "2"}]}, resp.data)
-
- def test_ns_pkg_get_one(self):
- NSDModel(id="14", nsd_id="2", vendor="3", version="4").save()
- NSInstModel(id="1", nspackage_id="14", name="11").save()
-
- resp = self.client.get("/api/nslcm/v1/nspackage/14")
- self.assertEqual(resp.status_code, status.HTTP_200_OK)
- expect_data = {
- "csarId": "14",
- "packageInfo": {
- "nsdId": "2",
- "nsdProvider": "3",
- "nsdVersion": "4"
- },
- "nsInstanceInfo": [
- {
- "nsInstanceId": "1",
- "nsInstanceName": "11"
- }
- ]
- }
- self.assertEqual(expect_data, resp.data)
from lcm.packages import views
urlpatterns = [
- url(r'^api/nslcm/v1/nspackage$', views.ns_distribute_get, name='ns_distribute_get'),
- url(r'^api/nslcm/v1/nspackage/(?P<csarId>[0-9a-zA-Z\-\_]+)$', views.ns_rd_csar, name='ns_rd_csar'),
url(r'^api/nslcm/v1/vnfpackage$', views.nf_distribute_get, name='nf_distribute_get'),
url(r'^api/nslcm/v1/vnfpackage/(?P<csarId>[0-9a-zA-Z\-\_]+)$', views.nf_rd_csar, name='nf_rd_csar'),
#########################################################################################
- url(r'^api/nslcm/v0/nspackage/(?P<csarId>[0-9a-zA-Z\-\_]+)$', views.ns_access_csar, name='ns_access_csar'),
- url(r'^api/nslcm/v0/nspackage$', views.ns_on_boarding, name='ns_on_boarding'),
url(r'^api/nslcm/v0/nspackage/(?P<csarId>[0-9a-zA-Z\-\_]+)/deletionpending$',
views.ns_delete_pending_csar, name='ns_delete_pending_csar'),
url(r'^api/nslcm/v0/nspackage/(?P<csarId>[0-9a-zA-Z\-\_]+)/(?P<operation>(disabled|enabled))$',
from lcm.pub.utils.values import ignore_case_get
from lcm.pub.utils.syscomm import fun_name
-from lcm.packages import ns_package, nf_package, sdc_ns_package, sdc_nf_package
+from lcm.packages import ns_package, nf_package, sdc_nf_package
logger = logging.getLogger(__name__)
-@api_view(http_method_names=['POST', 'GET'])
-def ns_distribute_get(request, *args, **kwargs):
- logger.debug("Enter %s, method is %s", fun_name(), request.method)
- ret, normal_status = None, None
- if request.method == 'GET':
- ret = sdc_ns_package.SdcNsPackage().get_csars()
- normal_status = status.HTTP_200_OK
- else:
- csar_id = ignore_case_get(request.data, "csarId")
- logger.debug("csar_id is %s", csar_id)
- ret = sdc_ns_package.ns_on_distribute(csar_id)
- normal_status = status.HTTP_202_ACCEPTED
- logger.debug("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)
- return Response(data=ret[1], status=normal_status)
-
-
-@api_view(http_method_names=['GET', 'DELETE'])
-def ns_rd_csar(request, *args, **kwargs):
- csar_id = ignore_case_get(kwargs, "csarId")
- logger.info("Enter %s, method is %s, csar_id is %s", fun_name(), request.method, csar_id)
- ret, normal_status = None, None
- if request.method == 'GET':
- ret = sdc_ns_package.ns_get_csar(csar_id)
- normal_status = status.HTTP_200_OK
- else:
- force_delete = csar_id.endswith("force")
- if force_delete:
- csar_id = csar_id[:-5]
- ret = sdc_ns_package.ns_delete_csar(csar_id, force_delete)
- normal_status = status.HTTP_202_ACCEPTED
- logger.info("Leave %s, Return value is %s", fun_name(), str(ret))
- if ret[0] != 0:
- return Response(data={'error': ret[1]}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
- return Response(data=ret[1], status=normal_status)
-
-
####################################################################################################
@api_view(http_method_names=['POST', 'GET'])
def nf_distribute_get(request, *args, **kwargs):
####################################################################################################
-@api_view(http_method_names=['POST'])
-def ns_on_boarding(request, *args, **kwargs):
- csar_id = ignore_case_get(request.data, "csarId")
- logger.info("Enter %s, method is %s, csar_id is %s", fun_name(), request.method, csar_id)
- ret = ns_package.ns_on_boarding(csar_id)
- logger.info("Leave %s, Return value is %s", fun_name(), str(ret))
- if ret[0] != 0:
- return Response(data={'error': ret[1]}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
- return Response(data=ret[1], status=status.HTTP_202_ACCEPTED)
-
-
-@api_view(http_method_names=['GET', 'DELETE'])
-def ns_access_csar(request, *args, **kwargs):
- csar_id = ignore_case_get(kwargs, "csarId")
- logger.info("Enter %s, method is %s, csar_id is %s", fun_name(), request.method, csar_id)
- ret, normal_status = None, None
- if request.method == 'GET':
- ret = ns_package.ns_get_csar(csar_id)
- normal_status = status.HTTP_200_OK
- else:
- ret = ns_package.ns_delete_csar(csar_id)
- normal_status = status.HTTP_202_ACCEPTED
- logger.info("Leave %s, Return value is %s", fun_name(), str(ret))
- if ret[0] != 0:
- return Response(data={'error': ret[1]}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
- return Response(data=ret[1], status=normal_status)
@api_view(http_method_names=['DELETE'])
Code Review / vfc / nfvo / lcm.git / commitdiff