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
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_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"])
161 'detail': 'Subscription has already existed with'
162 ' the same callbackUri and filter'
164 response = self.client.post("/api/nsd/v1/subscriptions",
165 data=self.subscription, format='json')
166 self.assertEqual(303, response.status_code)
167 self.assertEqual(expected_data, response.data)
169 @mock.patch("requests.get")
170 def test_nsdm_bad_request(self, mock_requests):
171 dummy_subscription = {
172 "callbackUri": "http://callbackuri.com",
174 "authType": ["BASIC"],
176 "userName": "username",
177 "password": "password"
181 "nsdId": "b632bddc-bccd-4180-bd8d-4e8a9578eff7",
184 response = self.client.post("/api/nsd/v1/subscriptions",
185 data=dummy_subscription, format='json')
186 self.assertEqual(400, response.status_code)
188 @mock.patch("requests.get")
189 def test_nsdm_invalid_authtype_subscription(self, mock_requests):
190 dummy_subscription = {
191 "callbackUri": "http://callbackuri.com",
193 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
195 "userName": "username",
196 "password": "password"
200 mock_requests.return_value.status_code = 204
201 mock_requests.get.return_value.status_code = 204
204 'detail': 'Auth type should be BASIC'
206 response = self.client.post("/api/nsd/v1/subscriptions",
207 data=dummy_subscription, format='json')
208 self.assertEqual(400, response.status_code)
209 self.assertEqual(expected_data, response.data)
211 @mock.patch("requests.get")
212 def test_nsdm_invalid_authtype_oauthclient_subscription(
213 self, mock_requests):
214 dummy_subscription = {
215 "callbackUri": "http://callbackuri.com",
217 "authType": ["BASIC"],
218 "paramsOauth2ClientCredentials": {
219 "clientId": "clientId",
220 "clientPassword": "password",
221 "tokenEndpoint": "http://tokenEndpoint"
225 mock_requests.return_value.status_code = 204
226 mock_requests.get.return_value.status_code = 204
229 'detail': 'Auth type should be OAUTH2_CLIENT_CREDENTIALS'
231 response = self.client.post("/api/nsd/v1/subscriptions",
232 data=dummy_subscription, format='json')
233 self.assertEqual(400, response.status_code)
234 self.assertEqual(expected_data, response.data)
236 @mock.patch("requests.get")
237 def test_nsdm_invalid_authparams_subscription(self, mock_requests):
238 dummy_subscription = {
239 "callbackUri": "http://callbackuri.com",
241 "authType": ["BASIC"],
243 "userName": "username"
247 mock_requests.return_value.status_code = 204
248 mock_requests.get.return_value.status_code = 204
251 'detail': 'userName and password needed for BASIC'
253 response = self.client.post("/api/nsd/v1/subscriptions",
254 data=dummy_subscription, format='json')
255 self.assertEqual(400, response.status_code)
256 self.assertEqual(expected_data, response.data)
258 @mock.patch("requests.get")
259 def test_nsdm_invalid_authparams_oauthclient_subscription(
260 self, mock_requests):
261 dummy_subscription = {
262 "callbackUri": "http://callbackuri.com",
264 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
265 "paramsOauth2ClientCredentials": {
266 "clientPassword": "password",
267 "tokenEndpoint": "http://tokenEndpoint"
271 mock_requests.return_value.status_code = 204
272 mock_requests.get.return_value.status_code = 204
275 'detail': 'clientId, clientPassword and tokenEndpoint'
276 ' required for OAUTH2_CLIENT_CREDENTIALS'
278 response = self.client.post("/api/nsd/v1/subscriptions",
279 data=dummy_subscription, format='json')
280 self.assertEqual(400, response.status_code)
281 self.assertEqual(expected_data, response.data)
283 @mock.patch("requests.get")
284 def test_nsdm_invalid_filter_subscription(self, mock_requests):
285 dummy_subscription = {
286 "callbackUri": "http://callbackuri.com",
288 "authType": ["BASIC"],
290 "userName": "username",
291 "password": "password"
295 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
296 "nsdInfoId": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
299 mock_requests.return_value.status_code = 204
300 mock_requests.get.return_value.status_code = 204
303 'detail': 'Notification Filter should contain'
304 ' either nsdId or nsdInfoId'
306 response = self.client.post("/api/nsd/v1/subscriptions",
307 data=dummy_subscription, format='json')
308 self.assertEqual(400, response.status_code)
309 self.assertEqual(expected_data, response.data)
311 @mock.patch("requests.get")
312 def test_nsdm_invalid_filter_pnfd_subscription(self, mock_requests):
313 dummy_subscription = {
314 "callbackUri": "http://callbackuri.com",
316 "authType": ["BASIC"],
318 "userName": "username",
319 "password": "password"
323 "pnfdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
324 "pnfdInfoIds": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
327 mock_requests.return_value.status_code = 204
328 mock_requests.get.return_value.status_code = 204
331 'detail': 'Notification Filter should contain'
332 ' either pnfdId or pnfdInfoIds'
334 response = self.client.post("/api/nsd/v1/subscriptions",
335 data=dummy_subscription, format='json')
336 self.assertEqual(400, response.status_code)
337 self.assertEqual(expected_data, response.data)
339 @mock.patch.object(NsdmSubscription, 'create')
340 def test_nsdmsubscription_create_when_catch_exception(self, mock_create):
341 mock_create.side_effect = TypeError("Unicode type")
342 response = self.client.post('/api/nsd/v1/subscriptions',
343 data=self.subscription, format='json')
344 self.assertEqual(response.status_code,
345 status.HTTP_500_INTERNAL_SERVER_ERROR)
347 def test_nsdm_get_subscriptions(self):
348 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
349 callback_uri="http://callbackuri.com",
351 notificationTypes=json.dumps(
352 ["NsdOnBoardingNotification"]),
353 nsdId=[], nsdVersion=[],
354 nsdInfoId=[], nsdDesigner=[],
355 nsdName=[], nsdInvariantId=[],
356 vnfPkgIds=[], pnfdInfoIds=[],
357 nestedNsdInfoIds=[], nsdOnboardingState=[],
358 nsdOperationalState=[], nsdUsageState=[],
359 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
360 pnfdName=[], pnfdInvariantId=[],
361 pnfdOnboardingState=[], pnfdUsageState=[],
362 links=json.dumps(self.links)).save()
363 response = self.client.get("/api/nsd/v1/subscriptions",
365 self.assertEqual(status.HTTP_200_OK, response.status_code)
366 self.assertEqual([self.test_subscription], response.data)
368 def test_nsdm_get_subscriptions_filter(self):
369 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
370 callback_uri="http://callbackuri.com",
372 notificationTypes=json.dumps(
373 ["NsdOnBoardingNotification"]),
374 nsdId=[], nsdVersion=[],
375 nsdInfoId=[], nsdDesigner=[],
376 nsdName=[], nsdInvariantId=[],
377 vnfPkgIds=[], pnfdInfoIds=[],
378 nestedNsdInfoIds=[], nsdOnboardingState=[],
379 nsdOperationalState=[], nsdUsageState=[],
380 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
381 pnfdName=[], pnfdInvariantId=[],
382 pnfdOnboardingState=[], pnfdUsageState=[],
383 links=json.dumps(self.links)).save()
384 response = self.client.get("/api/nsd/v1/subscriptions"
386 "=NsdOnBoardingNotification",
388 self.assertEqual(status.HTTP_200_OK, response.status_code)
389 self.assertEqual([self.test_subscription], response.data)
391 def test_nsdm_get_subscriptions_filter_failure(self):
392 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
393 callback_uri="http://callbackuri.com",
395 notificationTypes=json.dumps(
396 ["NsdOnBoardingNotification"]),
397 nsdId=[], nsdVersion=[],
398 nsdInfoId=[], nsdDesigner=[],
399 nsdName=[], nsdInvariantId=[],
400 vnfPkgIds=[], pnfdInfoIds=[],
401 nestedNsdInfoIds=[], nsdOnboardingState=[],
402 nsdOperationalState=[], nsdUsageState=[],
403 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
404 pnfdName=[], pnfdInvariantId=[],
405 pnfdOnboardingState=[], pnfdUsageState=[],
406 links=json.dumps(self.links)).save()
407 response = self.client.get("/api/nsd/v1/subscriptions"
408 "?notificationTypes="
409 "PnfdOnBoardingFailureNotification",
411 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
413 def test_nsdm_get_subscriptions_invalid_filter(self):
414 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
415 callback_uri="http://callbackuri.com",
417 notificationTypes=json.dumps(
418 ["NsdOnBoardingNotification"]),
419 nsdId=[], nsdVersion=[],
420 nsdInfoId=[], nsdDesigner=[],
421 nsdName=[], nsdInvariantId=[],
422 vnfPkgIds=[], pnfdInfoIds=[],
423 nestedNsdInfoIds=[], nsdOnboardingState=[],
424 nsdOperationalState=[], nsdUsageState=[],
425 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
426 pnfdName=[], pnfdInvariantId=[],
427 pnfdOnboardingState=[], pnfdUsageState=[],
428 links=json.dumps(self.links)).save()
429 response = self.client.get("/api/nsd/v1/subscriptions"
430 "?notificationTypes="
431 "PnfdOnBoardingFailureNotificati",
433 self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
435 @mock.patch.object(NsdmSubscription, 'query_multi_subscriptions')
436 def test_nsdmsubscription_get_when_catch_exception(self, mock_create):
437 mock_create.side_effect = TypeError("Unicode type")
438 response = self.client.get('/api/nsd/v1/subscriptions', format='json')
439 self.assertEqual(response.status_code,
440 status.HTTP_500_INTERNAL_SERVER_ERROR)
442 def test_nsdm_get_subscription(self):
443 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
444 callback_uri="http://callbackuri.com",
446 notificationTypes=json.dumps(
447 ["NsdOnBoardingNotification"]),
448 nsdId=[], nsdVersion=[],
449 nsdInfoId=[], nsdDesigner=[],
450 nsdName=[], nsdInvariantId=[],
451 vnfPkgIds=[], pnfdInfoIds=[],
452 nestedNsdInfoIds=[], nsdOnboardingState=[],
453 nsdOperationalState=[], nsdUsageState=[],
454 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
455 pnfdName=[], pnfdInvariantId=[],
456 pnfdOnboardingState=[], pnfdUsageState=[],
457 links=json.dumps(self.links)).save()
458 response = self.client.get('/api/nsd/v1/'
459 'subscriptions/' + self.subscription_id,
461 self.assertEqual(status.HTTP_200_OK, response.status_code)
462 self.assertEqual(self.test_subscription, response.data)
464 def test_nsdm_get_subscription_failure(self):
467 "detail": "Subscription(" + self.subscription_id + ") "
470 response = self.client.get('/api/nsd/v1/'
471 'subscriptions/' + self.subscription_id,
473 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
474 self.assertEqual(expected_data, response.data)
476 def test_nsdm_get_subscription_failure_bad_request(self):
477 response = self.client.get("/api/nsd/v1/subscriptions/123",
479 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
481 @mock.patch.object(NsdmSubscription, 'query_single_subscription')
482 def test_nsdmsubscription_getsingle_when_catch_exception(
484 mock_create.side_effect = TypeError("Unicode type")
485 response = self.client.get('/api/nsd/v1/'
486 'subscriptions/' + self.subscription_id,
488 self.assertEqual(response.status_code,
489 status.HTTP_500_INTERNAL_SERVER_ERROR)
491 def test_ndsm_delete_subscription(self):
492 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
493 callback_uri="http://callbackuri.com",
495 notificationTypes=json.dumps(
496 ["NsdOnBoardingNotification"]),
497 nsdId=[], nsdVersion=[],
498 nsdInfoId=[], nsdDesigner=[],
499 nsdName=[], nsdInvariantId=[],
500 vnfPkgIds=[], pnfdInfoIds=[],
501 nestedNsdInfoIds=[], nsdOnboardingState=[],
502 nsdOperationalState=[], nsdUsageState=[],
503 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
504 pnfdName=[], pnfdInvariantId=[],
505 pnfdOnboardingState=[], pnfdUsageState=[],
506 links=json.dumps(self.links)).save()
507 response = self.client.delete('/api/nsd/v1/'
508 'subscriptions/' + self.subscription_id,
510 self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code)
512 def test_ndsm_delete_subscription_failure(self):
513 response = self.client.delete('/api/nsd/v1/'
514 'subscriptions/' + self.subscription_id,
516 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
518 def test_nsdm_delete_subscription_failure_bad_request(self):
519 response = self.client.delete("/api/nsd/v1/subscriptions/123",
521 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
523 @mock.patch.object(NsdmSubscription, 'delete_single_subscription')
524 def test_nsdmsubscription_delete_when_catch_exception(self, mock_create):
525 mock_create.side_effect = TypeError("Unicode type")
526 response = self.client.delete('/api/nsd/v1/'
527 'subscriptions/' + self.subscription_id,
529 self.assertEqual(response.status_code,
530 status.HTTP_500_INTERNAL_SERVER_ERROR)
532 @mock.patch("requests.post")
533 @mock.patch.object(toscaparser, 'parse_nsd')
534 @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
535 @mock.patch("requests.get")
536 @mock.patch.object(uuid, 'uuid4')
537 def test_nsdm_subscribe_trigger_notification(self, mock_uuid4, mock_requests, mock_nowtime, mock_parse_nsd,
539 mock_requests.return_value.status_code = 204
540 mock_requests.get.return_value.status_code = 204
541 mock_uuid4.return_value = "1111"
542 mock_nowtime.return_value = "nowtime()"
545 "callbackUri": "http://callbackuri.com",
547 "authType": ["BASIC"],
549 "userName": "username",
550 "password": "password"
554 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"]
557 response = self.client.post("/api/nsd/v1/subscriptions",
558 data=subscription_req, format='json')
559 self.assertEqual(201, response.status_code)
561 self.user_defined_data = {
566 user_defined_data_json = json.JSONEncoder().encode(self.user_defined_data)
567 mock_parse_nsd.return_value = json.JSONEncoder().encode(nsd_data)
570 vnfdId="vcpe_vfw_zte_1_0"
579 nsPackageId='d0ea5ec3-0b98-438a-9bea-488230cff174',
580 operationalState='DISABLED',
581 usageState='NOT_IN_USE',
582 userDefinedData=user_defined_data_json,
585 with open('nsd_content.txt', 'wt') as fp:
587 with open('nsd_content.txt', 'rt') as fp:
588 resp = self.client.put(
589 "/api/nsd/v1/ns_descriptors/d0ea5ec3-0b98-438a-9bea-488230cff174/nsd_content",
593 with open(os.path.join(CATALOG_ROOT_PATH, 'd0ea5ec3-0b98-438a-9bea-488230cff174/nsd_content.txt')) as fp:
595 file_content = '%s%s' % (file_content, data)
596 ns_pkg = NSPackageModel.objects.filter(nsPackageId="d0ea5ec3-0b98-438a-9bea-488230cff174")
597 self.assertEqual("b632bddc-bccd-4180-bd8d-4e8a9578eff7", ns_pkg[0].nsdId)
598 self.assertEqual(const.PKG_STATUS.ONBOARDED, ns_pkg[0].onboardingState)
599 self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
600 self.assertEqual(None, resp.data)
601 self.assertEqual(file_content, 'test')
602 os.remove('nsd_content.txt')
603 expect_callbackuri = "http://callbackuri.com"
604 expect_notification = {
606 'notificationType': const.NSD_NOTIFICATION_TYPE.NSD_ONBOARDING,
607 'timeStamp': "nowtime()",
608 'nsdInfoId': "d0ea5ec3-0b98-438a-9bea-488230cff174",
609 'nsdId': "b632bddc-bccd-4180-bd8d-4e8a9578eff7",
610 "subscriptionId": "1111",
613 'href': 'http://%s:%s/%s%s' % (pub_config.MSB_SERVICE_IP,
614 pub_config.MSB_SERVICE_PORT,
615 const.NSDM_SUBSCRIPTION_ROOT_URI,
618 'href': 'http://%s:%s/%s/ns_descriptors/%s' % (pub_config.MSB_SERVICE_IP,
619 pub_config.MSB_SERVICE_PORT,
620 const.NSD_URL_PREFIX,
621 "d0ea5ec3-0b98-438a-9bea-488230cff174")
625 mock_requests_post.assert_called_with(expect_callbackuri, data=expect_notification,
626 auth=HTTPBasicAuth("username", "password"),
627 headers={'Connection': 'close',
628 'content-type': 'application/json',
629 'accept': 'application/json'},
633 class NotificationTest(TestCase):
635 NsdmSubscriptionModel(subscriptionid="1",
636 callback_uri="http://127.0.0.1/self",
637 notificationTypes=const.NOTIFICATION_TYPES,
639 nsdInfoId="nsdinfoid1",
640 pnfdInfoIds="pnfdInfoIds1",
645 NsdmSubscriptionModel.objects.all().delete()
647 @mock.patch("requests.post")
648 @mock.patch("uuid.uuid4")
649 @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
650 def test_nsdpkg_notify(self, mock_nowtime, mock_uuid, mock_requests_post):
651 mock_nowtime.return_value = "nowtime()"
652 mock_uuid.return_value = "1111"
653 notify = NsdNotifications(const.NSD_NOTIFICATION_TYPE.NSD_ONBOARDING_FAILURE,
654 nsd_info_id="nsdinfoid1",
656 failure_details="NSD(nsdid1) already exists.", operational_state=None)
657 notify.send_notification()
658 expect_callbackuri = "http://127.0.0.1/self"
659 expect_notification = {
661 'notificationType': const.NSD_NOTIFICATION_TYPE.NSD_ONBOARDING_FAILURE,
662 'timeStamp': "nowtime()",
663 'nsdInfoId': "nsdinfoid1",
665 'onboardingFailureDetails': "NSD(nsdid1) already exists.",
666 "subscriptionId": "1",
669 'href': 'http://%s:%s/%s%s' % (pub_config.MSB_SERVICE_IP,
670 pub_config.MSB_SERVICE_PORT,
671 const.NSDM_SUBSCRIPTION_ROOT_URI,
674 'href': 'http://%s:%s/%s/ns_descriptors/%s' % (pub_config.MSB_SERVICE_IP,
675 pub_config.MSB_SERVICE_PORT,
676 const.NSD_URL_PREFIX,
681 mock_requests_post.assert_called_with(expect_callbackuri,
682 data=expect_notification,
683 headers={'Connection': 'close',
684 'content-type': 'application/json',
685 'accept': 'application/json'},
689 @mock.patch("requests.post")
690 @mock.patch("uuid.uuid4")
691 @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
692 def test_pnfpkg_notify(self, mock_nowtime, mock_uuid, mock_requests_post):
693 mock_nowtime.return_value = "nowtime()"
694 mock_uuid.return_value = "1111"
695 notify = PnfNotifications(const.NSD_NOTIFICATION_TYPE.PNFD_ONBOARDING,
696 pnfd_info_id="pnfdInfoIds1",
698 failure_details=None)
699 notify.send_notification()
700 expect_callbackuri = "http://127.0.0.1/self"
701 expect_notification = {
703 'notificationType': const.NSD_NOTIFICATION_TYPE.PNFD_ONBOARDING,
704 'timeStamp': "nowtime()",
705 'pnfdInfoIds': "pnfdInfoIds1",
707 "subscriptionId": "1",
710 'href': 'http://%s:%s/%s%s' % (pub_config.MSB_SERVICE_IP,
711 pub_config.MSB_SERVICE_PORT,
712 const.NSDM_SUBSCRIPTION_ROOT_URI,
715 'href': 'http://%s:%s/%s/pnf_descriptors/%s' % (pub_config.MSB_SERVICE_IP,
716 pub_config.MSB_SERVICE_PORT,
717 const.NSD_URL_PREFIX,
722 mock_requests_post.assert_called_with(expect_callbackuri, data=expect_notification,
723 headers={'Connection': 'close',
724 'content-type': 'application/json',
725 'accept': 'application/json'},