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 genericparser.packages.biz.nsdm_subscription import NsdmSubscription
23 from genericparser.pub.database.models import NsdmSubscriptionModel
26 class TestNsdmSubscription(TestCase):
29 self.client = APIClient()
30 NsdmSubscriptionModel.objects.all().delete()
31 self.subscription_id = str(uuid.uuid4())
33 "callbackUri": "http://callbackuri.com",
35 "authType": ["BASIC"],
37 "userName": "username",
38 "password": "password"
42 "nsdId": ["b632bddc-abcd-4180-bd8d-4e8a9578eff7"],
47 "href": "/api/v1/subscriptions/" + self.subscription_id
50 self.test_subscription = {
51 "callbackUri": "http://callbackuri.com",
52 "id": self.subscription_id,
54 "notificationTypes": [
55 "NsdOnBoardingNotification"
63 "nestedNsdInfoIds": [],
64 "nsdOnboardingState": [],
65 "nsdOperationalState": [],
72 "pnfdInvariantId": [],
73 "pnfdOnboardingState": [],
82 @mock.patch("requests.get")
83 @mock.patch.object(uuid, 'uuid4')
84 def test_nsdm_subscribe_notification(self, mock_uuid4, mock_requests):
85 temp_uuid = str(uuid.uuid4())
86 mock_requests.return_value.status_code = 204
87 mock_requests.get.return_value.status_code = 204
88 mock_uuid4.return_value = temp_uuid
89 response = self.client.post("/api/nsd/v1/subscriptions",
90 data=self.subscription, format='json')
91 self.assertEqual(201, response.status_code)
92 self.assertEqual(self.subscription["callbackUri"],
93 response.data["callbackUri"])
94 self.assertEqual(temp_uuid, response.data["id"])
96 @mock.patch("requests.get")
97 @mock.patch.object(uuid, 'uuid4')
98 def test_nsdm_subscribe_callbackFailure(self, mock_uuid4, mock_requests):
99 temp_uuid = str(uuid.uuid4())
100 mock_requests.return_value.status_code = 500
101 mock_requests.get.return_value.status_code = 500
102 mock_uuid4.return_value = temp_uuid
105 'detail': "callbackUri http://callbackuri.com didn't"
106 " return 204 statuscode."
108 response = self.client.post("/api/nsd/v1/subscriptions",
109 data=self.subscription, format='json')
110 self.assertEqual(500, response.status_code)
111 self.assertEqual(expected_data, response.data)
113 @mock.patch("requests.get")
114 def test_nsdm_second_subscription(self, mock_requests):
115 mock_requests.return_value.status_code = 204
116 mock_requests.get.return_value.status_code = 204
117 response = self.client.post("/api/nsd/v1/subscriptions",
118 data=self.subscription, format='json')
119 self.assertEqual(201, response.status_code)
120 self.assertEqual(self.subscription["callbackUri"],
121 response.data["callbackUri"])
122 dummy_subscription = {
123 "callbackUri": "http://callbackuri.com",
125 "authType": ["BASIC"],
127 "userName": "username",
128 "password": "password"
132 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
135 response = self.client.post("/api/nsd/v1/subscriptions",
136 data=dummy_subscription, format='json')
137 self.assertEqual(201, response.status_code)
138 self.assertEqual(dummy_subscription["callbackUri"],
139 response.data["callbackUri"])
141 @mock.patch("requests.get")
142 def test_nsdm_duplicate_subscription(self, mock_requests):
143 mock_requests.return_value.status_code = 204
144 mock_requests.get.return_value.status_code = 204
145 response = self.client.post("/api/nsd/v1/subscriptions",
146 data=self.subscription, format='json')
147 self.assertEqual(201, response.status_code)
148 self.assertEqual(self.subscription["callbackUri"],
149 response.data["callbackUri"])
152 'detail': 'Already Subscription exists with'
153 ' the same callbackUri and filter',
155 response = self.client.post("/api/nsd/v1/subscriptions",
156 data=self.subscription, format='json')
157 self.assertEqual(303, response.status_code)
158 self.assertEqual(expected_data, response.data)
160 @mock.patch("requests.get")
161 def test_nsdm_bad_request(self, mock_requests):
162 dummy_subscription = {
163 "callbackUri": "http://callbackuri.com",
165 "authType": ["BASIC"],
167 "userName": "username",
168 "password": "password"
172 "nsdId": "b632bddc-bccd-4180-bd8d-4e8a9578eff7",
175 response = self.client.post("/api/nsd/v1/subscriptions",
176 data=dummy_subscription, format='json')
177 self.assertEqual(400, response.status_code)
179 @mock.patch("requests.get")
180 def test_nsdm_invalid_authtype_subscription(self, mock_requests):
181 dummy_subscription = {
182 "callbackUri": "http://callbackuri.com",
184 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
186 "userName": "username",
187 "password": "password"
191 mock_requests.return_value.status_code = 204
192 mock_requests.get.return_value.status_code = 204
195 'detail': 'Auth type should be BASIC',
197 response = self.client.post("/api/nsd/v1/subscriptions",
198 data=dummy_subscription, format='json')
199 self.assertEqual(400, response.status_code)
200 self.assertEqual(expected_data, response.data)
202 @mock.patch("requests.get")
203 def test_nsdm_invalid_authtype_oauthclient_subscription(
204 self, mock_requests):
205 dummy_subscription = {
206 "callbackUri": "http://callbackuri.com",
208 "authType": ["BASIC"],
209 "paramsOauth2ClientCredentials": {
210 "clientId": "clientId",
211 "clientPassword": "password",
212 "tokenEndpoint": "http://tokenEndpoint"
216 mock_requests.return_value.status_code = 204
217 mock_requests.get.return_value.status_code = 204
220 'detail': 'Auth type should be OAUTH2_CLIENT_CREDENTIALS',
222 response = self.client.post("/api/nsd/v1/subscriptions",
223 data=dummy_subscription, format='json')
224 self.assertEqual(400, response.status_code)
225 self.assertEqual(expected_data, response.data)
227 @mock.patch("requests.get")
228 def test_nsdm_invalid_authparams_subscription(self, mock_requests):
229 dummy_subscription = {
230 "callbackUri": "http://callbackuri.com",
232 "authType": ["BASIC"],
234 "userName": "username"
238 mock_requests.return_value.status_code = 204
239 mock_requests.get.return_value.status_code = 204
242 'detail': 'userName and password needed for BASIC',
244 response = self.client.post("/api/nsd/v1/subscriptions",
245 data=dummy_subscription, format='json')
246 self.assertEqual(400, response.status_code)
247 self.assertEqual(expected_data, response.data)
249 @mock.patch("requests.get")
250 def test_nsdm_invalid_authparams_oauthclient_subscription(
251 self, mock_requests):
252 dummy_subscription = {
253 "callbackUri": "http://callbackuri.com",
255 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
256 "paramsOauth2ClientCredentials": {
257 "clientPassword": "password",
258 "tokenEndpoint": "http://tokenEndpoint"
262 mock_requests.return_value.status_code = 204
263 mock_requests.get.return_value.status_code = 204
266 'detail': 'clientId, clientPassword and tokenEndpoint'
267 ' required for OAUTH2_CLIENT_CREDENTIALS',
269 response = self.client.post("/api/nsd/v1/subscriptions",
270 data=dummy_subscription, format='json')
271 self.assertEqual(400, response.status_code)
272 self.assertEqual(expected_data, response.data)
274 @mock.patch("requests.get")
275 def test_nsdm_invalid_filter_subscription(self, mock_requests):
276 dummy_subscription = {
277 "callbackUri": "http://callbackuri.com",
279 "authType": ["BASIC"],
281 "userName": "username",
282 "password": "password"
286 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
287 "nsdInfoId": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
290 mock_requests.return_value.status_code = 204
291 mock_requests.get.return_value.status_code = 204
294 'detail': 'Notification Filter should contain'
295 ' either nsdId or nsdInfoId',
297 response = self.client.post("/api/nsd/v1/subscriptions",
298 data=dummy_subscription, format='json')
299 self.assertEqual(400, response.status_code)
300 self.assertEqual(expected_data, response.data)
302 @mock.patch("requests.get")
303 def test_nsdm_invalid_filter_pnfd_subscription(self, mock_requests):
304 dummy_subscription = {
305 "callbackUri": "http://callbackuri.com",
307 "authType": ["BASIC"],
309 "userName": "username",
310 "password": "password"
314 "pnfdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
315 "pnfdInfoIds": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
318 mock_requests.return_value.status_code = 204
319 mock_requests.get.return_value.status_code = 204
322 'detail': 'Notification Filter should contain'
323 ' either pnfdId or pnfdInfoIds',
325 response = self.client.post("/api/nsd/v1/subscriptions",
326 data=dummy_subscription, format='json')
327 self.assertEqual(400, response.status_code)
328 self.assertEqual(expected_data, response.data)
330 @mock.patch.object(NsdmSubscription, 'create')
331 def test_nsdmsubscription_create_when_catch_exception(self, mock_create):
332 mock_create.side_effect = TypeError("Unicode type")
333 response = self.client.post('/api/nsd/v1/subscriptions',
334 data=self.subscription, format='json')
335 self.assertEqual(response.status_code,
336 status.HTTP_500_INTERNAL_SERVER_ERROR)
338 def test_nsdm_get_subscriptions(self):
339 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
340 callback_uri="http://callbackuri.com",
342 notificationTypes=json.dumps(
343 ["NsdOnBoardingNotification"]),
344 nsdId=[], nsdVersion=[],
345 nsdInfoId=[], nsdDesigner=[],
346 nsdName=[], nsdInvariantId=[],
347 vnfPkgIds=[], pnfdInfoIds=[],
348 nestedNsdInfoIds=[], nsdOnboardingState=[],
349 nsdOperationalState=[], nsdUsageState=[],
350 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
351 pnfdName=[], pnfdInvariantId=[],
352 pnfdOnboardingState=[], pnfdUsageState=[],
353 links=json.dumps(self.links)).save()
354 response = self.client.get("/api/nsd/v1/subscriptions",
356 self.assertEqual(status.HTTP_200_OK, response.status_code)
357 self.assertEqual([self.test_subscription], response.data)
359 def test_nsdm_get_subscriptions_filter(self):
360 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
361 callback_uri="http://callbackuri.com",
363 notificationTypes=json.dumps(
364 ["NsdOnBoardingNotification"]),
365 nsdId=[], nsdVersion=[],
366 nsdInfoId=[], nsdDesigner=[],
367 nsdName=[], nsdInvariantId=[],
368 vnfPkgIds=[], pnfdInfoIds=[],
369 nestedNsdInfoIds=[], nsdOnboardingState=[],
370 nsdOperationalState=[], nsdUsageState=[],
371 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
372 pnfdName=[], pnfdInvariantId=[],
373 pnfdOnboardingState=[], pnfdUsageState=[],
374 links=json.dumps(self.links)).save()
375 response = self.client.get("/api/nsd/v1/subscriptions"
377 "=NsdOnBoardingNotification",
379 self.assertEqual(status.HTTP_200_OK, response.status_code)
380 self.assertEqual([self.test_subscription], response.data)
382 def test_nsdm_get_subscriptions_filter_failure(self):
383 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
384 callback_uri="http://callbackuri.com",
386 notificationTypes=json.dumps(
387 ["NsdOnBoardingNotification"]),
388 nsdId=[], nsdVersion=[],
389 nsdInfoId=[], nsdDesigner=[],
390 nsdName=[], nsdInvariantId=[],
391 vnfPkgIds=[], pnfdInfoIds=[],
392 nestedNsdInfoIds=[], nsdOnboardingState=[],
393 nsdOperationalState=[], nsdUsageState=[],
394 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
395 pnfdName=[], pnfdInvariantId=[],
396 pnfdOnboardingState=[], pnfdUsageState=[],
397 links=json.dumps(self.links)).save()
398 response = self.client.get("/api/nsd/v1/subscriptions"
399 "?notificationTypes="
400 "PnfdOnBoardingFailureNotification",
402 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
404 def test_nsdm_get_subscriptions_invalid_filter(self):
405 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
406 callback_uri="http://callbackuri.com",
408 notificationTypes=json.dumps(
409 ["NsdOnBoardingNotification"]),
410 nsdId=[], nsdVersion=[],
411 nsdInfoId=[], nsdDesigner=[],
412 nsdName=[], nsdInvariantId=[],
413 vnfPkgIds=[], pnfdInfoIds=[],
414 nestedNsdInfoIds=[], nsdOnboardingState=[],
415 nsdOperationalState=[], nsdUsageState=[],
416 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
417 pnfdName=[], pnfdInvariantId=[],
418 pnfdOnboardingState=[], pnfdUsageState=[],
419 links=json.dumps(self.links)).save()
420 response = self.client.get("/api/nsd/v1/subscriptions"
421 "?notificationTypes="
422 "PnfdOnBoardingFailureNotificati",
424 self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
426 @mock.patch.object(NsdmSubscription, 'query_multi_subscriptions')
427 def test_nsdmsubscription_get_when_catch_exception(self, mock_create):
428 mock_create.side_effect = TypeError("Unicode type")
429 response = self.client.get('/api/nsd/v1/subscriptions', format='json')
430 self.assertEqual(response.status_code,
431 status.HTTP_500_INTERNAL_SERVER_ERROR)
433 def test_nsdm_get_subscription(self):
434 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
435 callback_uri="http://callbackuri.com",
437 notificationTypes=json.dumps(
438 ["NsdOnBoardingNotification"]),
439 nsdId=[], nsdVersion=[],
440 nsdInfoId=[], nsdDesigner=[],
441 nsdName=[], nsdInvariantId=[],
442 vnfPkgIds=[], pnfdInfoIds=[],
443 nestedNsdInfoIds=[], nsdOnboardingState=[],
444 nsdOperationalState=[], nsdUsageState=[],
445 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
446 pnfdName=[], pnfdInvariantId=[],
447 pnfdOnboardingState=[], pnfdUsageState=[],
448 links=json.dumps(self.links)).save()
449 response = self.client.get('/api/nsd/v1/'
450 'subscriptions/' + self.subscription_id,
452 self.assertEqual(status.HTTP_200_OK, response.status_code)
453 self.assertEqual(self.test_subscription, response.data)
455 def test_nsdm_get_subscription_failure(self):
458 "detail": "Subscription(" + self.subscription_id + ") "
461 response = self.client.get('/api/nsd/v1/'
462 'subscriptions/' + self.subscription_id,
464 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
465 self.assertEqual(expected_data, response.data)
467 def test_nsdm_get_subscription_failure_bad_request(self):
468 response = self.client.get("/api/nsd/v1/subscriptions/123",
470 self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
472 @mock.patch.object(NsdmSubscription, 'query_single_subscription')
473 def test_nsdmsubscription_getsingle_when_catch_exception(
475 mock_create.side_effect = TypeError("Unicode type")
476 response = self.client.get('/api/nsd/v1/'
477 'subscriptions/' + self.subscription_id,
479 self.assertEqual(response.status_code,
480 status.HTTP_500_INTERNAL_SERVER_ERROR)
482 def test_ndsm_delete_subscription(self):
483 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
484 callback_uri="http://callbackuri.com",
486 notificationTypes=json.dumps(
487 ["NsdOnBoardingNotification"]),
488 nsdId=[], nsdVersion=[],
489 nsdInfoId=[], nsdDesigner=[],
490 nsdName=[], nsdInvariantId=[],
491 vnfPkgIds=[], pnfdInfoIds=[],
492 nestedNsdInfoIds=[], nsdOnboardingState=[],
493 nsdOperationalState=[], nsdUsageState=[],
494 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
495 pnfdName=[], pnfdInvariantId=[],
496 pnfdOnboardingState=[], pnfdUsageState=[],
497 links=json.dumps(self.links)).save()
498 response = self.client.delete('/api/nsd/v1/'
499 'subscriptions/' + self.subscription_id,
501 self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code)
503 def test_ndsm_delete_subscription_failure(self):
504 response = self.client.delete('/api/nsd/v1/'
505 'subscriptions/' + self.subscription_id,
507 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
509 def test_nsdm_delete_subscription_failure_bad_request(self):
510 response = self.client.delete("/api/nsd/v1/subscriptions/123",
512 self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
514 @mock.patch.object(NsdmSubscription, 'delete_single_subscription')
515 def test_nsdmsubscription_delete_when_catch_exception(self, mock_create):
516 mock_create.side_effect = TypeError("Unicode type")
517 response = self.client.delete('/api/nsd/v1/'
518 'subscriptions/' + self.subscription_id,
520 self.assertEqual(response.status_code,
521 status.HTTP_500_INTERNAL_SERVER_ERROR)