# limitations under the License.
import json
+import os
+
import mock
-from rest_framework.test import APIClient
from django.test import TestCase
+from requests.auth import HTTPBasicAuth
from rest_framework import status
+from rest_framework.test import APIClient
+
+import catalog.pub.utils.timeutil
+from catalog.packages import const
from catalog.packages.biz.sdc_vnf_package import NfDistributeThread, NfPkgDeleteThread
+from catalog.pub.config.config import CATALOG_ROOT_PATH
from catalog.pub.database.models import JobStatusModel, JobModel
from catalog.pub.database.models import VnfPackageModel
from catalog.pub.msapi import sdc
"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')
"inputs": []
}
resp = self.client.post(
- "/api/catalog/v1/parservnfd",
+ "/api/parser/v1/parservnfd",
req_data,
format='json'
)
def test_vnfd_parse_when_csar_not_exist(self):
req_data = {"csarId": "1", "inputs": []}
resp = self.client.post(
- "/api/catalog/v1/parservnfd",
+ "/api/parser/v1/parservnfd",
req_data,
format='json'
)
self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
self.assertEqual(resp.data, {"error": "VNF CSAR(1) does not exist."})
+
+ @mock.patch("requests.post")
+ @mock.patch.object(sdc, 'download_artifacts')
+ @mock.patch.object(sdc, 'get_artifact')
+ @mock.patch("requests.get")
+ @mock.patch("uuid.uuid4")
+ @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
+ def test_service_pkg_distribute_and_notify(self, mock_nowtime, mock_uuid, mock_requests_get, mock_get_artifact,
+ mock_download_artifacts, mock_requests_post):
+ mock_nowtime.return_value = "2019-02-16 14:41:16"
+ uuid_subscriptid = "99442b18-a5c7-11e8-998c-bf1755941f13"
+ uuid_csarid = "1234"
+ mock_uuid.side_effect = [uuid_subscriptid, "1111", "2222"]
+ mock_requests_get.return_value.status_code = 204
+ vnf_subscription_data = {
+ "filter": {
+ "notificationTypes": [
+ "VnfPackageOnboardingNotification",
+ "VnfPackageChangeNotification"
+ ],
+ "vnfPkgId": [uuid_csarid],
+ "operationalState": ["ENABLED", "DISABLED"]
+ },
+ "callbackUri": "https://so-vnfm-simulator.onap:9093/vnfpkgm/v1/notification",
+ "authentication": {
+ "authType": [
+ "BASIC"
+ ],
+ "paramsBasic": {
+ "userName": "vnfm",
+ "password": "password1$"
+ }
+ }
+ }
+ response = self.client.post(
+ "/api/vnfpkgm/v1/subscriptions",
+ data=vnf_subscription_data,
+ format='json')
+ self.assertEqual(201, response.status_code)
+ mock_download_artifacts.return_value = os.path.join(CATALOG_ROOT_PATH, "Sol004VnfCSAR.csar")
+ mock_get_artifact.return_value = {
+ "uuid": "c94490a0-f7ef-48be-b3f8-8d8662a37236",
+ "invariantUUID": "63eaec39-ffbe-411c-a838-448f2c73f7eb",
+ "name": "Sol004VnfCSAR",
+ "version": "2.0",
+ "toscaModelURL": "/sdc/v1/catalog/resources/c94490a0-f7ef-48be-b3f8-8d8662a37236/toscaModel",
+ "category": "Volte",
+ "subCategory": "VolteVF",
+ "resourceType": "VF",
+ "lifecycleState": "CERTIFIED",
+ "lastUpdaterUserId": "jh0003"
+ }
+ NfDistributeThread(csar_id=uuid_csarid, vim_ids=["1"], lab_vim_id="", job_id="4").on_distribute()
+ print(VnfPackageModel.objects.all().values())
+ expect_onboarding_notification = {
+ 'id': "1111",
+ 'notificationType': const.PKG_NOTIFICATION_TYPE.ONBOARDING,
+ 'timeStamp': "2019-02-16 14:41:16",
+ 'vnfPkgId': "1234",
+ 'vnfdId': "b1bb0ce7-2222-4fa7-95ed-4840d70a1177",
+ '_links': {
+ 'vnfPackage': {
+ 'href': '/%s/vnf_packages/%s' % (const.PKG_URL_PREFIX, uuid_csarid)},
+ 'subscription': {
+ 'href': '/%s%s' % (const.VNFPKG_SUBSCRIPTION_ROOT_URI, uuid_subscriptid)}
+
+ },
+ "subscriptionId": uuid_subscriptid
+ }
+ mock_requests_post.return_value.status_code = 204
+ mock_requests_post.assert_called_with(vnf_subscription_data["callbackUri"],
+ data=json.dumps(expect_onboarding_notification),
+ headers={'Connection': 'close',
+ 'content-type': 'application/json',
+ 'accept': 'application/json'},
+ auth=HTTPBasicAuth("vnfm", "password1$"),
+ verify=False)
+ mock_requests_post.return_value.status_code = 204
+ expect_deleted_notification = {
+ 'id': "2222",
+ 'notificationType': const.PKG_NOTIFICATION_TYPE.CHANGE,
+ 'timeStamp': "2019-02-16 14:41:16",
+ 'vnfPkgId': "1234",
+ 'vnfdId': "b1bb0ce7-2222-4fa7-95ed-4840d70a1177",
+ '_links': {
+ 'vnfPackage': {
+ 'href': '/%s/vnf_packages/%s' % (const.PKG_URL_PREFIX,
+ uuid_csarid)},
+ 'subscription': {
+ 'href': '/%s%s' % (const.VNFPKG_SUBSCRIPTION_ROOT_URI,
+ uuid_subscriptid)}
+
+ },
+ 'changeType': const.PKG_CHANGE_TYPE.PKG_DELETE,
+ 'operationalState': None,
+ "subscriptionId": uuid_subscriptid
+ }
+ NfPkgDeleteThread(csar_id=uuid_csarid, job_id="5").delete_csar()
+ mock_requests_post.assert_called_with(vnf_subscription_data["callbackUri"],
+ data=json.dumps(expect_deleted_notification),
+ headers={'Connection': 'close',
+ 'content-type': 'application/json',
+ 'accept': 'application/json'},
+ auth=HTTPBasicAuth("vnfm", "password1$"),
+ verify=False)