from catalog.pub.utils.values import remove_none_key
from catalog.pub.config import config as pub_config
import traceback
+from django.db.models import Q
logger = logging.getLogger(__name__)
pass
def send_notification(self, notification, filters, isvnfpkg):
- logger.info("Send Notifications to the callbackUri")
subscriptions_filter = {v + "__contains": notification[k] for k, v in filters.items()}
- logger.debug('send_notification subscriptions_filter = %s' % subscriptions_filter)
subscriptions_filter = remove_none_key(subscriptions_filter)
+ logger.debug('send_notification subscriptions_filter = %s' % subscriptions_filter)
+ q1 = Q()
+ q1.connector = 'OR'
+ for k, v in subscriptions_filter.items():
+ q1.children.append((k, v))
if isvnfpkg:
- subscriptions = VnfPkgSubscriptionModel.objects.filter(**subscriptions_filter)
+ subscriptions = VnfPkgSubscriptionModel.objects.filter(q1)
subscription_root_uri = const.VNFPKG_SUBSCRIPTION_ROOT_URI
else:
- subscriptions = NsdmSubscriptionModel.objects.filter(**subscriptions_filter)
+ subscriptions = NsdmSubscriptionModel.objects.filter(q1)
subscription_root_uri = const.NSDM_SUBSCRIPTION_ROOT_URI
+
if not subscriptions.exists():
logger.info("No subscriptions created for the filters %s" % notification)
return
try:
resp = requests.post(callbackuri, data=notification, headers={'Connection': 'close'})
if resp.status_code != status.HTTP_204_NO_CONTENT:
- logger.error("Sendingnotification to %s failed: %s" % (callbackuri, resp.text))
+ logger.error("Sending notification to %s failed: %s" % (callbackuri, resp.text))
else:
logger.info("Sending notification to %s successfully.", callbackuri)
except:
import uuid
from catalog.packages.biz.common import parse_file_range, read, save
-from catalog.packages.const import PKG_STATUS
from catalog.pub.config.config import CATALOG_ROOT_PATH
from catalog.pub.database.models import NSPackageModel, PnfPackageModel, VnfPackageModel
from catalog.pub.exceptions import CatalogException, ResourceNotFoundException
from catalog.pub.utils import fileutil, toscaparser
from catalog.pub.utils.values import ignore_case_get
+from catalog.packages.biz.notificationsutil import prepare_nsd_notification, NotificationsUtil
+from catalog.packages import const
logger = logging.getLogger(__name__)
user_defined_data = ignore_case_get(data, 'userDefinedData', {})
data = {
'id': id if id else str(uuid.uuid4()),
- 'nsdOnboardingState': PKG_STATUS.CREATED,
- 'nsdOperationalState': PKG_STATUS.DISABLED,
- 'nsdUsageState': PKG_STATUS.NOT_IN_USE,
+ 'nsdOnboardingState': const.PKG_STATUS.CREATED,
+ 'nsdOperationalState': const.PKG_STATUS.DISABLED,
+ 'nsdUsageState': const.PKG_STATUS.NOT_IN_USE,
'userDefinedData': user_defined_data,
'_links': None # TODO
}
ns_pkgs.delete()
ns_pkg_path = os.path.join(CATALOG_ROOT_PATH, nsd_info_id)
fileutil.delete_dirs(ns_pkg_path)
+ send_notification(const.NSD_NOTIFICATION_TYPE.NSD_DELETION, nsd_info_id)
logger.info('NSD(%s) has been deleted.' % nsd_info_id)
def upload(self, nsd_info_id, remote_file):
if not ns_pkgs.exists():
logger.error('NSD(%s) does not exist.' % nsd_info_id)
raise CatalogException('NSD(%s) does not exist.' % nsd_info_id)
- ns_pkgs.update(onboardingState=PKG_STATUS.UPLOADING)
+ ns_pkgs.update(onboardingState=const.PKG_STATUS.UPLOADING)
local_file_name = save(remote_file, nsd_info_id)
logger.info('NSD(%s) content has been uploaded.' % nsd_info_id)
if not ns_pkgs.exists():
logger.error('NSD(%s) does not exist.' % nsd_info_id)
raise ResourceNotFoundException('NSD(%s) does not exist.' % nsd_info_id)
- if ns_pkgs[0].onboardingState != PKG_STATUS.ONBOARDED:
+ if ns_pkgs[0].onboardingState != const.PKG_STATUS.ONBOARDED:
logger.error('NSD(%s) is not ONBOARDED.' % nsd_info_id)
raise CatalogException('NSD(%s) is not ONBOARDED.' % nsd_info_id)
def parse_nsd_and_save(self, nsd_info_id, local_file_name):
logger.info('Start to process NSD(%s)...' % nsd_info_id)
ns_pkgs = NSPackageModel.objects.filter(nsPackageId=nsd_info_id)
- ns_pkgs.update(onboardingState=PKG_STATUS.PROCESSING)
+ ns_pkgs.update(onboardingState=const.PKG_STATUS.PROCESSING)
nsd_json = toscaparser.parse_nsd(local_file_name)
logger.debug("%s", nsd_json)
other_nspkg = NSPackageModel.objects.filter(nsdId=nsd_id)
if other_nspkg and other_nspkg[0].nsPackageId != nsd_info_id:
logger.warn("NSD(%s,%s) already exists.", nsd_id, other_nspkg[0].nsPackageId)
+ send_notification(const.NSD_NOTIFICATION_TYPE.NSD_ONBOARDING_FAILURE, nsd_info_id, nsd_id)
raise CatalogException("NSD(%s) already exists." % nsd_id)
for vnf in nsd["vnfs"]:
nsdDescription=nsd.get("description", ""),
nsdVersion=nsd_version,
invariantId=invariant_id,
- onboardingState=PKG_STATUS.ONBOARDED,
- operationalState=PKG_STATUS.ENABLED,
- usageState=PKG_STATUS.NOT_IN_USE,
+ onboardingState=const.PKG_STATUS.ONBOARDED,
+ operationalState=const.PKG_STATUS.ENABLED,
+ usageState=const.PKG_STATUS.NOT_IN_USE,
nsPackageUri=local_file_name,
sdcCsarId=nsd_info_id,
localFilePath=local_file_name,
nsdModel=nsd_json
)
+ send_notification(const.NSD_NOTIFICATION_TYPE.NSD_ONBOARDING, nsd_info_id, nsd_id)
logger.info('NSD(%s) has been processed.' % nsd_info_id)
def fill_resp_data(self, ns_pkg):
def handle_upload_failed(self, nsd_info_id):
ns_pkg = NSPackageModel.objects.filter(nsPackageId=nsd_info_id)
- ns_pkg.update(onboardingState=PKG_STATUS.CREATED)
+ ns_pkg.update(onboardingState=const.PKG_STATUS.CREATED)
+
+
+def send_notification(type, nsd_info_id, nsd_id=None, failure_details=None, operational_state=None):
+ data = prepare_nsd_notification(nsd_info_id=nsd_info_id,
+ nsd_id=nsd_id,
+ notification_type=type,
+ failure_details=failure_details,
+ operational_state=operational_state)
+ filters = {
+ 'nsdInfoId': 'nsdInfoId',
+ 'nsdId': 'nsdId',
+ }
+ logger.debug('Notify request data = %s' % data)
+ logger.debug('Notify request filters = %s' % filters)
+ NotificationsUtil().send_notification(data, filters, False)
from catalog.pub.exceptions import CatalogException, ResourceNotFoundException
from catalog.pub.utils import fileutil, toscaparser
from catalog.pub.utils.values import ignore_case_get
+from catalog.packages.biz.notificationsutil import prepare_pnfd_notification, NotificationsUtil
+from catalog.packages import const
logger = logging.getLogger(__name__)
logger.info('Start to upload PNFD(%s)...' % pnfd_info_id)
pnf_pkgs = PnfPackageModel.objects.filter(pnfPackageId=pnfd_info_id)
if not pnf_pkgs.exists():
- logger.info('PNFD(%s) does not exist.' % pnfd_info_id)
- raise CatalogException('PNFD (%s) does not exist.' % pnfd_info_id)
+ details = 'PNFD(%s) is not CREATED.' % pnfd_info_id
+ logger.info(details)
+ send_notification(
+ type=const.NSD_NOTIFICATION_TYPE.PNFD_ONBOARDING_FAILURE,
+ pnfd_info_id=pnfd_info_id,
+ failure_details=details
+ )
+ raise CatalogException(details)
pnf_pkgs.update(onboardingState=PKG_STATUS.UPLOADING)
local_file_name = save(remote_file, pnfd_info_id)
pnf_pkgs.delete()
pnf_pkg_path = os.path.join(CATALOG_ROOT_PATH, pnfd_info_id)
fileutil.delete_dirs(pnf_pkg_path)
+ send_notification(const.NSD_NOTIFICATION_TYPE.PNFD_DELETION, pnfd_info_id, del_pnfd_id)
logger.debug('PNFD(%s) has been deleted.' % pnfd_info_id)
def download(self, pnfd_info_id):
localFilePath=local_file_name,
pnfdModel=pnfd_json
)
+ send_notification(const.NSD_NOTIFICATION_TYPE.PNFD_ONBOARDING, pnfd_info_id, pnfd_id)
logger.info('PNFD(%s) has been processed.' % pnfd_info_id)
def fill_response_data(self, pnf_pkg):
logger.error(e.args[0])
return [1, e.args[0]]
return [0, ret]
+
+
+def send_notification(type, pnfd_info_id, pnfd_id=None, failure_details=None):
+ data = prepare_pnfd_notification(pnfd_info_id=pnfd_info_id,
+ pnfd_id=pnfd_id,
+ notification_type=type,
+ failure_details=failure_details)
+ filters = {
+ 'pnfdId': 'pnfdId',
+ 'pnfdInfoIds': 'pnfdInfoIds',
+ }
+ logger.debug('Notify request data = %s' % data)
+ logger.debug('Notify request filters = %s' % filters)
+ NotificationsUtil().send_notification(data, filters, False)
# logger.error("VNF package(%s) is not CREATED" % vnf_pkg_id)
# raise CatalogException("VNF package(%s) is not CREATED" % vnf_pkg_id)
vnf_pkg.update(onboardingState=const.PKG_STATUS.UPLOADING)
- send_notification(vnf_pkg_id, const.PKG_NOTIFICATION_TYPE.ONBOARDING,
- const.PKG_CHANGE_TYPE.OP_STATE_CHANGE)
local_file_name = save(remote_file, vnf_pkg_id)
logger.info('VNF package(%s) has been uploaded.' % vnf_pkg_id)
localFilePath=vnf_pkg_path,
vnfPackageUri=os.path.split(vnf_pkg_path)[-1]
)
+ send_notification(vnf_pkg_id, const.PKG_NOTIFICATION_TYPE.ONBOARDING,
+ const.PKG_CHANGE_TYPE.OP_STATE_CHANGE)
else:
raise CatalogException("VNF propeties and metadata in VNF Package(id=%s) are empty." % vnf_pkg_id)
logger.info('VNF package(%s) has been processed(done).' % vnf_pkg_id)
"vnfProvider": "zte",
"vnfmInfo": "zte",
"defaultLocalizationLanguage": "english",
- "vnfdId": "zte-hss-1.0",
+ "vnfdId": "00342b18-a5c7-11e8-998c-bf1755941f12",
"id": "zte-hss-1.0",
"vnfProductInfoDescription": "hss",
"vnfdVersion": "1.0.0",
},
"vnf": {
"properties": {
- "descriptor_id": "zte-hss-1.0",
+ "descriptor_id": "00342b18-a5c7-11e8-998c-bf1755941f12",
"descriptor_version": "1.0.0",
"software_version": "1.0.0",
"provider": "zte"
"description": "vcpe_ns"},
"ns": {
"properties": {
- "descriptor_id": "VCPE_NS",
+ # "descriptor_id": "VCPE_NS",
+ "descriptor_id": "b632bddc-bccd-4180-bd8d-4e8a9578eff7",
"version": 1,
"name": "VCPE_NS",
"desginer": "ZTE",
}
},
"vnfdId": [
- "3fa85f64-5717-4562-b3fc-2c963f66afa6"
+ "00342b18-a5c7-11e8-998c-bf1755941f12"
],
"vnfPkgId": [
"3fa85f64-5717-4562-b3fc-2c963f66afa6"
data = fp.read()
file_content = '%s%s' % (file_content, data)
ns_pkg = NSPackageModel.objects.filter(nsPackageId="22")
- self.assertEqual("VCPE_NS", ns_pkg[0].nsdId)
+ self.assertEqual("b632bddc-bccd-4180-bd8d-4e8a9578eff7", ns_pkg[0].nsdId)
self.assertEqual(PKG_STATUS.ONBOARDED, ns_pkg[0].onboardingState)
self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
self.assertEqual(None, resp.data)
import json
import mock
import uuid
+import os
from django.test import TestCase
from rest_framework.test import APIClient
from rest_framework import status
from catalog.packages import const
from catalog.pub.config import config as pub_config
import catalog.pub.utils.timeutil
+from catalog.packages.tests.const import nsd_data
+from catalog.pub.database.models import NSPackageModel, VnfPackageModel, PnfPackageModel
+from catalog.pub.config.config import CATALOG_ROOT_PATH
+from catalog.pub.utils import toscaparser
class TestNsdmSubscription(TestCase):
self.assertEqual(response.status_code,
status.HTTP_500_INTERNAL_SERVER_ERROR)
+ @mock.patch("requests.post")
+ @mock.patch.object(toscaparser, 'parse_nsd')
+ @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
+ @mock.patch("requests.get")
+ @mock.patch.object(uuid, 'uuid4')
+ def test_nsdm_subscribe_trigger_notification(self, mock_uuid4, mock_requests, mock_nowtime, mock_parse_nsd,
+ mock_requests_post):
+ mock_requests.return_value.status_code = 204
+ mock_requests.get.return_value.status_code = 204
+ mock_uuid4.return_value = "1111"
+ mock_nowtime.return_value = "nowtime()"
+
+ subscription_req = {
+ "callbackUri": "http://callbackuri.com",
+ "authentication": {
+ "authType": ["BASIC"],
+ "paramsBasic": {
+ "userName": "username",
+ "password": "password"
+ }
+ },
+ "filter": {
+ "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"]
+ }
+ }
+ response = self.client.post("/api/nsd/v1/subscriptions",
+ data=subscription_req, format='json')
+ self.assertEqual(201, response.status_code)
+
+ self.user_defined_data = {
+ 'key1': 'value1',
+ 'key2': 'value2',
+ 'key3': 'value3',
+ }
+ user_defined_data_json = json.JSONEncoder().encode(self.user_defined_data)
+ mock_parse_nsd.return_value = json.JSONEncoder().encode(nsd_data)
+ VnfPackageModel(
+ vnfPackageId="111",
+ vnfdId="vcpe_vfw_zte_1_0"
+ ).save()
+
+ PnfPackageModel(
+ pnfPackageId="112",
+ pnfdId="m6000_s"
+ ).save()
+
+ NSPackageModel(
+ nsPackageId='d0ea5ec3-0b98-438a-9bea-488230cff174',
+ operationalState='DISABLED',
+ usageState='NOT_IN_USE',
+ userDefinedData=user_defined_data_json,
+ ).save()
+
+ with open('nsd_content.txt', 'wt') as fp:
+ fp.write('test')
+ with open('nsd_content.txt', 'rt') as fp:
+ resp = self.client.put(
+ "/api/nsd/v1/ns_descriptors/d0ea5ec3-0b98-438a-9bea-488230cff174/nsd_content",
+ {'file': fp},
+ )
+ file_content = ''
+ with open(os.path.join(CATALOG_ROOT_PATH, 'd0ea5ec3-0b98-438a-9bea-488230cff174/nsd_content.txt')) as fp:
+ data = fp.read()
+ file_content = '%s%s' % (file_content, data)
+ ns_pkg = NSPackageModel.objects.filter(nsPackageId="d0ea5ec3-0b98-438a-9bea-488230cff174")
+ self.assertEqual("b632bddc-bccd-4180-bd8d-4e8a9578eff7", ns_pkg[0].nsdId)
+ self.assertEqual(const.PKG_STATUS.ONBOARDED, ns_pkg[0].onboardingState)
+ self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
+ self.assertEqual(None, resp.data)
+ self.assertEqual(file_content, 'test')
+ os.remove('nsd_content.txt')
+ expect_callbackuri = "http://callbackuri.com"
+ expect_notification = {
+ 'id': "1111",
+ 'notificationType': const.NSD_NOTIFICATION_TYPE.NSD_ONBOARDING,
+ 'timeStamp': "nowtime()",
+ 'nsdInfoId': "d0ea5ec3-0b98-438a-9bea-488230cff174",
+ 'nsdId': "b632bddc-bccd-4180-bd8d-4e8a9578eff7",
+ 'onboardingFailureDetails': None,
+ 'nsdOperationalState': None,
+ "subscriptionId": "1111",
+ '_links': {
+ 'subscription': {
+ 'href': 'http://%s:%s/%s%s' % (pub_config.MSB_SERVICE_IP,
+ pub_config.MSB_SERVICE_PORT,
+ const.NSDM_SUBSCRIPTION_ROOT_URI,
+ "1111")},
+ 'nsdInfo': {
+ 'href': 'http://%s:%s/%s/ns_descriptors/%s' % (pub_config.MSB_SERVICE_IP,
+ pub_config.MSB_SERVICE_PORT,
+ const.NSD_URL_PREFIX,
+ "d0ea5ec3-0b98-438a-9bea-488230cff174")
+ }
+ }
+ }
+ mock_requests_post.assert_called_with(expect_callbackuri, data=expect_notification, headers={'Connection': 'close'})
+
class NotificationTest(TestCase):
def setUp(self):
"toscaModelURL": "https://127.0.0.1:1234/sdc/v1/vcpe.csar",
"distributionStatus": "DISTRIBUTED"
}), '200']
- NSPackageModel(nsPackageId="2", nsdId="VCPE_NS").save()
+ NSPackageModel(nsPackageId="2", nsdId="b632bddc-bccd-4180-bd8d-4e8a9578eff7").save()
resp = self.client.post(
"/api/catalog/v1/nspackages", {"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.",
+ "NSD(b632bddc-bccd-4180-bd8d-4e8a9578eff7) already exists.",
resp.data["statusDescription"])
@mock.patch.object(restcall, 'call_req')
mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
response = self.client.put("%s/222/package_content" % VNF_BASE_URL, data=data)
vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId="222")
- self.assertEqual("zte-hss-1.0", vnf_pkg[0].vnfdId)
+ self.assertEqual("00342b18-a5c7-11e8-998c-bf1755941f12", vnf_pkg[0].vnfdId)
self.assertEqual(PKG_STATUS.ONBOARDED, vnf_pkg[0].onboardingState)
self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
vnf_pkg_id = vnf_pkg.vnfPackageId
VnfPkgUploadThread(req_data, vnf_pkg_id).run()
vnf_pkg1 = VnfPackageModel.objects.filter(vnfPackageId="222")
- self.assertEqual("zte-hss-1.0", vnf_pkg1[0].vnfdId)
+ self.assertEqual("00342b18-a5c7-11e8-998c-bf1755941f12", vnf_pkg1[0].vnfdId)
def test_upload_from_uri_bad_req(self):
req_data = {"username": "123"}
self.assertTrue(zipfile.is_zipfile("vnfd.csar"))
os.remove("vnfd.csar")
+ def test_download_vnfd_when_pkg_not_exist(self):
+ response = self.client.get(VNF_BASE_URL + "/222/vnfd")
+ self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+ def test_download_vnfd_when_catch_cataloge_exception(self):
+ VnfPackageModel.objects.create(
+ vnfPackageId="222",
+ onboardingState="CREATED",
+ localFilePath="vnfPackage.csar"
+ )
+ response = self.client.get(VNF_BASE_URL + "/222/vnfd")
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @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(VNF_BASE_URL, 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(VNF_BASE_URL + "/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(VNF_BASE_URL + "/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(VNF_BASE_URL)
+ 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(VNF_BASE_URL + "/222/package_content", data=data)
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
-def test_download_vnfd_when_pkg_not_exist(self):
- response = self.client.get(VNF_BASE_URL + "/222/vnfd")
- self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
-
-
-def test_download_vnfd_when_catch_cataloge_exception(self):
- VnfPackageModel.objects.create(
- vnfPackageId="222",
- onboardingState="CREATED",
- localFilePath="vnfPackage.csar"
- )
- response = self.client.get(VNF_BASE_URL + "/222/vnfd")
- self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
-
-
-@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(VNF_BASE_URL, 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(VNF_BASE_URL + "/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(VNF_BASE_URL + "/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(VNF_BASE_URL)
- 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(VNF_BASE_URL + "/222/package_content", data=data)
- self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
-
-
-@mock.patch.object(VnfPkgUploadThread, 'start')
-def test_upload_from_uri_when_catch_exception(self, mock_start):
- req_data = {"addressInformation": "https://127.0.0.1:1234/sdc/v1/hss.csar"}
- mock_start.side_effect = TypeError("integer type")
- response = self.client.post(VNF_BASE_URL + "/111/package_content/upload_from_uri", data=req_data)
- self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
-
-
-@mock.patch.object(VnfPackage, 'download')
-def test_fetch_vnf_pkg_when_catch_exception(self, mock_download):
- mock_download.side_effect = TypeError("integer type")
- response = self.client.get(VNF_BASE_URL + "/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(VNF_BASE_URL + "/222/package_content", data=data)
- self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
- response = self.client.get(VNF_BASE_URL + "/222/artifacts/image")
- self.assertEqual(response.status_code, status.HTTP_200_OK)
- self.assertEqual(response.getvalue(), b"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(VNF_BASE_URL + "/222/package_content", data=data)
- self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
- response = self.client.get(VNF_BASE_URL + "/1451/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(VNF_BASE_URL + "/222/package_content", data=data)
- self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
- response = self.client.get(VNF_BASE_URL + "/222/artifacts/image1")
- self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+ @mock.patch.object(VnfPkgUploadThread, 'start')
+ def test_upload_from_uri_when_catch_exception(self, mock_start):
+ req_data = {"addressInformation": "https://127.0.0.1:1234/sdc/v1/hss.csar"}
+ mock_start.side_effect = TypeError("integer type")
+ response = self.client.post(VNF_BASE_URL + "/111/package_content/upload_from_uri", data=req_data)
+ self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+ @mock.patch.object(VnfPackage, 'download')
+ def test_fetch_vnf_pkg_when_catch_exception(self, mock_download):
+ mock_download.side_effect = TypeError("integer type")
+ response = self.client.get(VNF_BASE_URL + "/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(VNF_BASE_URL + "/222/package_content", data=data)
+ self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
+ response = self.client.get(VNF_BASE_URL + "/222/artifacts/image")
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual(response.getvalue(), b"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(VNF_BASE_URL + "/222/package_content", data=data)
+ self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
+ response = self.client.get(VNF_BASE_URL + "/1451/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(VNF_BASE_URL + "/222/package_content", data=data)
+ self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
+ response = self.client.get(VNF_BASE_URL + "/222/artifacts/image1")
+ self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
import uuid
import mock
+import json
+import os
from rest_framework.test import APIClient
from django.test import TestCase
from catalog.pub.database.models import VnfPkgSubscriptionModel, VnfPackageModel
-from .const import vnf_subscription_data
+from .const import vnf_subscription_data, vnfd_data
from catalog.packages.biz.notificationsutil import NotificationsUtil, prepare_vnfpkg_notification
from catalog.packages import const
from catalog.pub.config import config as pub_config
import catalog.pub.utils.timeutil
+from catalog.pub.utils import toscaparser
+from catalog.pub.config.config import CATALOG_ROOT_PATH
+from rest_framework import status
class TestNfPackageSubscription(TestCase):
response = self.client.delete("/api/vnfpkgm/v1/subscriptions/%s" % dummy_uuid)
self.assertEqual(404, response.status_code)
+ @mock.patch("requests.get")
+ @mock.patch.object(toscaparser, 'parse_vnfd')
+ @mock.patch("requests.post")
+ @mock.patch("uuid.uuid4")
+ @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
+ def test_vnfpkg_subscript_notify(self, mock_nowtime, mock_uuid, mock_requests_post, mock_parse_vnfd, mock_requests_get):
+ mock_nowtime.return_value = "nowtime()"
+ uuid_subscriptid = "99442b18-a5c7-11e8-998c-bf1755941f13"
+ uuid_vnfPackageId = "3fa85f64-5717-4562-b3fc-2c963f66afa6"
+ uuid_vnfdid = "00342b18-a5c7-11e8-998c-bf1755941f12"
+ mock_uuid.side_effect = [uuid_subscriptid, "1111"]
+ mock_requests_get.return_value.status_code = 204
+ mock_parse_vnfd.return_value = json.JSONEncoder().encode(vnfd_data)
+
+ response = self.client.post(
+ "/api/vnfpkgm/v1/subscriptions",
+ data=vnf_subscription_data,
+ format='json')
+ self.assertEqual(201, response.status_code)
+
+ data = {'file': open(os.path.join(CATALOG_ROOT_PATH, "empty.txt"), "rt")}
+ VnfPackageModel.objects.create(
+ vnfPackageId=uuid_vnfPackageId,
+ onboardingState="CREATED"
+ )
+
+ response = self.client.put("/api/vnfpkgm/v1/vnf_packages/%s/package_content" % uuid_vnfPackageId, data=data)
+ vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId=uuid_vnfPackageId)
+ self.assertEqual(uuid_vnfdid, vnf_pkg[0].vnfdId)
+ self.assertEqual(const.PKG_STATUS.ONBOARDED, vnf_pkg[0].onboardingState)
+ self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
+
+ expect_notification = {
+ 'id': "1111",
+ 'notificationType': const.PKG_NOTIFICATION_TYPE.ONBOARDING,
+ 'timeStamp': "nowtime()",
+ 'vnfPkgId': uuid_vnfPackageId,
+ 'vnfdId': uuid_vnfdid,
+ 'changeType': const.PKG_CHANGE_TYPE.OP_STATE_CHANGE,
+ 'operationalState': None,
+ "subscriptionId": uuid_subscriptid,
+ '_links': {
+ 'subscription': {
+ 'href': 'http://%s:%s/%s%s' % (pub_config.MSB_SERVICE_IP,
+ pub_config.MSB_SERVICE_PORT,
+ const.VNFPKG_SUBSCRIPTION_ROOT_URI,
+ uuid_subscriptid)},
+ 'vnfPackage': {
+ 'href': 'http://%s:%s/%s/vnf_packages/%s' % (pub_config.MSB_SERVICE_IP,
+ pub_config.MSB_SERVICE_PORT,
+ const.PKG_URL_PREFIX,
+ uuid_vnfPackageId)
+ }
+ }
+ }
+ mock_requests_post.assert_called_with(vnf_subscription_data["callbackUri"], data=expect_notification,
+ headers={'Connection': 'close'})
+
class NotificationTest(TestCase):
def setUp(self):
+ VnfPackageModel.objects.all().delete()
+ VnfPkgSubscriptionModel.objects.all().delete()
+
+ def tearDown(self):
+ VnfPackageModel.objects.all().delete()
+ VnfPkgSubscriptionModel.objects.all().delete()
+
+ @mock.patch("requests.post")
+ @mock.patch("uuid.uuid4")
+ @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
+ def test_vnfpkg_manual_notify(self, mock_nowtime, mock_uuid, mock_requests_post):
VnfPackageModel(vnfPackageId="vnfpkgid1",
vnfdId="vnfdid1"
).save()
vnfd_id="vnfdid1",
vnf_pkg_id="vnfpkgid1"
).save()
-
- def tearDown(self):
- VnfPackageModel.objects.all().delete()
- VnfPkgSubscriptionModel.objects.all().delete()
-
- @mock.patch("requests.post")
- @mock.patch("uuid.uuid4")
- @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
- def test_vnfpkg_notify(self, mock_nowtime, mock_uuid, mock_requests_post):
mock_nowtime.return_value = "nowtime()"
mock_uuid.return_value = "1111"
notification_content = prepare_vnfpkg_notification("vnfpkgid1", const.PKG_NOTIFICATION_TYPE.CHANGE,
"uuid": "1",
"toscaModelURL": "https://127.0.0.1:1234/sdc/v1/hss.csar"
}]), '200']
- VnfPackageModel(vnfPackageId="2", vnfdId="zte-hss-1.0").save()
+ VnfPackageModel(vnfPackageId="2", vnfdId="00342b18-a5c7-11e8-998c-bf1755941f12").save()
NfDistributeThread(
csar_id="1",
vim_ids=["1"],
lab_vim_id="",
job_id="2"
).run()
- self.assert_job_result("2", 255, "VNF package(zte-hss-1.0) already exists.")
+ self.assert_job_result("2", 255, "VNF package(00342b18-a5c7-11e8-998c-bf1755941f12) already exists.")
@mock.patch.object(restcall, 'call_req')
@mock.patch.object(sdc, 'download_artifacts')
)
@view_safe_call_with_log(logger=logger)
def post(self, request):
- logger.debug("Create VNF package Subscription> %s" % request.data)
+ mydata = request.data
+ # if hasattr(request.data, "lists"):
+ # mydata = dict(request.data.lists())
+ logger.debug("Create VNF package Subscription> %s" % mydata)
- vnf_pkg_subscription_request = validate_req_data(request.data, PkgmSubscriptionRequestSerializer)
+ vnf_pkg_subscription_request = validate_req_data(mydata, PkgmSubscriptionRequestSerializer)
data = CreateSubscription(vnf_pkg_subscription_request.data).do_biz()
subscription_info = validate_data(data, PkgmSubscriptionSerializer)
return Response(data=subscription_info.data, status=status.HTTP_201_CREATED)
Code Review / modeling / etsicatalog.git / commitdiff