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.
18 from django.test import TestCase
19 from rest_framework.test import APIClient
20 from rest_framework import status
22 from catalog.packages.biz.nsdm_subscription import NsdmSubscription
23 from catalog.pub.database.models import NsdmSubscriptionModel
24 from catalog.packages.biz.notificationsutil import NotificationsUtil, prepare_nsd_notification, prepare_pnfd_notification
25 from catalog.packages import const
26 from catalog.pub.config import config as pub_config
27 import catalog.pub.utils.timeutil
30 class TestNsdmSubscription(TestCase):
33 self.client = APIClient()
34 NsdmSubscriptionModel.objects.all().delete()
35 self.subscription_id = str(uuid.uuid4())
37 "callbackUri": "http://callbackuri.com",
39 "authType": ["BASIC"],
41 "userName": "username",
42 "password": "password"
46 "nsdId": ["b632bddc-abcd-4180-bd8d-4e8a9578eff7"],
51 "href": "/api/v1/subscriptions/" + self.subscription_id
54 self.test_subscription = {
55 "callbackUri": "http://callbackuri.com",
56 "id": self.subscription_id,
58 "notificationTypes": [
59 "NsdOnBoardingNotification"
67 "nestedNsdInfoIds": [],
68 "nsdOnboardingState": [],
69 "nsdOperationalState": [],
76 "pnfdInvariantId": [],
77 "pnfdOnboardingState": [],
86 @mock.patch("requests.get")
87 @mock.patch.object(uuid, 'uuid4')
88 def test_nsdm_subscribe_notification(self, mock_uuid4, mock_requests):
89 temp_uuid = str(uuid.uuid4())
90 mock_requests.return_value.status_code = 204
91 mock_requests.get.return_value.status_code = 204
92 mock_uuid4.return_value = temp_uuid
93 response = self.client.post("/api/nsd/v1/subscriptions",
94 data=self.subscription, format='json')
95 self.assertEqual(201, response.status_code)
96 self.assertEqual(self.subscription["callbackUri"],
97 response.data["callbackUri"])
98 self.assertEqual(temp_uuid, response.data["id"])
100 @mock.patch("requests.get")
101 @mock.patch.object(uuid, 'uuid4')
102 def test_nsdm_subscribe_callbackFailure(self, mock_uuid4, mock_requests):
103 temp_uuid = str(uuid.uuid4())
104 mock_requests.return_value.status_code = 500
105 mock_requests.get.return_value.status_code = 500
106 mock_uuid4.return_value = temp_uuid
109 'detail': "callbackUri http://callbackuri.com didn't"
110 " return 204 statuscode."
112 response = self.client.post("/api/nsd/v1/subscriptions",
113 data=self.subscription, format='json')
114 self.assertEqual(500, response.status_code)
115 self.assertEqual(expected_data, response.data)
117 @mock.patch("requests.get")
118 def test_nsdm_second_subscription(self, mock_requests):
119 mock_requests.return_value.status_code = 204
120 mock_requests.get.return_value.status_code = 204
121 response = self.client.post("/api/nsd/v1/subscriptions",
122 data=self.subscription, format='json')
123 self.assertEqual(201, response.status_code)
124 self.assertEqual(self.subscription["callbackUri"],
125 response.data["callbackUri"])
126 dummy_subscription = {
127 "callbackUri": "http://callbackuri.com",
129 "authType": ["BASIC"],
131 "userName": "username",
132 "password": "password"
136 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
139 response = self.client.post("/api/nsd/v1/subscriptions",
140 data=dummy_subscription, format='json')
141 self.assertEqual(201, response.status_code)
142 self.assertEqual(dummy_subscription["callbackUri"],
143 response.data["callbackUri"])
145 @mock.patch("requests.get")
146 def test_nsdm_duplicate_subscription(self, mock_requests):
147 mock_requests.return_value.status_code = 204
148 mock_requests.get.return_value.status_code = 204
149 response = self.client.post("/api/nsd/v1/subscriptions",
150 data=self.subscription, format='json')
151 self.assertEqual(201, response.status_code)
152 self.assertEqual(self.subscription["callbackUri"],
153 response.data["callbackUri"])
156 'detail': 'Already Subscription exists with'
157 ' the same callbackUri and filter'
159 response = self.client.post("/api/nsd/v1/subscriptions",
160 data=self.subscription, format='json')
161 self.assertEqual(303, response.status_code)
162 self.assertEqual(expected_data, response.data)
164 @mock.patch("requests.get")
165 def test_nsdm_bad_request(self, mock_requests):
166 dummy_subscription = {
167 "callbackUri": "http://callbackuri.com",
169 "authType": ["BASIC"],
171 "userName": "username",
172 "password": "password"
176 "nsdId": "b632bddc-bccd-4180-bd8d-4e8a9578eff7",
179 response = self.client.post("/api/nsd/v1/subscriptions",
180 data=dummy_subscription, format='json')
181 self.assertEqual(400, response.status_code)
183 @mock.patch("requests.get")
184 def test_nsdm_invalid_authtype_subscription(self, mock_requests):
185 dummy_subscription = {
186 "callbackUri": "http://callbackuri.com",
188 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
190 "userName": "username",
191 "password": "password"
195 mock_requests.return_value.status_code = 204
196 mock_requests.get.return_value.status_code = 204
199 'detail': 'Auth type should be BASIC'
201 response = self.client.post("/api/nsd/v1/subscriptions",
202 data=dummy_subscription, format='json')
203 self.assertEqual(400, response.status_code)
204 self.assertEqual(expected_data, response.data)
206 @mock.patch("requests.get")
207 def test_nsdm_invalid_authtype_oauthclient_subscription(
208 self, mock_requests):
209 dummy_subscription = {
210 "callbackUri": "http://callbackuri.com",
212 "authType": ["BASIC"],
213 "paramsOauth2ClientCredentials": {
214 "clientId": "clientId",
215 "clientPassword": "password",
216 "tokenEndpoint": "http://tokenEndpoint"
220 mock_requests.return_value.status_code = 204
221 mock_requests.get.return_value.status_code = 204
224 'detail': 'Auth type should be OAUTH2_CLIENT_CREDENTIALS'
226 response = self.client.post("/api/nsd/v1/subscriptions",
227 data=dummy_subscription, format='json')
228 self.assertEqual(400, response.status_code)
229 self.assertEqual(expected_data, response.data)
231 @mock.patch("requests.get")
232 def test_nsdm_invalid_authparams_subscription(self, mock_requests):
233 dummy_subscription = {
234 "callbackUri": "http://callbackuri.com",
236 "authType": ["BASIC"],
238 "userName": "username"
242 mock_requests.return_value.status_code = 204
243 mock_requests.get.return_value.status_code = 204
246 'detail': 'userName and password needed for BASIC'
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_oauthclient_subscription(
255 self, mock_requests):
256 dummy_subscription = {
257 "callbackUri": "http://callbackuri.com",
259 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
260 "paramsOauth2ClientCredentials": {
261 "clientPassword": "password",
262 "tokenEndpoint": "http://tokenEndpoint"
266 mock_requests.return_value.status_code = 204
267 mock_requests.get.return_value.status_code = 204
270 'detail': 'clientId, clientPassword and tokenEndpoint'
271 ' required for OAUTH2_CLIENT_CREDENTIALS'
273 response = self.client.post("/api/nsd/v1/subscriptions",
274 data=dummy_subscription, format='json')
275 self.assertEqual(400, response.status_code)
276 self.assertEqual(expected_data, response.data)
278 @mock.patch("requests.get")
279 def test_nsdm_invalid_filter_subscription(self, mock_requests):
280 dummy_subscription = {
281 "callbackUri": "http://callbackuri.com",
283 "authType": ["BASIC"],
285 "userName": "username",
286 "password": "password"
290 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
291 "nsdInfoId": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
294 mock_requests.return_value.status_code = 204
295 mock_requests.get.return_value.status_code = 204
298 'detail': 'Notification Filter should contain'
299 ' either nsdId or nsdInfoId'
301 response = self.client.post("/api/nsd/v1/subscriptions",
302 data=dummy_subscription, format='json')
303 self.assertEqual(400, response.status_code)
304 self.assertEqual(expected_data, response.data)
306 @mock.patch("requests.get")
307 def test_nsdm_invalid_filter_pnfd_subscription(self, mock_requests):
308 dummy_subscription = {
309 "callbackUri": "http://callbackuri.com",
311 "authType": ["BASIC"],
313 "userName": "username",
314 "password": "password"
318 "pnfdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
319 "pnfdInfoIds": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
322 mock_requests.return_value.status_code = 204
323 mock_requests.get.return_value.status_code = 204
326 'detail': 'Notification Filter should contain'
327 ' either pnfdId or pnfdInfoIds'
329 response = self.client.post("/api/nsd/v1/subscriptions",
330 data=dummy_subscription, format='json')
331 self.assertEqual(400, response.status_code)
332 self.assertEqual(expected_data, response.data)
334 @mock.patch.object(NsdmSubscription, 'create')
335 def test_nsdmsubscription_create_when_catch_exception(self, mock_create):
336 mock_create.side_effect = TypeError("Unicode type")
337 response = self.client.post('/api/nsd/v1/subscriptions',
338 data=self.subscription, format='json')
339 self.assertEqual(response.status_code,
340 status.HTTP_500_INTERNAL_SERVER_ERROR)
342 def test_nsdm_get_subscriptions(self):
343 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
344 callback_uri="http://callbackuri.com",
346 notificationTypes=json.dumps(
347 ["NsdOnBoardingNotification"]),
348 nsdId=[], nsdVersion=[],
349 nsdInfoId=[], nsdDesigner=[],
350 nsdName=[], nsdInvariantId=[],
351 vnfPkgIds=[], pnfdInfoIds=[],
352 nestedNsdInfoIds=[], nsdOnboardingState=[],
353 nsdOperationalState=[], nsdUsageState=[],
354 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
355 pnfdName=[], pnfdInvariantId=[],
356 pnfdOnboardingState=[], pnfdUsageState=[],
357 links=json.dumps(self.links)).save()
358 response = self.client.get("/api/nsd/v1/subscriptions",
360 self.assertEqual(status.HTTP_200_OK, response.status_code)
361 self.assertEqual([self.test_subscription], response.data)
363 def test_nsdm_get_subscriptions_filter(self):
364 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
365 callback_uri="http://callbackuri.com",
367 notificationTypes=json.dumps(
368 ["NsdOnBoardingNotification"]),
369 nsdId=[], nsdVersion=[],
370 nsdInfoId=[], nsdDesigner=[],
371 nsdName=[], nsdInvariantId=[],
372 vnfPkgIds=[], pnfdInfoIds=[],
373 nestedNsdInfoIds=[], nsdOnboardingState=[],
374 nsdOperationalState=[], nsdUsageState=[],
375 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
376 pnfdName=[], pnfdInvariantId=[],
377 pnfdOnboardingState=[], pnfdUsageState=[],
378 links=json.dumps(self.links)).save()
379 response = self.client.get("/api/nsd/v1/subscriptions"
381 "=NsdOnBoardingNotification",
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_failure(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"
403 "?notificationTypes="
404 "PnfdOnBoardingFailureNotification",
406 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
408 def test_nsdm_get_subscriptions_invalid_filter(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 "PnfdOnBoardingFailureNotificati",
428 self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
430 @mock.patch.object(NsdmSubscription, 'query_multi_subscriptions')
431 def test_nsdmsubscription_get_when_catch_exception(self, mock_create):
432 mock_create.side_effect = TypeError("Unicode type")
433 response = self.client.get('/api/nsd/v1/subscriptions', format='json')
434 self.assertEqual(response.status_code,
435 status.HTTP_500_INTERNAL_SERVER_ERROR)
437 def test_nsdm_get_subscription(self):
438 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
439 callback_uri="http://callbackuri.com",
441 notificationTypes=json.dumps(
442 ["NsdOnBoardingNotification"]),
443 nsdId=[], nsdVersion=[],
444 nsdInfoId=[], nsdDesigner=[],
445 nsdName=[], nsdInvariantId=[],
446 vnfPkgIds=[], pnfdInfoIds=[],
447 nestedNsdInfoIds=[], nsdOnboardingState=[],
448 nsdOperationalState=[], nsdUsageState=[],
449 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
450 pnfdName=[], pnfdInvariantId=[],
451 pnfdOnboardingState=[], pnfdUsageState=[],
452 links=json.dumps(self.links)).save()
453 response = self.client.get('/api/nsd/v1/'
454 'subscriptions/' + self.subscription_id,
456 self.assertEqual(status.HTTP_200_OK, response.status_code)
457 self.assertEqual(self.test_subscription, response.data)
459 def test_nsdm_get_subscription_failure(self):
462 "detail": "Subscription(" + self.subscription_id + ") "
465 response = self.client.get('/api/nsd/v1/'
466 'subscriptions/' + self.subscription_id,
468 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
469 self.assertEqual(expected_data, response.data)
471 def test_nsdm_get_subscription_failure_bad_request(self):
472 response = self.client.get("/api/nsd/v1/subscriptions/123",
474 self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
476 @mock.patch.object(NsdmSubscription, 'query_single_subscription')
477 def test_nsdmsubscription_getsingle_when_catch_exception(
479 mock_create.side_effect = TypeError("Unicode type")
480 response = self.client.get('/api/nsd/v1/'
481 'subscriptions/' + self.subscription_id,
483 self.assertEqual(response.status_code,
484 status.HTTP_500_INTERNAL_SERVER_ERROR)
486 def test_ndsm_delete_subscription(self):
487 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
488 callback_uri="http://callbackuri.com",
490 notificationTypes=json.dumps(
491 ["NsdOnBoardingNotification"]),
492 nsdId=[], nsdVersion=[],
493 nsdInfoId=[], nsdDesigner=[],
494 nsdName=[], nsdInvariantId=[],
495 vnfPkgIds=[], pnfdInfoIds=[],
496 nestedNsdInfoIds=[], nsdOnboardingState=[],
497 nsdOperationalState=[], nsdUsageState=[],
498 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
499 pnfdName=[], pnfdInvariantId=[],
500 pnfdOnboardingState=[], pnfdUsageState=[],
501 links=json.dumps(self.links)).save()
502 response = self.client.delete('/api/nsd/v1/'
503 'subscriptions/' + self.subscription_id,
505 self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code)
507 def test_ndsm_delete_subscription_failure(self):
508 response = self.client.delete('/api/nsd/v1/'
509 'subscriptions/' + self.subscription_id,
511 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
513 def test_nsdm_delete_subscription_failure_bad_request(self):
514 response = self.client.delete("/api/nsd/v1/subscriptions/123",
516 self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
518 @mock.patch.object(NsdmSubscription, 'delete_single_subscription')
519 def test_nsdmsubscription_delete_when_catch_exception(self, mock_create):
520 mock_create.side_effect = TypeError("Unicode type")
521 response = self.client.delete('/api/nsd/v1/'
522 'subscriptions/' + self.subscription_id,
524 self.assertEqual(response.status_code,
525 status.HTTP_500_INTERNAL_SERVER_ERROR)
528 class NotificationTest(TestCase):
530 NsdmSubscriptionModel(subscriptionid="1",
531 callback_uri="http://127.0.0.1/self",
532 notificationTypes=const.NOTIFICATION_TYPES,
534 nsdInfoId="nsdinfoid1",
535 pnfdInfoIds="pnfdInfoIds1",
540 NsdmSubscriptionModel.objects.all().delete()
542 @mock.patch("requests.post")
543 @mock.patch("uuid.uuid4")
544 @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
545 def test_nsdpkg_notify(self, mock_nowtime, mock_uuid, mock_requests_post):
546 mock_nowtime.return_value = "nowtime()"
547 mock_uuid.return_value = "1111"
548 notification_content = prepare_nsd_notification("nsdinfoid1", "nsdid1",
549 const.NSD_NOTIFICATION_TYPE.NSD_ONBOARDING_FAILURE,
550 "NSD(nsdid1) already exists.", operational_state=None)
552 'nsdInfoId': 'nsdInfoId',
555 NotificationsUtil().send_notification(notification_content, filters, False)
556 expect_callbackuri = "http://127.0.0.1/self"
557 expect_notification = {
559 'notificationType': const.NSD_NOTIFICATION_TYPE.NSD_ONBOARDING_FAILURE,
560 'timeStamp': "nowtime()",
561 'nsdInfoId': "nsdinfoid1",
563 'onboardingFailureDetails': "NSD(nsdid1) already exists.",
564 'nsdOperationalState': None,
565 "subscriptionId": "1",
568 'href': 'http://%s:%s/%s%s' % (pub_config.MSB_SERVICE_IP,
569 pub_config.MSB_SERVICE_PORT,
570 const.NSDM_SUBSCRIPTION_ROOT_URI,
573 'href': 'http://%s:%s/%s/ns_descriptors/%s' % (pub_config.MSB_SERVICE_IP,
574 pub_config.MSB_SERVICE_PORT,
575 const.NSD_URL_PREFIX,
580 mock_requests_post.assert_called_with(expect_callbackuri, data=expect_notification, headers={'Connection': 'close'})
582 @mock.patch("requests.post")
583 @mock.patch("uuid.uuid4")
584 @mock.patch.object(catalog.pub.utils.timeutil, "now_time")
585 def test_pnfpkg_notify(self, mock_nowtime, mock_uuid, mock_requests_post):
586 mock_nowtime.return_value = "nowtime()"
587 mock_uuid.return_value = "1111"
588 notification_content = prepare_pnfd_notification("pnfdInfoIds1", 'pnfdId1',
589 const.NSD_NOTIFICATION_TYPE.PNFD_ONBOARDING)
592 'pnfdInfoIds': 'pnfdInfoIds',
594 NotificationsUtil().send_notification(notification_content, filters, False)
595 expect_callbackuri = "http://127.0.0.1/self"
596 expect_notification = {
598 'notificationType': const.NSD_NOTIFICATION_TYPE.PNFD_ONBOARDING,
599 'timeStamp': "nowtime()",
600 'pnfdInfoIds': "pnfdInfoIds1",
602 'onboardingFailureDetails': None,
603 "subscriptionId": "1",
606 'href': 'http://%s:%s/%s%s' % (pub_config.MSB_SERVICE_IP,
607 pub_config.MSB_SERVICE_PORT,
608 const.NSDM_SUBSCRIPTION_ROOT_URI,
611 'href': 'http://%s:%s/%s/pnf_descriptors/%s' % (pub_config.MSB_SERVICE_IP,
612 pub_config.MSB_SERVICE_PORT,
613 const.NSD_URL_PREFIX,
618 mock_requests_post.assert_called_with(expect_callbackuri, data=expect_notification,
619 headers={'Connection': 'close'})