1 # Copyright 2017 ZTE Corporation.
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
19 from django.test import TestCase
20 from requests.auth import HTTPBasicAuth
21 from rest_framework import status
22 from rest_framework.test import APIClient
24 import catalog.pub.utils.timeutil
25 from catalog.packages import const
26 from catalog.packages.biz.sdc_vnf_package import NfDistributeThread, NfPkgDeleteThread
27 from catalog.pub.config.config import CATALOG_ROOT_PATH
28 from catalog.pub.database.models import JobStatusModel, JobModel
29 from catalog.pub.database.models import VnfPackageModel
30 from catalog.pub.msapi import sdc
31 from catalog.pub.utils import restcall, toscaparser
32 from .const import vnfd_data
35 class TestNfPackage(TestCase):
37 self.client = APIClient()
38 VnfPackageModel.objects.filter().delete()
39 JobModel.objects.filter().delete()
40 JobStatusModel.objects.filter().delete()
41 self.vnfd_data = vnfd_data
46 def assert_job_result(self, job_id, job_progress, job_detail):
47 jobs = JobStatusModel.objects.filter(
49 progress=job_progress,
51 self.assertEqual(1, len(jobs))
53 @mock.patch.object(NfDistributeThread, 'run')
54 def test_nf_pkg_distribute_normal(self, mock_run):
55 resp = self.client.post(
56 "/api/catalog/v1/vnfpackages",
63 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
65 def test_nf_pkg_distribute_when_csar_already_exist(self):
68 vnfdId="vcpe_vfw_zte_1_0"
76 self.assert_job_result("2", 255, "NF CSAR(1) already exists.")
78 @mock.patch.object(restcall, 'call_req')
79 @mock.patch.object(sdc, 'download_artifacts')
80 @mock.patch.object(toscaparser, 'parse_vnfd')
81 def test_nf_pkg_distribute_when_vnfd_already_exist(self,
83 mock_download_artifacts,
85 mock_parse_vnfd.return_value = json.JSONEncoder().encode(self.vnfd_data)
86 mock_download_artifacts.return_value = "/home/hss.csar"
87 mock_call_req.return_value = [0, json.JSONEncoder().encode([{
89 "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/hss.csar"
91 VnfPackageModel(vnfPackageId="2", vnfdId="00342b18-a5c7-11e8-998c-bf1755941f12").save()
98 self.assert_job_result("2", 255, "VNF package(00342b18-a5c7-11e8-998c-bf1755941f12) already exists.")
100 @mock.patch.object(restcall, 'call_req')
101 @mock.patch.object(sdc, 'download_artifacts')
102 @mock.patch.object(toscaparser, 'parse_vnfd')
103 def test_nf_pkg_distribute_successfully(self,
105 mock_download_artifacts,
107 mock_parse_vnfd.return_value = json.JSONEncoder().encode(self.vnfd_data)
108 mock_download_artifacts.return_value = "/home/hss.csar"
109 mock_call_req.return_value = [0, json.JSONEncoder().encode([{
111 "toscaModelURL": "https://127.0.0.1:1234/sdc/v1/hss.csar"
119 self.assert_job_result("4", 100, "CSAR(1) distribute successfully.")
121 ###############################################################################################################
123 @mock.patch.object(NfPkgDeleteThread, 'run')
124 def test_nf_pkg_delete_normal(self, mock_run):
125 resp = self.client.delete("/api/catalog/v1/vnfpackages/1")
126 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
128 def test_nf_pkg_normal_delete(self):
131 vnfdId="vcpe_vfw_zte_1_0"
137 self.assert_job_result("2", 100, "Delete CSAR(2) successfully.")
139 def test_nf_pkg_get_all(self):
145 vnfSoftwareVersion='',
154 vnfSoftwareVersion='',
158 resp = self.client.get("/api/catalog/v1/vnfpackages")
159 self.assertEqual(resp.status_code, status.HTTP_200_OK)
169 "downloadUrl": "http://127.0.0.1:8806/static/catalog/3/",
183 "downloadUrl": "http://127.0.0.1:8806/static/catalog/4/",
190 self.assertEqual(expect_data, resp.data)
192 def test_nf_pkg_get_one(self):
198 vnfSoftwareVersion='',
203 resp = self.client.get("/api/catalog/v1/vnfpackages/4")
204 self.assertEqual(resp.status_code, status.HTTP_200_OK)
213 "downloadUrl": "http://127.0.0.1:8806/static/catalog/4/",
219 self.assertEqual(expect_data, resp.data)
221 def test_nf_pkg_get_one_failed(self):
227 vnfSoftwareVersion='',
232 resp = self.client.get("/api/catalog/v1/vnfpackages/2")
233 self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
234 self.assertEqual({'error': 'Vnf package[2] not Found.'}, resp.data)
236 ###############################################################################################################
238 @mock.patch.object(toscaparser, 'parse_vnfd')
239 def test_vnfd_parse_normal(self, mock_parse_vnfd):
244 mock_parse_vnfd.return_value = json.JSONEncoder().encode({"c": "d"})
249 resp = self.client.post(
250 "/api/parser/v1/parservnfd",
254 self.assertEqual(resp.status_code, status.HTTP_202_ACCEPTED)
255 self.assertEqual({"model": '{"c": "d"}'}, resp.data)
257 def test_vnfd_parse_when_csar_not_exist(self):
258 req_data = {"csarId": "1", "inputs": []}
259 resp = self.client.post(
260 "/api/parser/v1/parservnfd",
264 self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
265 self.assertEqual(resp.data, {"error": "VNF CSAR(1) does not exist."})
267 @mock.patch("requests.post")
268 @mock.patch.object(sdc, 'download_artifacts')
269 @mock.patch.object(sdc, 'get_artifact')
270 @mock.patch("requests.get")
271 @mock.patch("uuid.uuid4")
272 @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
273 def test_service_pkg_distribute_and_notify(self, mock_nowtime, mock_uuid, mock_requests_get, mock_get_artifact,
274 mock_download_artifacts, mock_requests_post):
275 mock_nowtime.return_value = "2019-02-16 14:41:16"
276 uuid_subscriptid = "99442b18-a5c7-11e8-998c-bf1755941f13"
278 mock_uuid.side_effect = [uuid_subscriptid, "1111", "2222"]
279 mock_requests_get.return_value.status_code = 204
280 vnf_subscription_data = {
282 "notificationTypes": [
283 "VnfPackageOnboardingNotification",
284 "VnfPackageChangeNotification"
286 "vnfPkgId": [uuid_csarid],
287 "operationalState": ["ENABLED", "DISABLED"]
289 "callbackUri": "https://so-vnfm-simulator.onap:9093/vnfpkgm/v1/notification",
296 "password": "password1$"
300 response = self.client.post(
301 "/api/vnfpkgm/v1/subscriptions",
302 data=vnf_subscription_data,
304 self.assertEqual(201, response.status_code)
305 mock_download_artifacts.return_value = os.path.join(CATALOG_ROOT_PATH, "Sol004VnfCSAR.csar")
306 mock_get_artifact.return_value = {
307 "uuid": "c94490a0-f7ef-48be-b3f8-8d8662a37236",
308 "invariantUUID": "63eaec39-ffbe-411c-a838-448f2c73f7eb",
309 "name": "Sol004VnfCSAR",
311 "toscaModelURL": "/sdc/v1/catalog/resources/c94490a0-f7ef-48be-b3f8-8d8662a37236/toscaModel",
313 "subCategory": "VolteVF",
314 "resourceType": "VF",
315 "lifecycleState": "CERTIFIED",
316 "lastUpdaterUserId": "jh0003"
318 NfDistributeThread(csar_id=uuid_csarid, vim_ids=["1"], lab_vim_id="", job_id="4").on_distribute()
319 print(VnfPackageModel.objects.all().values())
320 expect_onboarding_notification = {
322 'notificationType': const.PKG_NOTIFICATION_TYPE.ONBOARDING,
323 'timeStamp': "2019-02-16 14:41:16",
325 'vnfdId': "b1bb0ce7-2222-4fa7-95ed-4840d70a1177",
328 'href': '/%s/vnf_packages/%s' % (const.PKG_URL_PREFIX, uuid_csarid)},
330 'href': '/%s%s' % (const.VNFPKG_SUBSCRIPTION_ROOT_URI, uuid_subscriptid)}
333 "subscriptionId": uuid_subscriptid
335 mock_requests_post.return_value.status_code = 204
336 mock_requests_post.assert_called_with(vnf_subscription_data["callbackUri"],
337 data=json.dumps(expect_onboarding_notification),
338 headers={'Connection': 'close',
339 'content-type': 'application/json',
340 'accept': 'application/json'},
341 auth=HTTPBasicAuth("vnfm", "password1$"),
343 mock_requests_post.return_value.status_code = 204
344 expect_deleted_notification = {
346 'notificationType': const.PKG_NOTIFICATION_TYPE.CHANGE,
347 'timeStamp': "2019-02-16 14:41:16",
349 'vnfdId': "b1bb0ce7-2222-4fa7-95ed-4840d70a1177",
352 'href': '/%s/vnf_packages/%s' % (const.PKG_URL_PREFIX,
355 'href': '/%s%s' % (const.VNFPKG_SUBSCRIPTION_ROOT_URI,
359 'changeType': const.PKG_CHANGE_TYPE.PKG_DELETE,
360 'operationalState': None,
361 "subscriptionId": uuid_subscriptid
363 NfPkgDeleteThread(csar_id=uuid_csarid, job_id="5").delete_csar()
364 mock_requests_post.assert_called_with(vnf_subscription_data["callbackUri"],
365 data=json.dumps(expect_deleted_notification),
366 headers={'Connection': 'close',
367 'content-type': 'application/json',
368 'accept': 'application/json'},
369 auth=HTTPBasicAuth("vnfm", "password1$"),