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.
20 from django.test import TestCase
21 from requests.auth import HTTPBasicAuth
22 from rest_framework import status
23 from rest_framework.test import APIClient
25 import catalog.pub.utils.timeutil
26 from catalog.packages import const
27 from catalog.packages.biz.notificationsutil import NsdNotifications, PnfNotifications
28 from catalog.packages.biz.nsdm_subscription import NsdmSubscription
29 from catalog.packages.tests.const import nsd_data
30 from catalog.pub.config.config import CATALOG_ROOT_PATH
31 from catalog.pub.database.models import NSPackageModel, VnfPackageModel, PnfPackageModel
32 from catalog.pub.database.models import NsdmSubscriptionModel
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 = "/%s" % (os.path.join(const.NSDM_SUBSCRIPTION_ROOT_URI, subscriptionid))
184 self.assertEqual(redirect_addr, response["Location"])
186 @mock.patch("requests.get")
187 def test_nsdm_bad_request(self, mock_requests):
188 dummy_subscription = {
189 "callbackUri": "http://callbackuri.com",
191 "authType": ["BASIC"],
193 "userName": "username",
194 "password": "password"
198 "nsdId": "b632bddc-bccd-4180-bd8d-4e8a9578eff7",
201 response = self.client.post("/api/nsd/v1/subscriptions",
202 data=dummy_subscription, format='json')
203 self.assertEqual(400, response.status_code)
205 @mock.patch("requests.get")
206 def test_nsdm_invalid_authtype_subscription(self, mock_requests):
207 dummy_subscription = {
208 "callbackUri": "http://callbackuri.com",
210 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
212 "userName": "username",
213 "password": "password"
217 mock_requests.return_value.status_code = 204
218 mock_requests.get.return_value.status_code = 204
221 'detail': 'Auth type should be BASIC'
223 response = self.client.post("/api/nsd/v1/subscriptions",
224 data=dummy_subscription, format='json')
225 self.assertEqual(400, response.status_code)
226 self.assertEqual(expected_data, response.data)
228 @mock.patch("requests.get")
229 def test_nsdm_invalid_authtype_oauthclient_subscription(
230 self, mock_requests):
231 dummy_subscription = {
232 "callbackUri": "http://callbackuri.com",
234 "authType": ["BASIC"],
235 "paramsOauth2ClientCredentials": {
236 "clientId": "clientId",
237 "clientPassword": "password",
238 "tokenEndpoint": "http://tokenEndpoint"
242 mock_requests.return_value.status_code = 204
243 mock_requests.get.return_value.status_code = 204
246 'detail': 'Auth type should be OAUTH2_CLIENT_CREDENTIALS'
248 response = self.client.post("/api/nsd/v1/subscriptions",
249 data=dummy_subscription, format='json')
250 self.assertEqual(400, response.status_code)
251 self.assertEqual(expected_data, response.data)
253 @mock.patch("requests.get")
254 def test_nsdm_invalid_authparams_subscription(self, mock_requests):
255 dummy_subscription = {
256 "callbackUri": "http://callbackuri.com",
258 "authType": ["BASIC"],
260 "userName": "username"
264 mock_requests.return_value.status_code = 204
265 mock_requests.get.return_value.status_code = 204
268 'detail': 'userName and password needed for BASIC'
270 response = self.client.post("/api/nsd/v1/subscriptions",
271 data=dummy_subscription, format='json')
272 self.assertEqual(400, response.status_code)
273 self.assertEqual(expected_data, response.data)
275 @mock.patch("requests.get")
276 def test_nsdm_invalid_authparams_oauthclient_subscription(
277 self, mock_requests):
278 dummy_subscription = {
279 "callbackUri": "http://callbackuri.com",
281 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
282 "paramsOauth2ClientCredentials": {
283 "clientPassword": "password",
284 "tokenEndpoint": "http://tokenEndpoint"
288 mock_requests.return_value.status_code = 204
289 mock_requests.get.return_value.status_code = 204
292 'detail': 'clientId, clientPassword and tokenEndpoint'
293 ' required for OAUTH2_CLIENT_CREDENTIALS'
295 response = self.client.post("/api/nsd/v1/subscriptions",
296 data=dummy_subscription, format='json')
297 self.assertEqual(400, response.status_code)
298 self.assertEqual(expected_data, response.data)
300 @mock.patch("requests.get")
301 def test_nsdm_invalid_filter_subscription(self, mock_requests):
302 dummy_subscription = {
303 "callbackUri": "http://callbackuri.com",
305 "authType": ["BASIC"],
307 "userName": "username",
308 "password": "password"
312 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
313 "nsdInfoId": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
316 mock_requests.return_value.status_code = 204
317 mock_requests.get.return_value.status_code = 204
320 'detail': 'Notification Filter should contain'
321 ' either nsdId or nsdInfoId'
323 response = self.client.post("/api/nsd/v1/subscriptions",
324 data=dummy_subscription, format='json')
325 self.assertEqual(400, response.status_code)
326 self.assertEqual(expected_data, response.data)
328 @mock.patch("requests.get")
329 def test_nsdm_invalid_filter_pnfd_subscription(self, mock_requests):
330 dummy_subscription = {
331 "callbackUri": "http://callbackuri.com",
333 "authType": ["BASIC"],
335 "userName": "username",
336 "password": "password"
340 "pnfdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
341 "pnfdInfoIds": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
344 mock_requests.return_value.status_code = 204
345 mock_requests.get.return_value.status_code = 204
348 'detail': 'Notification Filter should contain'
349 ' either pnfdId or pnfdInfoIds'
351 response = self.client.post("/api/nsd/v1/subscriptions",
352 data=dummy_subscription, format='json')
353 self.assertEqual(400, response.status_code)
354 self.assertEqual(expected_data, response.data)
356 @mock.patch.object(NsdmSubscription, 'create')
357 def test_nsdmsubscription_create_when_catch_exception(self, mock_create):
358 mock_create.side_effect = TypeError("Unicode type")
359 response = self.client.post('/api/nsd/v1/subscriptions',
360 data=self.subscription, format='json')
361 self.assertEqual(response.status_code,
362 status.HTTP_500_INTERNAL_SERVER_ERROR)
364 def test_nsdm_get_subscriptions(self):
365 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
366 callback_uri="http://callbackuri.com",
368 notificationTypes=json.dumps(
369 ["NsdOnBoardingNotification"]),
370 nsdId=[], nsdVersion=[],
371 nsdInfoId=[], nsdDesigner=[],
372 nsdName=[], nsdInvariantId=[],
373 vnfPkgIds=[], pnfdInfoIds=[],
374 nestedNsdInfoIds=[], nsdOnboardingState=[],
375 nsdOperationalState=[], nsdUsageState=[],
376 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
377 pnfdName=[], pnfdInvariantId=[],
378 pnfdOnboardingState=[], pnfdUsageState=[],
379 links=json.dumps(self.links)).save()
380 response = self.client.get("/api/nsd/v1/subscriptions",
382 self.assertEqual(status.HTTP_200_OK, response.status_code)
383 self.assertEqual([self.test_subscription], response.data)
385 def test_nsdm_get_subscriptions_filter(self):
386 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
387 callback_uri="http://callbackuri.com",
389 notificationTypes=json.dumps(
390 ["NsdOnBoardingNotification"]),
391 nsdId=[], nsdVersion=[],
392 nsdInfoId=[], nsdDesigner=[],
393 nsdName=[], nsdInvariantId=[],
394 vnfPkgIds=[], pnfdInfoIds=[],
395 nestedNsdInfoIds=[], nsdOnboardingState=[],
396 nsdOperationalState=[], nsdUsageState=[],
397 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
398 pnfdName=[], pnfdInvariantId=[],
399 pnfdOnboardingState=[], pnfdUsageState=[],
400 links=json.dumps(self.links)).save()
401 response = self.client.get("/api/nsd/v1/subscriptions"
403 "=NsdOnBoardingNotification",
405 self.assertEqual(status.HTTP_200_OK, response.status_code)
406 self.assertEqual([self.test_subscription], response.data)
408 def test_nsdm_get_subscriptions_filter_failure(self):
409 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
410 callback_uri="http://callbackuri.com",
412 notificationTypes=json.dumps(
413 ["NsdOnBoardingNotification"]),
414 nsdId=[], nsdVersion=[],
415 nsdInfoId=[], nsdDesigner=[],
416 nsdName=[], nsdInvariantId=[],
417 vnfPkgIds=[], pnfdInfoIds=[],
418 nestedNsdInfoIds=[], nsdOnboardingState=[],
419 nsdOperationalState=[], nsdUsageState=[],
420 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
421 pnfdName=[], pnfdInvariantId=[],
422 pnfdOnboardingState=[], pnfdUsageState=[],
423 links=json.dumps(self.links)).save()
424 response = self.client.get("/api/nsd/v1/subscriptions"
425 "?notificationTypes="
426 "PnfdOnBoardingFailureNotification",
428 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
430 def test_nsdm_get_subscriptions_invalid_filter(self):
431 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
432 callback_uri="http://callbackuri.com",
434 notificationTypes=json.dumps(
435 ["NsdOnBoardingNotification"]),
436 nsdId=[], nsdVersion=[],
437 nsdInfoId=[], nsdDesigner=[],
438 nsdName=[], nsdInvariantId=[],
439 vnfPkgIds=[], pnfdInfoIds=[],
440 nestedNsdInfoIds=[], nsdOnboardingState=[],
441 nsdOperationalState=[], nsdUsageState=[],
442 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
443 pnfdName=[], pnfdInvariantId=[],
444 pnfdOnboardingState=[], pnfdUsageState=[],
445 links=json.dumps(self.links)).save()
446 response = self.client.get("/api/nsd/v1/subscriptions"
447 "?notificationTypes="
448 "PnfdOnBoardingFailureNotificati",
450 self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
452 @mock.patch.object(NsdmSubscription, 'query_multi_subscriptions')
453 def test_nsdmsubscription_get_when_catch_exception(self, mock_create):
454 mock_create.side_effect = TypeError("Unicode type")
455 response = self.client.get('/api/nsd/v1/subscriptions', format='json')
456 self.assertEqual(response.status_code,
457 status.HTTP_500_INTERNAL_SERVER_ERROR)
459 def test_nsdm_get_subscription(self):
460 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
461 callback_uri="http://callbackuri.com",
463 notificationTypes=json.dumps(
464 ["NsdOnBoardingNotification"]),
465 nsdId=[], nsdVersion=[],
466 nsdInfoId=[], nsdDesigner=[],
467 nsdName=[], nsdInvariantId=[],
468 vnfPkgIds=[], pnfdInfoIds=[],
469 nestedNsdInfoIds=[], nsdOnboardingState=[],
470 nsdOperationalState=[], nsdUsageState=[],
471 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
472 pnfdName=[], pnfdInvariantId=[],
473 pnfdOnboardingState=[], pnfdUsageState=[],
474 links=json.dumps(self.links)).save()
475 response = self.client.get('/api/nsd/v1/'
476 'subscriptions/' + self.subscription_id,
478 self.assertEqual(status.HTTP_200_OK, response.status_code)
479 self.assertEqual(self.test_subscription, response.data)
481 def test_nsdm_get_subscription_failure(self):
484 "detail": "Subscription(" + self.subscription_id + ") "
487 response = self.client.get('/api/nsd/v1/'
488 'subscriptions/' + self.subscription_id,
490 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
491 self.assertEqual(expected_data, response.data)
493 def test_nsdm_get_subscription_failure_bad_request(self):
494 response = self.client.get("/api/nsd/v1/subscriptions/123",
496 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
498 @mock.patch.object(NsdmSubscription, 'query_single_subscription')
499 def test_nsdmsubscription_getsingle_when_catch_exception(
501 mock_create.side_effect = TypeError("Unicode type")
502 response = self.client.get('/api/nsd/v1/'
503 'subscriptions/' + self.subscription_id,
505 self.assertEqual(response.status_code,
506 status.HTTP_500_INTERNAL_SERVER_ERROR)
508 def test_ndsm_delete_subscription(self):
509 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
510 callback_uri="http://callbackuri.com",
512 notificationTypes=json.dumps(
513 ["NsdOnBoardingNotification"]),
514 nsdId=[], nsdVersion=[],
515 nsdInfoId=[], nsdDesigner=[],
516 nsdName=[], nsdInvariantId=[],
517 vnfPkgIds=[], pnfdInfoIds=[],
518 nestedNsdInfoIds=[], nsdOnboardingState=[],
519 nsdOperationalState=[], nsdUsageState=[],
520 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
521 pnfdName=[], pnfdInvariantId=[],
522 pnfdOnboardingState=[], pnfdUsageState=[],
523 links=json.dumps(self.links)).save()
524 response = self.client.delete('/api/nsd/v1/'
525 'subscriptions/' + self.subscription_id,
527 self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code)
529 def test_ndsm_delete_subscription_failure(self):
530 response = self.client.delete('/api/nsd/v1/'
531 'subscriptions/' + self.subscription_id,
533 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
535 def test_nsdm_delete_subscription_failure_bad_request(self):
536 response = self.client.delete("/api/nsd/v1/subscriptions/123",
538 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
540 @mock.patch.object(NsdmSubscription, 'delete_single_subscription')
541 def test_nsdmsubscription_delete_when_catch_exception(self, mock_create):
542 mock_create.side_effect = TypeError("Unicode type")
543 response = self.client.delete('/api/nsd/v1/'
544 'subscriptions/' + self.subscription_id,
546 self.assertEqual(response.status_code,
547 status.HTTP_500_INTERNAL_SERVER_ERROR)
549 @mock.patch("requests.post")
550 @mock.patch.object(toscaparser, 'parse_nsd')
551 @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
552 @mock.patch("requests.get")
553 @mock.patch.object(uuid, 'uuid4')
554 def test_nsdm_subscribe_trigger_notification(self, mock_uuid4, mock_requests, mock_nowtime, mock_parse_nsd,
556 mock_requests.return_value.status_code = 204
557 mock_requests.get.return_value.status_code = 204
558 mock_uuid4.return_value = "1111"
559 mock_nowtime.return_value = "nowtime()"
562 "callbackUri": "http://callbackuri.com",
564 "authType": ["BASIC"],
566 "userName": "username",
567 "password": "password"
571 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"]
574 response = self.client.post("/api/nsd/v1/subscriptions",
575 data=subscription_req, format='json')
576 self.assertEqual(201, response.status_code)
578 self.user_defined_data = {
583 user_defined_data_json = json.JSONEncoder().encode(self.user_defined_data)
584 mock_parse_nsd.return_value = json.JSONEncoder().encode(nsd_data)
587 vnfdId="vcpe_vfw_zte_1_0"
596 nsPackageId='d0ea5ec3-0b98-438a-9bea-488230cff174',
597 operationalState='DISABLED',
598 usageState='NOT_IN_USE',
599 userDefinedData=user_defined_data_json,
602 with open('nsd_content.txt', 'wt') as fp:
604 with open('nsd_content.txt', 'rt') as fp:
605 resp = self.client.put(
606 "/api/nsd/v1/ns_descriptors/d0ea5ec3-0b98-438a-9bea-488230cff174/nsd_content",
610 with open(os.path.join(CATALOG_ROOT_PATH, 'd0ea5ec3-0b98-438a-9bea-488230cff174/nsd_content.txt')) as fp:
612 file_content = '%s%s' % (file_content, data)
613 ns_pkg = NSPackageModel.objects.filter(nsPackageId="d0ea5ec3-0b98-438a-9bea-488230cff174")
614 self.assertEqual("b632bddc-bccd-4180-bd8d-4e8a9578eff7", ns_pkg[0].nsdId)
615 self.assertEqual(const.PKG_STATUS.ONBOARDED, ns_pkg[0].onboardingState)
616 self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
617 self.assertEqual(None, resp.data)
618 self.assertEqual(file_content, 'test')
619 os.remove('nsd_content.txt')
620 expect_callbackuri = "http://callbackuri.com"
621 expect_notification = {
623 'notificationType': const.NSD_NOTIFICATION_TYPE.NSD_ONBOARDING,
624 'timeStamp': "nowtime()",
625 'nsdInfoId': "d0ea5ec3-0b98-438a-9bea-488230cff174",
626 'nsdId': "b632bddc-bccd-4180-bd8d-4e8a9578eff7",
629 'href': '/%s/ns_descriptors/%s' % (
630 const.NSD_URL_PREFIX, "d0ea5ec3-0b98-438a-9bea-488230cff174")},
632 'href': '/%s%s' % (const.NSDM_SUBSCRIPTION_ROOT_URI, "1111")}
635 "subscriptionId": "1111"
637 mock_requests_post.assert_called_with(expect_callbackuri, data=json.dumps(expect_notification),
638 auth=HTTPBasicAuth("username", "password"),
639 headers={'Connection': 'close',
640 'content-type': 'application/json',
641 'accept': 'application/json'},
645 class NotificationTest(TestCase):
647 NsdmSubscriptionModel(subscriptionid="1",
648 callback_uri="http://127.0.0.1/self",
649 notificationTypes=const.NOTIFICATION_TYPES,
651 nsdInfoId="nsdinfoid1",
652 pnfdInfoIds="pnfdInfoIds1",
657 NsdmSubscriptionModel.objects.all().delete()
659 @mock.patch("requests.post")
660 @mock.patch("uuid.uuid4")
661 @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
662 def test_nsdpkg_notify(self, mock_nowtime, mock_uuid, mock_requests_post):
663 mock_nowtime.return_value = "nowtime()"
664 mock_uuid.return_value = "1111"
665 notify = NsdNotifications(const.NSD_NOTIFICATION_TYPE.NSD_ONBOARDING_FAILURE,
666 nsd_info_id="nsdinfoid1",
668 failure_details="NSD(nsdid1) already exists.", operational_state=None)
669 notify.send_notification()
670 expect_callbackuri = "http://127.0.0.1/self"
671 expect_notification = {
673 'notificationType': const.NSD_NOTIFICATION_TYPE.NSD_ONBOARDING_FAILURE,
674 'timeStamp': "nowtime()",
675 'nsdInfoId': "nsdinfoid1",
679 'href': '/%s/ns_descriptors/%s' % (const.NSD_URL_PREFIX, "nsdinfoid1")
682 'href': '/%s%s' % (const.NSDM_SUBSCRIPTION_ROOT_URI, "1")}
684 'onboardingFailureDetails': "NSD(nsdid1) already exists.",
685 "subscriptionId": "1"
687 mock_requests_post.assert_called_with(expect_callbackuri,
688 data=json.dumps(expect_notification),
689 headers={'Connection': 'close',
690 'content-type': 'application/json',
691 'accept': 'application/json'},
695 @mock.patch("requests.post")
696 @mock.patch("uuid.uuid4")
697 @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
698 def test_pnfpkg_notify(self, mock_nowtime, mock_uuid, mock_requests_post):
699 mock_nowtime.return_value = "nowtime()"
700 mock_uuid.return_value = "1111"
701 notify = PnfNotifications(const.NSD_NOTIFICATION_TYPE.PNFD_ONBOARDING,
702 pnfd_info_id="pnfdInfoIds1",
704 failure_details=None)
705 notify.send_notification()
706 expect_callbackuri = "http://127.0.0.1/self"
707 expect_notification = {
709 'notificationType': const.NSD_NOTIFICATION_TYPE.PNFD_ONBOARDING,
710 'timeStamp': "nowtime()",
711 'pnfdInfoIds': "pnfdInfoIds1",
715 'href': '/%s/pnf_descriptors/%s' % (const.NSD_URL_PREFIX, "pnfdInfoIds1")
718 'href': '/%s%s' % (const.NSDM_SUBSCRIPTION_ROOT_URI, "1")},
720 "subscriptionId": "1",
722 mock_requests_post.assert_called_with(expect_callbackuri, data=json.dumps(expect_notification),
723 headers={'Connection': 'close',
724 'content-type': 'application/json',
725 'accept': 'application/json'},