1 # Copyright (C) 2019 Verizon. All Rights Reserved
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 rest_framework.test import APIClient
21 from rest_framework import status
22 from requests.auth import HTTPBasicAuth
24 from catalog.packages.biz.nsdm_subscription import NsdmSubscription
25 from catalog.pub.database.models import NsdmSubscriptionModel
26 from catalog.packages.biz.notificationsutil import NsdNotifications, PnfNotifications
27 from catalog.packages import const
28 from catalog.pub.config import config as pub_config
29 import catalog.pub.utils.timeutil
30 from catalog.packages.tests.const import nsd_data
31 from catalog.pub.database.models import NSPackageModel, VnfPackageModel, PnfPackageModel
32 from catalog.pub.config.config import CATALOG_ROOT_PATH, MSB_SERVICE_IP, MSB_SERVICE_PORT
33 from catalog.pub.utils import toscaparser
36 class TestNsdmSubscription(TestCase):
38 self.client = APIClient()
39 NsdmSubscriptionModel.objects.all().delete()
40 self.subscription_id = str(uuid.uuid4())
42 "callbackUri": "http://callbackuri.com",
44 "authType": ["BASIC"],
46 "userName": "username",
47 "password": "password"
51 "nsdId": ["b632bddc-abcd-4180-bd8d-4e8a9578eff7"],
56 "href": "/api/v1/subscriptions/" + self.subscription_id
59 self.test_subscription = {
60 "callbackUri": "http://callbackuri.com",
61 "id": self.subscription_id,
63 "notificationTypes": [
64 "NsdOnBoardingNotification"
72 "nestedNsdInfoIds": [],
73 "nsdOnboardingState": [],
74 "nsdOperationalState": [],
81 "pnfdInvariantId": [],
82 "pnfdOnboardingState": [],
91 @mock.patch("requests.get")
92 @mock.patch.object(uuid, 'uuid4')
93 def test_nsdm_subscribe_notification(self, mock_uuid4, mock_requests):
94 temp_uuid = str(uuid.uuid4())
95 mock_requests.return_value.status_code = 204
96 mock_requests.get.return_value.status_code = 204
97 mock_uuid4.return_value = temp_uuid
98 response = self.client.post("/api/nsd/v1/subscriptions",
99 data=self.subscription, format='json')
100 self.assertEqual(201, response.status_code)
101 self.assertEqual(self.subscription["callbackUri"],
102 response.data["callbackUri"])
103 self.assertEqual(temp_uuid, response.data["id"])
105 @mock.patch("requests.get")
106 @mock.patch.object(uuid, 'uuid4')
107 def test_nsdm_subscribe_callbackFailure(self, mock_uuid4, mock_requests):
108 temp_uuid = str(uuid.uuid4())
109 mock_requests.return_value.status_code = 500
110 mock_requests.get.return_value.status_code = 500
111 mock_uuid4.return_value = temp_uuid
114 'detail': "callbackUri http://callbackuri.com didn't"
115 " return 204 statuscode."
117 response = self.client.post("/api/nsd/v1/subscriptions",
118 data=self.subscription, format='json')
119 self.assertEqual(500, response.status_code)
120 self.assertEqual(expected_data, response.data)
122 @mock.patch("requests.get")
123 def test_nsdm_second_subscription(self, mock_requests):
124 mock_requests.return_value.status_code = 204
125 mock_requests.get.return_value.status_code = 204
126 response = self.client.post("/api/nsd/v1/subscriptions",
127 data=self.subscription, format='json')
128 self.assertEqual(201, response.status_code)
129 self.assertEqual(self.subscription["callbackUri"],
130 response.data["callbackUri"])
131 dummy_subscription = {
132 "callbackUri": "http://callbackuri.com",
134 "authType": ["BASIC"],
136 "userName": "username",
137 "password": "password"
141 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
144 response = self.client.post("/api/nsd/v1/subscriptions",
145 data=dummy_subscription, format='json')
146 self.assertEqual(201, response.status_code)
147 self.assertEqual(dummy_subscription["callbackUri"],
148 response.data["callbackUri"])
150 @mock.patch("requests.get")
151 def test_nsdm_duplicate_filter_or_callbackuri_subscription(self, mock_requests):
152 mock_requests.return_value.status_code = 204
153 mock_requests.get.return_value.status_code = 204
154 response = self.client.post("/api/nsd/v1/subscriptions",
155 data=self.subscription, format='json')
156 self.assertEqual(201, response.status_code)
157 self.assertEqual(self.subscription["callbackUri"],
158 response.data["callbackUri"])
159 newsub1 = self.subscription
160 newsub1["callbackUri"] = "http://newcallbackuri.com"
161 response = self.client.post("/api/nsd/v1/subscriptions",
162 data=newsub1, format='json')
163 self.assertEqual(201, response.status_code)
164 newsub2 = self.subscription
165 newsub2["filter"]["nsdName"] = ["aaa"]
166 response = self.client.post("/api/nsd/v1/subscriptions",
167 data=newsub2, format='json')
168 self.assertEqual(201, response.status_code)
170 @mock.patch("requests.get")
171 def test_nsdm_duplicate_callbackuri_filter_subscription(self, mock_requests):
172 mock_requests.return_value.status_code = 204
173 mock_requests.get.return_value.status_code = 204
174 response = self.client.post("/api/nsd/v1/subscriptions",
175 data=self.subscription, format='json')
176 self.assertEqual(201, response.status_code)
177 self.assertEqual(self.subscription["callbackUri"],
178 response.data["callbackUri"])
179 subscriptionid = response.data["id"]
180 response = self.client.post("/api/nsd/v1/subscriptions",
181 data=self.subscription, format='json')
182 self.assertEqual(303, response.status_code)
183 redirect_addr = "https://%s:%s/%s" % (MSB_SERVICE_IP, MSB_SERVICE_PORT,
184 os.path.join(const.NSDM_SUBSCRIPTION_ROOT_URI, subscriptionid))
185 self.assertEqual(redirect_addr, response["Location"])
187 @mock.patch("requests.get")
188 def test_nsdm_bad_request(self, mock_requests):
189 dummy_subscription = {
190 "callbackUri": "http://callbackuri.com",
192 "authType": ["BASIC"],
194 "userName": "username",
195 "password": "password"
199 "nsdId": "b632bddc-bccd-4180-bd8d-4e8a9578eff7",
202 response = self.client.post("/api/nsd/v1/subscriptions",
203 data=dummy_subscription, format='json')
204 self.assertEqual(400, response.status_code)
206 @mock.patch("requests.get")
207 def test_nsdm_invalid_authtype_subscription(self, mock_requests):
208 dummy_subscription = {
209 "callbackUri": "http://callbackuri.com",
211 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
213 "userName": "username",
214 "password": "password"
218 mock_requests.return_value.status_code = 204
219 mock_requests.get.return_value.status_code = 204
222 'detail': 'Auth type should be BASIC'
224 response = self.client.post("/api/nsd/v1/subscriptions",
225 data=dummy_subscription, format='json')
226 self.assertEqual(400, response.status_code)
227 self.assertEqual(expected_data, response.data)
229 @mock.patch("requests.get")
230 def test_nsdm_invalid_authtype_oauthclient_subscription(
231 self, mock_requests):
232 dummy_subscription = {
233 "callbackUri": "http://callbackuri.com",
235 "authType": ["BASIC"],
236 "paramsOauth2ClientCredentials": {
237 "clientId": "clientId",
238 "clientPassword": "password",
239 "tokenEndpoint": "http://tokenEndpoint"
243 mock_requests.return_value.status_code = 204
244 mock_requests.get.return_value.status_code = 204
247 'detail': 'Auth type should be OAUTH2_CLIENT_CREDENTIALS'
249 response = self.client.post("/api/nsd/v1/subscriptions",
250 data=dummy_subscription, format='json')
251 self.assertEqual(400, response.status_code)
252 self.assertEqual(expected_data, response.data)
254 @mock.patch("requests.get")
255 def test_nsdm_invalid_authparams_subscription(self, mock_requests):
256 dummy_subscription = {
257 "callbackUri": "http://callbackuri.com",
259 "authType": ["BASIC"],
261 "userName": "username"
265 mock_requests.return_value.status_code = 204
266 mock_requests.get.return_value.status_code = 204
269 'detail': 'userName and password needed for BASIC'
271 response = self.client.post("/api/nsd/v1/subscriptions",
272 data=dummy_subscription, format='json')
273 self.assertEqual(400, response.status_code)
274 self.assertEqual(expected_data, response.data)
276 @mock.patch("requests.get")
277 def test_nsdm_invalid_authparams_oauthclient_subscription(
278 self, mock_requests):
279 dummy_subscription = {
280 "callbackUri": "http://callbackuri.com",
282 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
283 "paramsOauth2ClientCredentials": {
284 "clientPassword": "password",
285 "tokenEndpoint": "http://tokenEndpoint"
289 mock_requests.return_value.status_code = 204
290 mock_requests.get.return_value.status_code = 204
293 'detail': 'clientId, clientPassword and tokenEndpoint'
294 ' required for OAUTH2_CLIENT_CREDENTIALS'
296 response = self.client.post("/api/nsd/v1/subscriptions",
297 data=dummy_subscription, format='json')
298 self.assertEqual(400, response.status_code)
299 self.assertEqual(expected_data, response.data)
301 @mock.patch("requests.get")
302 def test_nsdm_invalid_filter_subscription(self, mock_requests):
303 dummy_subscription = {
304 "callbackUri": "http://callbackuri.com",
306 "authType": ["BASIC"],
308 "userName": "username",
309 "password": "password"
313 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
314 "nsdInfoId": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
317 mock_requests.return_value.status_code = 204
318 mock_requests.get.return_value.status_code = 204
321 'detail': 'Notification Filter should contain'
322 ' either nsdId or nsdInfoId'
324 response = self.client.post("/api/nsd/v1/subscriptions",
325 data=dummy_subscription, format='json')
326 self.assertEqual(400, response.status_code)
327 self.assertEqual(expected_data, response.data)
329 @mock.patch("requests.get")
330 def test_nsdm_invalid_filter_pnfd_subscription(self, mock_requests):
331 dummy_subscription = {
332 "callbackUri": "http://callbackuri.com",
334 "authType": ["BASIC"],
336 "userName": "username",
337 "password": "password"
341 "pnfdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
342 "pnfdInfoIds": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
345 mock_requests.return_value.status_code = 204
346 mock_requests.get.return_value.status_code = 204
349 'detail': 'Notification Filter should contain'
350 ' either pnfdId or pnfdInfoIds'
352 response = self.client.post("/api/nsd/v1/subscriptions",
353 data=dummy_subscription, format='json')
354 self.assertEqual(400, response.status_code)
355 self.assertEqual(expected_data, response.data)
357 @mock.patch.object(NsdmSubscription, 'create')
358 def test_nsdmsubscription_create_when_catch_exception(self, mock_create):
359 mock_create.side_effect = TypeError("Unicode type")
360 response = self.client.post('/api/nsd/v1/subscriptions',
361 data=self.subscription, format='json')
362 self.assertEqual(response.status_code,
363 status.HTTP_500_INTERNAL_SERVER_ERROR)
365 def test_nsdm_get_subscriptions(self):
366 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
367 callback_uri="http://callbackuri.com",
369 notificationTypes=json.dumps(
370 ["NsdOnBoardingNotification"]),
371 nsdId=[], nsdVersion=[],
372 nsdInfoId=[], nsdDesigner=[],
373 nsdName=[], nsdInvariantId=[],
374 vnfPkgIds=[], pnfdInfoIds=[],
375 nestedNsdInfoIds=[], nsdOnboardingState=[],
376 nsdOperationalState=[], nsdUsageState=[],
377 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
378 pnfdName=[], pnfdInvariantId=[],
379 pnfdOnboardingState=[], pnfdUsageState=[],
380 links=json.dumps(self.links)).save()
381 response = self.client.get("/api/nsd/v1/subscriptions",
383 self.assertEqual(status.HTTP_200_OK, response.status_code)
384 self.assertEqual([self.test_subscription], response.data)
386 def test_nsdm_get_subscriptions_filter(self):
387 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
388 callback_uri="http://callbackuri.com",
390 notificationTypes=json.dumps(
391 ["NsdOnBoardingNotification"]),
392 nsdId=[], nsdVersion=[],
393 nsdInfoId=[], nsdDesigner=[],
394 nsdName=[], nsdInvariantId=[],
395 vnfPkgIds=[], pnfdInfoIds=[],
396 nestedNsdInfoIds=[], nsdOnboardingState=[],
397 nsdOperationalState=[], nsdUsageState=[],
398 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
399 pnfdName=[], pnfdInvariantId=[],
400 pnfdOnboardingState=[], pnfdUsageState=[],
401 links=json.dumps(self.links)).save()
402 response = self.client.get("/api/nsd/v1/subscriptions"
404 "=NsdOnBoardingNotification",
406 self.assertEqual(status.HTTP_200_OK, response.status_code)
407 self.assertEqual([self.test_subscription], response.data)
409 def test_nsdm_get_subscriptions_filter_failure(self):
410 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
411 callback_uri="http://callbackuri.com",
413 notificationTypes=json.dumps(
414 ["NsdOnBoardingNotification"]),
415 nsdId=[], nsdVersion=[],
416 nsdInfoId=[], nsdDesigner=[],
417 nsdName=[], nsdInvariantId=[],
418 vnfPkgIds=[], pnfdInfoIds=[],
419 nestedNsdInfoIds=[], nsdOnboardingState=[],
420 nsdOperationalState=[], nsdUsageState=[],
421 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
422 pnfdName=[], pnfdInvariantId=[],
423 pnfdOnboardingState=[], pnfdUsageState=[],
424 links=json.dumps(self.links)).save()
425 response = self.client.get("/api/nsd/v1/subscriptions"
426 "?notificationTypes="
427 "PnfdOnBoardingFailureNotification",
429 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
431 def test_nsdm_get_subscriptions_invalid_filter(self):
432 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
433 callback_uri="http://callbackuri.com",
435 notificationTypes=json.dumps(
436 ["NsdOnBoardingNotification"]),
437 nsdId=[], nsdVersion=[],
438 nsdInfoId=[], nsdDesigner=[],
439 nsdName=[], nsdInvariantId=[],
440 vnfPkgIds=[], pnfdInfoIds=[],
441 nestedNsdInfoIds=[], nsdOnboardingState=[],
442 nsdOperationalState=[], nsdUsageState=[],
443 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
444 pnfdName=[], pnfdInvariantId=[],
445 pnfdOnboardingState=[], pnfdUsageState=[],
446 links=json.dumps(self.links)).save()
447 response = self.client.get("/api/nsd/v1/subscriptions"
448 "?notificationTypes="
449 "PnfdOnBoardingFailureNotificati",
451 self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
453 @mock.patch.object(NsdmSubscription, 'query_multi_subscriptions')
454 def test_nsdmsubscription_get_when_catch_exception(self, mock_create):
455 mock_create.side_effect = TypeError("Unicode type")
456 response = self.client.get('/api/nsd/v1/subscriptions', format='json')
457 self.assertEqual(response.status_code,
458 status.HTTP_500_INTERNAL_SERVER_ERROR)
460 def test_nsdm_get_subscription(self):
461 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
462 callback_uri="http://callbackuri.com",
464 notificationTypes=json.dumps(
465 ["NsdOnBoardingNotification"]),
466 nsdId=[], nsdVersion=[],
467 nsdInfoId=[], nsdDesigner=[],
468 nsdName=[], nsdInvariantId=[],
469 vnfPkgIds=[], pnfdInfoIds=[],
470 nestedNsdInfoIds=[], nsdOnboardingState=[],
471 nsdOperationalState=[], nsdUsageState=[],
472 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
473 pnfdName=[], pnfdInvariantId=[],
474 pnfdOnboardingState=[], pnfdUsageState=[],
475 links=json.dumps(self.links)).save()
476 response = self.client.get('/api/nsd/v1/'
477 'subscriptions/' + self.subscription_id,
479 self.assertEqual(status.HTTP_200_OK, response.status_code)
480 self.assertEqual(self.test_subscription, response.data)
482 def test_nsdm_get_subscription_failure(self):
485 "detail": "Subscription(" + self.subscription_id + ") "
488 response = self.client.get('/api/nsd/v1/'
489 'subscriptions/' + self.subscription_id,
491 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
492 self.assertEqual(expected_data, response.data)
494 def test_nsdm_get_subscription_failure_bad_request(self):
495 response = self.client.get("/api/nsd/v1/subscriptions/123",
497 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
499 @mock.patch.object(NsdmSubscription, 'query_single_subscription')
500 def test_nsdmsubscription_getsingle_when_catch_exception(
502 mock_create.side_effect = TypeError("Unicode type")
503 response = self.client.get('/api/nsd/v1/'
504 'subscriptions/' + self.subscription_id,
506 self.assertEqual(response.status_code,
507 status.HTTP_500_INTERNAL_SERVER_ERROR)
509 def test_ndsm_delete_subscription(self):
510 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
511 callback_uri="http://callbackuri.com",
513 notificationTypes=json.dumps(
514 ["NsdOnBoardingNotification"]),
515 nsdId=[], nsdVersion=[],
516 nsdInfoId=[], nsdDesigner=[],
517 nsdName=[], nsdInvariantId=[],
518 vnfPkgIds=[], pnfdInfoIds=[],
519 nestedNsdInfoIds=[], nsdOnboardingState=[],
520 nsdOperationalState=[], nsdUsageState=[],
521 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
522 pnfdName=[], pnfdInvariantId=[],
523 pnfdOnboardingState=[], pnfdUsageState=[],
524 links=json.dumps(self.links)).save()
525 response = self.client.delete('/api/nsd/v1/'
526 'subscriptions/' + self.subscription_id,
528 self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code)
530 def test_ndsm_delete_subscription_failure(self):
531 response = self.client.delete('/api/nsd/v1/'
532 'subscriptions/' + self.subscription_id,
534 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
536 def test_nsdm_delete_subscription_failure_bad_request(self):
537 response = self.client.delete("/api/nsd/v1/subscriptions/123",
539 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
541 @mock.patch.object(NsdmSubscription, 'delete_single_subscription')
542 def test_nsdmsubscription_delete_when_catch_exception(self, mock_create):
543 mock_create.side_effect = TypeError("Unicode type")
544 response = self.client.delete('/api/nsd/v1/'
545 'subscriptions/' + self.subscription_id,
547 self.assertEqual(response.status_code,
548 status.HTTP_500_INTERNAL_SERVER_ERROR)
550 @mock.patch("requests.post")
551 @mock.patch.object(toscaparser, 'parse_nsd')
552 @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
553 @mock.patch("requests.get")
554 @mock.patch.object(uuid, 'uuid4')
555 def test_nsdm_subscribe_trigger_notification(self, mock_uuid4, mock_requests, mock_nowtime, mock_parse_nsd,
557 mock_requests.return_value.status_code = 204
558 mock_requests.get.return_value.status_code = 204
559 mock_uuid4.return_value = "1111"
560 mock_nowtime.return_value = "nowtime()"
563 "callbackUri": "http://callbackuri.com",
565 "authType": ["BASIC"],
567 "userName": "username",
568 "password": "password"
572 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"]
575 response = self.client.post("/api/nsd/v1/subscriptions",
576 data=subscription_req, format='json')
577 self.assertEqual(201, response.status_code)
579 self.user_defined_data = {
584 user_defined_data_json = json.JSONEncoder().encode(self.user_defined_data)
585 mock_parse_nsd.return_value = json.JSONEncoder().encode(nsd_data)
588 vnfdId="vcpe_vfw_zte_1_0"
597 nsPackageId='d0ea5ec3-0b98-438a-9bea-488230cff174',
598 operationalState='DISABLED',
599 usageState='NOT_IN_USE',
600 userDefinedData=user_defined_data_json,
603 with open('nsd_content.txt', 'wt') as fp:
605 with open('nsd_content.txt', 'rt') as fp:
606 resp = self.client.put(
607 "/api/nsd/v1/ns_descriptors/d0ea5ec3-0b98-438a-9bea-488230cff174/nsd_content",
611 with open(os.path.join(CATALOG_ROOT_PATH, 'd0ea5ec3-0b98-438a-9bea-488230cff174/nsd_content.txt')) as fp:
613 file_content = '%s%s' % (file_content, data)
614 ns_pkg = NSPackageModel.objects.filter(nsPackageId="d0ea5ec3-0b98-438a-9bea-488230cff174")
615 self.assertEqual("b632bddc-bccd-4180-bd8d-4e8a9578eff7", ns_pkg[0].nsdId)
616 self.assertEqual(const.PKG_STATUS.ONBOARDED, ns_pkg[0].onboardingState)
617 self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
618 self.assertEqual(None, resp.data)
619 self.assertEqual(file_content, 'test')
620 os.remove('nsd_content.txt')
621 expect_callbackuri = "http://callbackuri.com"
622 expect_notification = {
624 'notificationType': const.NSD_NOTIFICATION_TYPE.NSD_ONBOARDING,
625 'timeStamp': "nowtime()",
626 'nsdInfoId': "d0ea5ec3-0b98-438a-9bea-488230cff174",
627 'nsdId': "b632bddc-bccd-4180-bd8d-4e8a9578eff7",
628 "subscriptionId": "1111",
631 'href': 'http://%s:%s/%s%s' % (pub_config.MSB_SERVICE_IP,
632 pub_config.MSB_SERVICE_PORT,
633 const.NSDM_SUBSCRIPTION_ROOT_URI,
636 'href': 'http://%s:%s/%s/ns_descriptors/%s' % (pub_config.MSB_SERVICE_IP,
637 pub_config.MSB_SERVICE_PORT,
638 const.NSD_URL_PREFIX,
639 "d0ea5ec3-0b98-438a-9bea-488230cff174")
643 mock_requests_post.assert_called_with(expect_callbackuri, data=expect_notification,
644 auth=HTTPBasicAuth("username", "password"),
645 headers={'Connection': 'close',
646 'content-type': 'application/json',
647 'accept': 'application/json'},
651 class NotificationTest(TestCase):
653 NsdmSubscriptionModel(subscriptionid="1",
654 callback_uri="http://127.0.0.1/self",
655 notificationTypes=const.NOTIFICATION_TYPES,
657 nsdInfoId="nsdinfoid1",
658 pnfdInfoIds="pnfdInfoIds1",
663 NsdmSubscriptionModel.objects.all().delete()
665 @mock.patch("requests.post")
666 @mock.patch("uuid.uuid4")
667 @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
668 def test_nsdpkg_notify(self, mock_nowtime, mock_uuid, mock_requests_post):
669 mock_nowtime.return_value = "nowtime()"
670 mock_uuid.return_value = "1111"
671 notify = NsdNotifications(const.NSD_NOTIFICATION_TYPE.NSD_ONBOARDING_FAILURE,
672 nsd_info_id="nsdinfoid1",
674 failure_details="NSD(nsdid1) already exists.", operational_state=None)
675 notify.send_notification()
676 expect_callbackuri = "http://127.0.0.1/self"
677 expect_notification = {
679 'notificationType': const.NSD_NOTIFICATION_TYPE.NSD_ONBOARDING_FAILURE,
680 'timeStamp': "nowtime()",
681 'nsdInfoId': "nsdinfoid1",
683 'onboardingFailureDetails': "NSD(nsdid1) already exists.",
684 "subscriptionId": "1",
687 'href': 'http://%s:%s/%s%s' % (pub_config.MSB_SERVICE_IP,
688 pub_config.MSB_SERVICE_PORT,
689 const.NSDM_SUBSCRIPTION_ROOT_URI,
692 'href': 'http://%s:%s/%s/ns_descriptors/%s' % (pub_config.MSB_SERVICE_IP,
693 pub_config.MSB_SERVICE_PORT,
694 const.NSD_URL_PREFIX,
699 mock_requests_post.assert_called_with(expect_callbackuri,
700 data=expect_notification,
701 headers={'Connection': 'close',
702 'content-type': 'application/json',
703 'accept': 'application/json'},
707 @mock.patch("requests.post")
708 @mock.patch("uuid.uuid4")
709 @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
710 def test_pnfpkg_notify(self, mock_nowtime, mock_uuid, mock_requests_post):
711 mock_nowtime.return_value = "nowtime()"
712 mock_uuid.return_value = "1111"
713 notify = PnfNotifications(const.NSD_NOTIFICATION_TYPE.PNFD_ONBOARDING,
714 pnfd_info_id="pnfdInfoIds1",
716 failure_details=None)
717 notify.send_notification()
718 expect_callbackuri = "http://127.0.0.1/self"
719 expect_notification = {
721 'notificationType': const.NSD_NOTIFICATION_TYPE.PNFD_ONBOARDING,
722 'timeStamp': "nowtime()",
723 'pnfdInfoIds': "pnfdInfoIds1",
725 "subscriptionId": "1",
728 'href': 'http://%s:%s/%s%s' % (pub_config.MSB_SERVICE_IP,
729 pub_config.MSB_SERVICE_PORT,
730 const.NSDM_SUBSCRIPTION_ROOT_URI,
733 'href': 'http://%s:%s/%s/pnf_descriptors/%s' % (pub_config.MSB_SERVICE_IP,
734 pub_config.MSB_SERVICE_PORT,
735 const.NSD_URL_PREFIX,
740 mock_requests_post.assert_called_with(expect_callbackuri, data=expect_notification,
741 headers={'Connection': 'close',
742 'content-type': 'application/json',
743 'accept': 'application/json'},