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):
39 self.client = APIClient()
40 NsdmSubscriptionModel.objects.all().delete()
41 self.subscription_id = str(uuid.uuid4())
43 "callbackUri": "http://callbackuri.com",
45 "authType": ["BASIC"],
47 "userName": "username",
48 "password": "password"
52 "nsdId": ["b632bddc-abcd-4180-bd8d-4e8a9578eff7"],
57 "href": "/api/v1/subscriptions/" + self.subscription_id
60 self.test_subscription = {
61 "callbackUri": "http://callbackuri.com",
62 "id": self.subscription_id,
64 "notificationTypes": [
65 "NsdOnBoardingNotification"
73 "nestedNsdInfoIds": [],
74 "nsdOnboardingState": [],
75 "nsdOperationalState": [],
82 "pnfdInvariantId": [],
83 "pnfdOnboardingState": [],
92 @mock.patch("requests.get")
93 @mock.patch.object(uuid, 'uuid4')
94 def test_nsdm_subscribe_notification(self, mock_uuid4, mock_requests):
95 temp_uuid = str(uuid.uuid4())
96 mock_requests.return_value.status_code = 204
97 mock_requests.get.return_value.status_code = 204
98 mock_uuid4.return_value = temp_uuid
99 response = self.client.post("/api/nsd/v1/subscriptions",
100 data=self.subscription, format='json')
101 self.assertEqual(201, response.status_code)
102 self.assertEqual(self.subscription["callbackUri"],
103 response.data["callbackUri"])
104 self.assertEqual(temp_uuid, response.data["id"])
106 @mock.patch("requests.get")
107 @mock.patch.object(uuid, 'uuid4')
108 def test_nsdm_subscribe_callbackFailure(self, mock_uuid4, mock_requests):
109 temp_uuid = str(uuid.uuid4())
110 mock_requests.return_value.status_code = 500
111 mock_requests.get.return_value.status_code = 500
112 mock_uuid4.return_value = temp_uuid
115 'detail': "callbackUri http://callbackuri.com didn't"
116 " return 204 statuscode."
118 response = self.client.post("/api/nsd/v1/subscriptions",
119 data=self.subscription, format='json')
120 self.assertEqual(500, response.status_code)
121 self.assertEqual(expected_data, response.data)
123 @mock.patch("requests.get")
124 def test_nsdm_second_subscription(self, mock_requests):
125 mock_requests.return_value.status_code = 204
126 mock_requests.get.return_value.status_code = 204
127 response = self.client.post("/api/nsd/v1/subscriptions",
128 data=self.subscription, format='json')
129 self.assertEqual(201, response.status_code)
130 self.assertEqual(self.subscription["callbackUri"],
131 response.data["callbackUri"])
132 dummy_subscription = {
133 "callbackUri": "http://callbackuri.com",
135 "authType": ["BASIC"],
137 "userName": "username",
138 "password": "password"
142 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
145 response = self.client.post("/api/nsd/v1/subscriptions",
146 data=dummy_subscription, format='json')
147 self.assertEqual(201, response.status_code)
148 self.assertEqual(dummy_subscription["callbackUri"],
149 response.data["callbackUri"])
151 @mock.patch("requests.get")
152 def test_nsdm_duplicate_subscription(self, mock_requests):
153 mock_requests.return_value.status_code = 204
154 mock_requests.get.return_value.status_code = 204
155 response = self.client.post("/api/nsd/v1/subscriptions",
156 data=self.subscription, format='json')
157 self.assertEqual(201, response.status_code)
158 self.assertEqual(self.subscription["callbackUri"],
159 response.data["callbackUri"])
162 'detail': 'Subscription has already existed with'
163 ' the same callbackUri and filter'
165 response = self.client.post("/api/nsd/v1/subscriptions",
166 data=self.subscription, format='json')
167 self.assertEqual(303, response.status_code)
168 self.assertEqual(expected_data, response.data)
170 @mock.patch("requests.get")
171 def test_nsdm_bad_request(self, mock_requests):
172 dummy_subscription = {
173 "callbackUri": "http://callbackuri.com",
175 "authType": ["BASIC"],
177 "userName": "username",
178 "password": "password"
182 "nsdId": "b632bddc-bccd-4180-bd8d-4e8a9578eff7",
185 response = self.client.post("/api/nsd/v1/subscriptions",
186 data=dummy_subscription, format='json')
187 self.assertEqual(400, response.status_code)
189 @mock.patch("requests.get")
190 def test_nsdm_invalid_authtype_subscription(self, mock_requests):
191 dummy_subscription = {
192 "callbackUri": "http://callbackuri.com",
194 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
196 "userName": "username",
197 "password": "password"
201 mock_requests.return_value.status_code = 204
202 mock_requests.get.return_value.status_code = 204
205 'detail': 'Auth type should be BASIC'
207 response = self.client.post("/api/nsd/v1/subscriptions",
208 data=dummy_subscription, format='json')
209 self.assertEqual(400, response.status_code)
210 self.assertEqual(expected_data, response.data)
212 @mock.patch("requests.get")
213 def test_nsdm_invalid_authtype_oauthclient_subscription(
214 self, mock_requests):
215 dummy_subscription = {
216 "callbackUri": "http://callbackuri.com",
218 "authType": ["BASIC"],
219 "paramsOauth2ClientCredentials": {
220 "clientId": "clientId",
221 "clientPassword": "password",
222 "tokenEndpoint": "http://tokenEndpoint"
226 mock_requests.return_value.status_code = 204
227 mock_requests.get.return_value.status_code = 204
230 'detail': 'Auth type should be OAUTH2_CLIENT_CREDENTIALS'
232 response = self.client.post("/api/nsd/v1/subscriptions",
233 data=dummy_subscription, format='json')
234 self.assertEqual(400, response.status_code)
235 self.assertEqual(expected_data, response.data)
237 @mock.patch("requests.get")
238 def test_nsdm_invalid_authparams_subscription(self, mock_requests):
239 dummy_subscription = {
240 "callbackUri": "http://callbackuri.com",
242 "authType": ["BASIC"],
244 "userName": "username"
248 mock_requests.return_value.status_code = 204
249 mock_requests.get.return_value.status_code = 204
252 'detail': 'userName and password needed for BASIC'
254 response = self.client.post("/api/nsd/v1/subscriptions",
255 data=dummy_subscription, format='json')
256 self.assertEqual(400, response.status_code)
257 self.assertEqual(expected_data, response.data)
259 @mock.patch("requests.get")
260 def test_nsdm_invalid_authparams_oauthclient_subscription(
261 self, mock_requests):
262 dummy_subscription = {
263 "callbackUri": "http://callbackuri.com",
265 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
266 "paramsOauth2ClientCredentials": {
267 "clientPassword": "password",
268 "tokenEndpoint": "http://tokenEndpoint"
272 mock_requests.return_value.status_code = 204
273 mock_requests.get.return_value.status_code = 204
276 'detail': 'clientId, clientPassword and tokenEndpoint'
277 ' required for OAUTH2_CLIENT_CREDENTIALS'
279 response = self.client.post("/api/nsd/v1/subscriptions",
280 data=dummy_subscription, format='json')
281 self.assertEqual(400, response.status_code)
282 self.assertEqual(expected_data, response.data)
284 @mock.patch("requests.get")
285 def test_nsdm_invalid_filter_subscription(self, mock_requests):
286 dummy_subscription = {
287 "callbackUri": "http://callbackuri.com",
289 "authType": ["BASIC"],
291 "userName": "username",
292 "password": "password"
296 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
297 "nsdInfoId": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
300 mock_requests.return_value.status_code = 204
301 mock_requests.get.return_value.status_code = 204
304 'detail': 'Notification Filter should contain'
305 ' either nsdId or nsdInfoId'
307 response = self.client.post("/api/nsd/v1/subscriptions",
308 data=dummy_subscription, format='json')
309 self.assertEqual(400, response.status_code)
310 self.assertEqual(expected_data, response.data)
312 @mock.patch("requests.get")
313 def test_nsdm_invalid_filter_pnfd_subscription(self, mock_requests):
314 dummy_subscription = {
315 "callbackUri": "http://callbackuri.com",
317 "authType": ["BASIC"],
319 "userName": "username",
320 "password": "password"
324 "pnfdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
325 "pnfdInfoIds": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
328 mock_requests.return_value.status_code = 204
329 mock_requests.get.return_value.status_code = 204
332 'detail': 'Notification Filter should contain'
333 ' either pnfdId or pnfdInfoIds'
335 response = self.client.post("/api/nsd/v1/subscriptions",
336 data=dummy_subscription, format='json')
337 self.assertEqual(400, response.status_code)
338 self.assertEqual(expected_data, response.data)
340 @mock.patch.object(NsdmSubscription, 'create')
341 def test_nsdmsubscription_create_when_catch_exception(self, mock_create):
342 mock_create.side_effect = TypeError("Unicode type")
343 response = self.client.post('/api/nsd/v1/subscriptions',
344 data=self.subscription, format='json')
345 self.assertEqual(response.status_code,
346 status.HTTP_500_INTERNAL_SERVER_ERROR)
348 def test_nsdm_get_subscriptions(self):
349 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
350 callback_uri="http://callbackuri.com",
352 notificationTypes=json.dumps(
353 ["NsdOnBoardingNotification"]),
354 nsdId=[], nsdVersion=[],
355 nsdInfoId=[], nsdDesigner=[],
356 nsdName=[], nsdInvariantId=[],
357 vnfPkgIds=[], pnfdInfoIds=[],
358 nestedNsdInfoIds=[], nsdOnboardingState=[],
359 nsdOperationalState=[], nsdUsageState=[],
360 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
361 pnfdName=[], pnfdInvariantId=[],
362 pnfdOnboardingState=[], pnfdUsageState=[],
363 links=json.dumps(self.links)).save()
364 response = self.client.get("/api/nsd/v1/subscriptions",
366 self.assertEqual(status.HTTP_200_OK, response.status_code)
367 self.assertEqual([self.test_subscription], response.data)
369 def test_nsdm_get_subscriptions_filter(self):
370 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
371 callback_uri="http://callbackuri.com",
373 notificationTypes=json.dumps(
374 ["NsdOnBoardingNotification"]),
375 nsdId=[], nsdVersion=[],
376 nsdInfoId=[], nsdDesigner=[],
377 nsdName=[], nsdInvariantId=[],
378 vnfPkgIds=[], pnfdInfoIds=[],
379 nestedNsdInfoIds=[], nsdOnboardingState=[],
380 nsdOperationalState=[], nsdUsageState=[],
381 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
382 pnfdName=[], pnfdInvariantId=[],
383 pnfdOnboardingState=[], pnfdUsageState=[],
384 links=json.dumps(self.links)).save()
385 response = self.client.get("/api/nsd/v1/subscriptions"
387 "=NsdOnBoardingNotification",
389 self.assertEqual(status.HTTP_200_OK, response.status_code)
390 self.assertEqual([self.test_subscription], response.data)
392 def test_nsdm_get_subscriptions_filter_failure(self):
393 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
394 callback_uri="http://callbackuri.com",
396 notificationTypes=json.dumps(
397 ["NsdOnBoardingNotification"]),
398 nsdId=[], nsdVersion=[],
399 nsdInfoId=[], nsdDesigner=[],
400 nsdName=[], nsdInvariantId=[],
401 vnfPkgIds=[], pnfdInfoIds=[],
402 nestedNsdInfoIds=[], nsdOnboardingState=[],
403 nsdOperationalState=[], nsdUsageState=[],
404 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
405 pnfdName=[], pnfdInvariantId=[],
406 pnfdOnboardingState=[], pnfdUsageState=[],
407 links=json.dumps(self.links)).save()
408 response = self.client.get("/api/nsd/v1/subscriptions"
409 "?notificationTypes="
410 "PnfdOnBoardingFailureNotification",
412 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
414 def test_nsdm_get_subscriptions_invalid_filter(self):
415 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
416 callback_uri="http://callbackuri.com",
418 notificationTypes=json.dumps(
419 ["NsdOnBoardingNotification"]),
420 nsdId=[], nsdVersion=[],
421 nsdInfoId=[], nsdDesigner=[],
422 nsdName=[], nsdInvariantId=[],
423 vnfPkgIds=[], pnfdInfoIds=[],
424 nestedNsdInfoIds=[], nsdOnboardingState=[],
425 nsdOperationalState=[], nsdUsageState=[],
426 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
427 pnfdName=[], pnfdInvariantId=[],
428 pnfdOnboardingState=[], pnfdUsageState=[],
429 links=json.dumps(self.links)).save()
430 response = self.client.get("/api/nsd/v1/subscriptions"
431 "?notificationTypes="
432 "PnfdOnBoardingFailureNotificati",
434 self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
436 @mock.patch.object(NsdmSubscription, 'query_multi_subscriptions')
437 def test_nsdmsubscription_get_when_catch_exception(self, mock_create):
438 mock_create.side_effect = TypeError("Unicode type")
439 response = self.client.get('/api/nsd/v1/subscriptions', format='json')
440 self.assertEqual(response.status_code,
441 status.HTTP_500_INTERNAL_SERVER_ERROR)
443 def test_nsdm_get_subscription(self):
444 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
445 callback_uri="http://callbackuri.com",
447 notificationTypes=json.dumps(
448 ["NsdOnBoardingNotification"]),
449 nsdId=[], nsdVersion=[],
450 nsdInfoId=[], nsdDesigner=[],
451 nsdName=[], nsdInvariantId=[],
452 vnfPkgIds=[], pnfdInfoIds=[],
453 nestedNsdInfoIds=[], nsdOnboardingState=[],
454 nsdOperationalState=[], nsdUsageState=[],
455 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
456 pnfdName=[], pnfdInvariantId=[],
457 pnfdOnboardingState=[], pnfdUsageState=[],
458 links=json.dumps(self.links)).save()
459 response = self.client.get('/api/nsd/v1/'
460 'subscriptions/' + self.subscription_id,
462 self.assertEqual(status.HTTP_200_OK, response.status_code)
463 self.assertEqual(self.test_subscription, response.data)
465 def test_nsdm_get_subscription_failure(self):
468 "detail": "Subscription(" + self.subscription_id + ") "
471 response = self.client.get('/api/nsd/v1/'
472 'subscriptions/' + self.subscription_id,
474 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
475 self.assertEqual(expected_data, response.data)
477 def test_nsdm_get_subscription_failure_bad_request(self):
478 response = self.client.get("/api/nsd/v1/subscriptions/123",
480 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
482 @mock.patch.object(NsdmSubscription, 'query_single_subscription')
483 def test_nsdmsubscription_getsingle_when_catch_exception(
485 mock_create.side_effect = TypeError("Unicode type")
486 response = self.client.get('/api/nsd/v1/'
487 'subscriptions/' + self.subscription_id,
489 self.assertEqual(response.status_code,
490 status.HTTP_500_INTERNAL_SERVER_ERROR)
492 def test_ndsm_delete_subscription(self):
493 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
494 callback_uri="http://callbackuri.com",
496 notificationTypes=json.dumps(
497 ["NsdOnBoardingNotification"]),
498 nsdId=[], nsdVersion=[],
499 nsdInfoId=[], nsdDesigner=[],
500 nsdName=[], nsdInvariantId=[],
501 vnfPkgIds=[], pnfdInfoIds=[],
502 nestedNsdInfoIds=[], nsdOnboardingState=[],
503 nsdOperationalState=[], nsdUsageState=[],
504 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
505 pnfdName=[], pnfdInvariantId=[],
506 pnfdOnboardingState=[], pnfdUsageState=[],
507 links=json.dumps(self.links)).save()
508 response = self.client.delete('/api/nsd/v1/'
509 'subscriptions/' + self.subscription_id,
511 self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code)
513 def test_ndsm_delete_subscription_failure(self):
514 response = self.client.delete('/api/nsd/v1/'
515 'subscriptions/' + self.subscription_id,
517 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
519 def test_nsdm_delete_subscription_failure_bad_request(self):
520 response = self.client.delete("/api/nsd/v1/subscriptions/123",
522 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
524 @mock.patch.object(NsdmSubscription, 'delete_single_subscription')
525 def test_nsdmsubscription_delete_when_catch_exception(self, mock_create):
526 mock_create.side_effect = TypeError("Unicode type")
527 response = self.client.delete('/api/nsd/v1/'
528 'subscriptions/' + self.subscription_id,
530 self.assertEqual(response.status_code,
531 status.HTTP_500_INTERNAL_SERVER_ERROR)
533 @mock.patch("requests.post")
534 @mock.patch.object(toscaparser, 'parse_nsd')
535 @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
536 @mock.patch("requests.get")
537 @mock.patch.object(uuid, 'uuid4')
538 def test_nsdm_subscribe_trigger_notification(self, mock_uuid4, mock_requests, mock_nowtime, mock_parse_nsd,
540 mock_requests.return_value.status_code = 204
541 mock_requests.get.return_value.status_code = 204
542 mock_uuid4.return_value = "1111"
543 mock_nowtime.return_value = "nowtime()"
546 "callbackUri": "http://callbackuri.com",
548 "authType": ["BASIC"],
550 "userName": "username",
551 "password": "password"
555 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"]
558 response = self.client.post("/api/nsd/v1/subscriptions",
559 data=subscription_req, format='json')
560 self.assertEqual(201, response.status_code)
562 self.user_defined_data = {
567 user_defined_data_json = json.JSONEncoder().encode(self.user_defined_data)
568 mock_parse_nsd.return_value = json.JSONEncoder().encode(nsd_data)
571 vnfdId="vcpe_vfw_zte_1_0"
580 nsPackageId='d0ea5ec3-0b98-438a-9bea-488230cff174',
581 operationalState='DISABLED',
582 usageState='NOT_IN_USE',
583 userDefinedData=user_defined_data_json,
586 with open('nsd_content.txt', 'wt') as fp:
588 with open('nsd_content.txt', 'rt') as fp:
589 resp = self.client.put(
590 "/api/nsd/v1/ns_descriptors/d0ea5ec3-0b98-438a-9bea-488230cff174/nsd_content",
594 with open(os.path.join(CATALOG_ROOT_PATH, 'd0ea5ec3-0b98-438a-9bea-488230cff174/nsd_content.txt')) as fp:
596 file_content = '%s%s' % (file_content, data)
597 ns_pkg = NSPackageModel.objects.filter(nsPackageId="d0ea5ec3-0b98-438a-9bea-488230cff174")
598 self.assertEqual("b632bddc-bccd-4180-bd8d-4e8a9578eff7", ns_pkg[0].nsdId)
599 self.assertEqual(const.PKG_STATUS.ONBOARDED, ns_pkg[0].onboardingState)
600 self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
601 self.assertEqual(None, resp.data)
602 self.assertEqual(file_content, 'test')
603 os.remove('nsd_content.txt')
604 expect_callbackuri = "http://callbackuri.com"
605 expect_notification = {
607 'notificationType': const.NSD_NOTIFICATION_TYPE.NSD_ONBOARDING,
608 'timeStamp': "nowtime()",
609 'nsdInfoId': "d0ea5ec3-0b98-438a-9bea-488230cff174",
610 'nsdId': "b632bddc-bccd-4180-bd8d-4e8a9578eff7",
611 "subscriptionId": "1111",
614 'href': 'http://%s:%s/%s%s' % (pub_config.MSB_SERVICE_IP,
615 pub_config.MSB_SERVICE_PORT,
616 const.NSDM_SUBSCRIPTION_ROOT_URI,
619 'href': 'http://%s:%s/%s/ns_descriptors/%s' % (pub_config.MSB_SERVICE_IP,
620 pub_config.MSB_SERVICE_PORT,
621 const.NSD_URL_PREFIX,
622 "d0ea5ec3-0b98-438a-9bea-488230cff174")
626 mock_requests_post.assert_called_with(expect_callbackuri, data=expect_notification,
627 auth=HTTPBasicAuth("username", "password"),
628 headers={'Connection': 'close'})
631 class NotificationTest(TestCase):
633 NsdmSubscriptionModel(subscriptionid="1",
634 callback_uri="http://127.0.0.1/self",
635 notificationTypes=const.NOTIFICATION_TYPES,
637 nsdInfoId="nsdinfoid1",
638 pnfdInfoIds="pnfdInfoIds1",
643 NsdmSubscriptionModel.objects.all().delete()
645 @mock.patch("requests.post")
646 @mock.patch("uuid.uuid4")
647 @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
648 def test_nsdpkg_notify(self, mock_nowtime, mock_uuid, mock_requests_post):
649 mock_nowtime.return_value = "nowtime()"
650 mock_uuid.return_value = "1111"
651 notify = NsdNotifications(const.NSD_NOTIFICATION_TYPE.NSD_ONBOARDING_FAILURE,
652 nsd_info_id="nsdinfoid1",
654 failure_details="NSD(nsdid1) already exists.", operational_state=None)
655 notify.send_notification()
656 expect_callbackuri = "http://127.0.0.1/self"
657 expect_notification = {
659 'notificationType': const.NSD_NOTIFICATION_TYPE.NSD_ONBOARDING_FAILURE,
660 'timeStamp': "nowtime()",
661 'nsdInfoId': "nsdinfoid1",
663 'onboardingFailureDetails': "NSD(nsdid1) already exists.",
664 "subscriptionId": "1",
667 'href': 'http://%s:%s/%s%s' % (pub_config.MSB_SERVICE_IP,
668 pub_config.MSB_SERVICE_PORT,
669 const.NSDM_SUBSCRIPTION_ROOT_URI,
672 'href': 'http://%s:%s/%s/ns_descriptors/%s' % (pub_config.MSB_SERVICE_IP,
673 pub_config.MSB_SERVICE_PORT,
674 const.NSD_URL_PREFIX,
679 mock_requests_post.assert_called_with(expect_callbackuri, data=expect_notification, headers={'Connection': 'close'})
681 @mock.patch("requests.post")
682 @mock.patch("uuid.uuid4")
683 @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
684 def test_pnfpkg_notify(self, mock_nowtime, mock_uuid, mock_requests_post):
685 mock_nowtime.return_value = "nowtime()"
686 mock_uuid.return_value = "1111"
687 notify = PnfNotifications(const.NSD_NOTIFICATION_TYPE.PNFD_ONBOARDING,
688 pnfd_info_id="pnfdInfoIds1",
690 failure_details=None)
691 notify.send_notification()
692 expect_callbackuri = "http://127.0.0.1/self"
693 expect_notification = {
695 'notificationType': const.NSD_NOTIFICATION_TYPE.PNFD_ONBOARDING,
696 'timeStamp': "nowtime()",
697 'pnfdInfoIds': "pnfdInfoIds1",
699 "subscriptionId": "1",
702 'href': 'http://%s:%s/%s%s' % (pub_config.MSB_SERVICE_IP,
703 pub_config.MSB_SERVICE_PORT,
704 const.NSDM_SUBSCRIPTION_ROOT_URI,
707 'href': 'http://%s:%s/%s/pnf_descriptors/%s' % (pub_config.MSB_SERVICE_IP,
708 pub_config.MSB_SERVICE_PORT,
709 const.NSD_URL_PREFIX,
714 mock_requests_post.assert_called_with(expect_callbackuri, data=expect_notification,
715 headers={'Connection': 'close'})