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"
44 "href": "/api/v1/subscriptions/" + self.subscription_id
47 self.test_subscription = {
48 "callbackUri": "http://callbackuri.com",
49 "id": self.subscription_id,
51 "notificationTypes": [
52 "NsdOnBoardingNotification"
60 "nestedNsdInfoIds": [],
61 "nsdOnboardingState": [],
62 "nsdOperationalState": [],
69 "pnfdInvariantId": [],
70 "pnfdOnboardingState": [],
79 @mock.patch("requests.get")
80 @mock.patch.object(uuid, 'uuid4')
81 def test_nsdm_subscribe_notification(self, mock_uuid4, mock_requests):
82 temp_uuid = str(uuid.uuid4())
83 mock_requests.return_value.status_code = 204
84 mock_requests.get.return_value.status_code = 204
85 mock_uuid4.return_value = temp_uuid
86 response = self.client.post("/api/nsd/v1/subscriptions",
87 data=self.subscription, format='json')
88 self.assertEqual(201, response.status_code)
89 self.assertEqual(self.subscription["callbackUri"],
90 response.data["callbackUri"])
91 self.assertEqual(temp_uuid, response.data["id"])
93 @mock.patch("requests.get")
94 @mock.patch.object(uuid, 'uuid4')
95 def test_nsdm_subscribe_callbackFailure(self, mock_uuid4, mock_requests):
96 temp_uuid = str(uuid.uuid4())
97 mock_requests.return_value.status_code = 500
98 mock_requests.get.return_value.status_code = 500
99 mock_uuid4.return_value = temp_uuid
102 'detail': "callbackUri http://callbackuri.com didn't"
103 " return 204 statuscode."
105 response = self.client.post("/api/nsd/v1/subscriptions",
106 data=self.subscription, format='json')
107 self.assertEqual(500, response.status_code)
108 self.assertEqual(expected_data, response.data)
110 @mock.patch("requests.get")
111 def test_nsdm_second_subscription(self, mock_requests):
112 mock_requests.return_value.status_code = 204
113 mock_requests.get.return_value.status_code = 204
114 response = self.client.post("/api/nsd/v1/subscriptions",
115 data=self.subscription, format='json')
116 self.assertEqual(201, response.status_code)
117 self.assertEqual(self.subscription["callbackUri"],
118 response.data["callbackUri"])
119 dummy_subscription = {
120 "callbackUri": "http://callbackuri.com",
122 "authType": ["BASIC"],
124 "userName": "username",
125 "password": "password"
129 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
132 response = self.client.post("/api/nsd/v1/subscriptions",
133 data=dummy_subscription, format='json')
134 self.assertEqual(201, response.status_code)
135 self.assertEqual(dummy_subscription["callbackUri"],
136 response.data["callbackUri"])
138 @mock.patch("requests.get")
139 def test_nsdm_duplicate_subscription(self, mock_requests):
140 mock_requests.return_value.status_code = 204
141 mock_requests.get.return_value.status_code = 204
142 response = self.client.post("/api/nsd/v1/subscriptions",
143 data=self.subscription, format='json')
144 self.assertEqual(201, response.status_code)
145 self.assertEqual(self.subscription["callbackUri"],
146 response.data["callbackUri"])
149 'detail': 'Already Subscription exists with'
150 ' the same callbackUri and filter',
152 response = self.client.post("/api/nsd/v1/subscriptions",
153 data=self.subscription, format='json')
154 self.assertEqual(303, response.status_code)
155 self.assertEqual(expected_data, response.data)
157 @mock.patch("requests.get")
158 def test_nsdm_bad_request(self, mock_requests):
159 dummy_subscription = {
160 "callbackUri": "http://callbackuri.com",
162 "authType": ["BASIC"],
164 "userName": "username",
165 "password": "password"
169 "nsdId": "b632bddc-bccd-4180-bd8d-4e8a9578eff7",
172 response = self.client.post("/api/nsd/v1/subscriptions",
173 data=dummy_subscription, format='json')
174 self.assertEqual(400, response.status_code)
176 @mock.patch("requests.get")
177 def test_nsdm_invalid_authtype_subscription(self, mock_requests):
178 dummy_subscription = {
179 "callbackUri": "http://callbackuri.com",
181 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
183 "userName": "username",
184 "password": "password"
188 mock_requests.return_value.status_code = 204
189 mock_requests.get.return_value.status_code = 204
192 'detail': 'Auth type should be BASIC',
194 response = self.client.post("/api/nsd/v1/subscriptions",
195 data=dummy_subscription, format='json')
196 self.assertEqual(400, response.status_code)
197 self.assertEqual(expected_data, response.data)
199 @mock.patch("requests.get")
200 def test_nsdm_invalid_authtype_oauthclient_subscription(
201 self, mock_requests):
202 dummy_subscription = {
203 "callbackUri": "http://callbackuri.com",
205 "authType": ["BASIC"],
206 "paramsOauth2ClientCredentials": {
207 "clientId": "clientId",
208 "clientPassword": "password",
209 "tokenEndpoint": "http://tokenEndpoint"
213 mock_requests.return_value.status_code = 204
214 mock_requests.get.return_value.status_code = 204
217 'detail': 'Auth type should be OAUTH2_CLIENT_CREDENTIALS',
219 response = self.client.post("/api/nsd/v1/subscriptions",
220 data=dummy_subscription, format='json')
221 self.assertEqual(400, response.status_code)
222 self.assertEqual(expected_data, response.data)
224 @mock.patch("requests.get")
225 def test_nsdm_invalid_authparams_subscription(self, mock_requests):
226 dummy_subscription = {
227 "callbackUri": "http://callbackuri.com",
229 "authType": ["BASIC"],
231 "userName": "username"
235 mock_requests.return_value.status_code = 204
236 mock_requests.get.return_value.status_code = 204
239 'detail': 'userName and password needed for BASIC',
241 response = self.client.post("/api/nsd/v1/subscriptions",
242 data=dummy_subscription, format='json')
243 self.assertEqual(400, response.status_code)
244 self.assertEqual(expected_data, response.data)
246 @mock.patch("requests.get")
247 def test_nsdm_invalid_authparams_oauthclient_subscription(
248 self, mock_requests):
249 dummy_subscription = {
250 "callbackUri": "http://callbackuri.com",
252 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
253 "paramsOauth2ClientCredentials": {
254 "clientPassword": "password",
255 "tokenEndpoint": "http://tokenEndpoint"
259 mock_requests.return_value.status_code = 204
260 mock_requests.get.return_value.status_code = 204
263 'detail': 'clientId, clientPassword and tokenEndpoint'
264 ' required for OAUTH2_CLIENT_CREDENTIALS',
266 response = self.client.post("/api/nsd/v1/subscriptions",
267 data=dummy_subscription, format='json')
268 self.assertEqual(400, response.status_code)
269 self.assertEqual(expected_data, response.data)
271 @mock.patch("requests.get")
272 def test_nsdm_invalid_filter_subscription(self, mock_requests):
273 dummy_subscription = {
274 "callbackUri": "http://callbackuri.com",
276 "authType": ["BASIC"],
278 "userName": "username",
279 "password": "password"
283 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
284 "nsdInfoId": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
287 mock_requests.return_value.status_code = 204
288 mock_requests.get.return_value.status_code = 204
291 'detail': 'Notification Filter should contain'
292 ' either nsdId or nsdInfoId',
294 response = self.client.post("/api/nsd/v1/subscriptions",
295 data=dummy_subscription, format='json')
296 self.assertEqual(400, response.status_code)
297 self.assertEqual(expected_data, response.data)
299 @mock.patch("requests.get")
300 def test_nsdm_invalid_filter_pnfd_subscription(self, mock_requests):
301 dummy_subscription = {
302 "callbackUri": "http://callbackuri.com",
304 "authType": ["BASIC"],
306 "userName": "username",
307 "password": "password"
311 "pnfdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
312 "pnfdInfoIds": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
315 mock_requests.return_value.status_code = 204
316 mock_requests.get.return_value.status_code = 204
319 'detail': 'Notification Filter should contain'
320 ' either pnfdId or pnfdInfoIds',
322 response = self.client.post("/api/nsd/v1/subscriptions",
323 data=dummy_subscription, format='json')
324 self.assertEqual(400, response.status_code)
325 self.assertEqual(expected_data, response.data)
327 @mock.patch.object(NsdmSubscription, 'create')
328 def test_nsdmsubscription_create_when_catch_exception(self, mock_create):
329 mock_create.side_effect = TypeError("Unicode type")
330 response = self.client.post('/api/nsd/v1/subscriptions',
331 data=self.subscription, format='json')
332 self.assertEqual(response.status_code,
333 status.HTTP_500_INTERNAL_SERVER_ERROR)
335 def test_nsdm_get_subscriptions(self):
336 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
337 callback_uri="http://callbackuri.com",
339 notificationTypes=json.dumps(
340 ["NsdOnBoardingNotification"]),
341 nsdId=[], nsdVersion=[],
342 nsdInfoId=[], nsdDesigner=[],
343 nsdName=[], nsdInvariantId=[],
344 vnfPkgIds=[], pnfdInfoIds=[],
345 nestedNsdInfoIds=[], nsdOnboardingState=[],
346 nsdOperationalState=[], nsdUsageState=[],
347 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
348 pnfdName=[], pnfdInvariantId=[],
349 pnfdOnboardingState=[], pnfdUsageState=[],
350 links=json.dumps(self.links)).save()
351 response = self.client.get("/api/nsd/v1/subscriptions",
353 self.assertEqual(status.HTTP_200_OK, response.status_code)
354 self.assertEqual([self.test_subscription], response.data)
356 def test_nsdm_get_subscriptions_filter(self):
357 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
358 callback_uri="http://callbackuri.com",
360 notificationTypes=json.dumps(
361 ["NsdOnBoardingNotification"]),
362 nsdId=[], nsdVersion=[],
363 nsdInfoId=[], nsdDesigner=[],
364 nsdName=[], nsdInvariantId=[],
365 vnfPkgIds=[], pnfdInfoIds=[],
366 nestedNsdInfoIds=[], nsdOnboardingState=[],
367 nsdOperationalState=[], nsdUsageState=[],
368 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
369 pnfdName=[], pnfdInvariantId=[],
370 pnfdOnboardingState=[], pnfdUsageState=[],
371 links=json.dumps(self.links)).save()
372 response = self.client.get("/api/nsd/v1/subscriptions"
374 "=NsdOnBoardingNotification",
376 self.assertEqual(status.HTTP_200_OK, response.status_code)
377 self.assertEqual([self.test_subscription], response.data)
379 def test_nsdm_get_subscriptions_filter_failure(self):
380 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
381 callback_uri="http://callbackuri.com",
383 notificationTypes=json.dumps(
384 ["NsdOnBoardingNotification"]),
385 nsdId=[], nsdVersion=[],
386 nsdInfoId=[], nsdDesigner=[],
387 nsdName=[], nsdInvariantId=[],
388 vnfPkgIds=[], pnfdInfoIds=[],
389 nestedNsdInfoIds=[], nsdOnboardingState=[],
390 nsdOperationalState=[], nsdUsageState=[],
391 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
392 pnfdName=[], pnfdInvariantId=[],
393 pnfdOnboardingState=[], pnfdUsageState=[],
394 links=json.dumps(self.links)).save()
395 response = self.client.get("/api/nsd/v1/subscriptions"
396 "?notificationTypes="
397 "PnfdOnBoardingFailureNotification",
399 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
401 def test_nsdm_get_subscriptions_invalid_filter(self):
402 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
403 callback_uri="http://callbackuri.com",
405 notificationTypes=json.dumps(
406 ["NsdOnBoardingNotification"]),
407 nsdId=[], nsdVersion=[],
408 nsdInfoId=[], nsdDesigner=[],
409 nsdName=[], nsdInvariantId=[],
410 vnfPkgIds=[], pnfdInfoIds=[],
411 nestedNsdInfoIds=[], nsdOnboardingState=[],
412 nsdOperationalState=[], nsdUsageState=[],
413 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
414 pnfdName=[], pnfdInvariantId=[],
415 pnfdOnboardingState=[], pnfdUsageState=[],
416 links=json.dumps(self.links)).save()
417 response = self.client.get("/api/nsd/v1/subscriptions"
418 "?notificationTypes="
419 "PnfdOnBoardingFailureNotificati",
421 self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
423 @mock.patch.object(NsdmSubscription, 'query_multi_subscriptions')
424 def test_nsdmsubscription_get_when_catch_exception(self, mock_create):
425 mock_create.side_effect = TypeError("Unicode type")
426 response = self.client.get('/api/nsd/v1/subscriptions', format='json')
427 self.assertEqual(response.status_code,
428 status.HTTP_500_INTERNAL_SERVER_ERROR)
430 def test_nsdm_get_subscription(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/'
447 'subscriptions/' + self.subscription_id,
449 self.assertEqual(status.HTTP_200_OK, response.status_code)
450 self.assertEqual(self.test_subscription, response.data)
452 def test_nsdm_get_subscription_failure(self):
455 "detail": "Subscription(" + self.subscription_id + ") "
458 response = self.client.get('/api/nsd/v1/'
459 'subscriptions/' + self.subscription_id,
461 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
462 self.assertEqual(expected_data, response.data)
464 def test_nsdm_get_subscription_failure_bad_request(self):
465 response = self.client.get("/api/nsd/v1/subscriptions/123",
467 self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
469 @mock.patch.object(NsdmSubscription, 'query_single_subscription')
470 def test_nsdmsubscription_getsingle_when_catch_exception(
472 mock_create.side_effect = TypeError("Unicode type")
473 response = self.client.get('/api/nsd/v1/'
474 'subscriptions/' + self.subscription_id,
476 self.assertEqual(response.status_code,
477 status.HTTP_500_INTERNAL_SERVER_ERROR)
479 def test_ndsm_delete_subscription(self):
480 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
481 callback_uri="http://callbackuri.com",
483 notificationTypes=json.dumps(
484 ["NsdOnBoardingNotification"]),
485 nsdId=[], nsdVersion=[],
486 nsdInfoId=[], nsdDesigner=[],
487 nsdName=[], nsdInvariantId=[],
488 vnfPkgIds=[], pnfdInfoIds=[],
489 nestedNsdInfoIds=[], nsdOnboardingState=[],
490 nsdOperationalState=[], nsdUsageState=[],
491 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
492 pnfdName=[], pnfdInvariantId=[],
493 pnfdOnboardingState=[], pnfdUsageState=[],
494 links=json.dumps(self.links)).save()
495 response = self.client.delete('/api/nsd/v1/'
496 'subscriptions/' + self.subscription_id,
498 self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code)
500 def test_ndsm_delete_subscription_failure(self):
501 response = self.client.delete('/api/nsd/v1/'
502 'subscriptions/' + self.subscription_id,
504 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
506 def test_nsdm_delete_subscription_failure_bad_request(self):
507 response = self.client.delete("/api/nsd/v1/subscriptions/123",
509 self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
511 @mock.patch.object(NsdmSubscription, 'delete_single_subscription')
512 def test_nsdmsubscription_delete_when_catch_exception(self, mock_create):
513 mock_create.side_effect = TypeError("Unicode type")
514 response = self.client.delete('/api/nsd/v1/'
515 'subscriptions/' + self.subscription_id,
517 self.assertEqual(response.status_code,
518 status.HTTP_500_INTERNAL_SERVER_ERROR)