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.",
104 'title': 'Creating Subscription Failed!'
106 response = self.client.post("/api/nsd/v1/subscriptions",
107 data=self.subscription, format='json')
108 self.assertEqual(500, response.status_code)
109 self.assertEqual(expected_data, response.data)
111 @mock.patch("requests.get")
112 def test_nsdm_second_subscription(self, mock_requests):
113 mock_requests.return_value.status_code = 204
114 mock_requests.get.return_value.status_code = 204
115 response = self.client.post("/api/nsd/v1/subscriptions",
116 data=self.subscription, format='json')
117 self.assertEqual(201, response.status_code)
118 self.assertEqual(self.subscription["callbackUri"],
119 response.data["callbackUri"])
120 dummy_subscription = {
121 "callbackUri": "http://callbackuri.com",
123 "authType": ["BASIC"],
125 "userName": "username",
126 "password": "password"
130 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
133 response = self.client.post("/api/nsd/v1/subscriptions",
134 data=dummy_subscription, format='json')
135 self.assertEqual(201, response.status_code)
136 self.assertEqual(dummy_subscription["callbackUri"],
137 response.data["callbackUri"])
139 @mock.patch("requests.get")
140 def test_nsdm_duplicate_subscription(self, mock_requests):
141 mock_requests.return_value.status_code = 204
142 mock_requests.get.return_value.status_code = 204
143 response = self.client.post("/api/nsd/v1/subscriptions",
144 data=self.subscription, format='json')
145 self.assertEqual(201, response.status_code)
146 self.assertEqual(self.subscription["callbackUri"],
147 response.data["callbackUri"])
150 'detail': 'Already Subscription exists with'
151 ' the same callbackUri and filter',
152 'title': 'Creating Subscription Failed!'
154 response = self.client.post("/api/nsd/v1/subscriptions",
155 data=self.subscription, format='json')
156 self.assertEqual(303, response.status_code)
157 self.assertEqual(expected_data, response.data)
159 @mock.patch("requests.get")
160 def test_nsdm_bad_request(self, mock_requests):
161 dummy_subscription = {
162 "callbackUri": "http://callbackuri.com",
164 "authType": ["BASIC"],
166 "userName": "username",
167 "password": "password"
171 "nsdId": "b632bddc-bccd-4180-bd8d-4e8a9578eff7",
174 response = self.client.post("/api/nsd/v1/subscriptions",
175 data=dummy_subscription, format='json')
176 self.assertEqual(400, response.status_code)
178 @mock.patch("requests.get")
179 def test_nsdm_invalid_authtype_subscription(self, mock_requests):
180 dummy_subscription = {
181 "callbackUri": "http://callbackuri.com",
183 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
185 "userName": "username",
186 "password": "password"
190 mock_requests.return_value.status_code = 204
191 mock_requests.get.return_value.status_code = 204
194 'detail': 'Auth type should be BASIC',
195 'title': 'Creating Subscription Failed!'
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',
221 'title': 'Creating Subscription Failed!'
223 response = self.client.post("/api/nsd/v1/subscriptions",
224 data=dummy_subscription, format='json')
225 self.assertEqual(400, response.status_code)
226 self.assertEqual(expected_data, response.data)
228 @mock.patch("requests.get")
229 def test_nsdm_invalid_authparams_subscription(self, mock_requests):
230 dummy_subscription = {
231 "callbackUri": "http://callbackuri.com",
233 "authType": ["BASIC"],
235 "userName": "username"
239 mock_requests.return_value.status_code = 204
240 mock_requests.get.return_value.status_code = 204
243 'detail': 'userName and password needed for BASIC',
244 'title': 'Creating Subscription Failed!'
246 response = self.client.post("/api/nsd/v1/subscriptions",
247 data=dummy_subscription, format='json')
248 self.assertEqual(400, response.status_code)
249 self.assertEqual(expected_data, response.data)
251 @mock.patch("requests.get")
252 def test_nsdm_invalid_authparams_oauthclient_subscription(
253 self, mock_requests):
254 dummy_subscription = {
255 "callbackUri": "http://callbackuri.com",
257 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
258 "paramsOauth2ClientCredentials": {
259 "clientPassword": "password",
260 "tokenEndpoint": "http://tokenEndpoint"
264 mock_requests.return_value.status_code = 204
265 mock_requests.get.return_value.status_code = 204
268 'detail': 'clientId, clientPassword and tokenEndpoint'
269 ' required for OAUTH2_CLIENT_CREDENTIALS',
270 'title': 'Creating Subscription Failed!'
272 response = self.client.post("/api/nsd/v1/subscriptions",
273 data=dummy_subscription, format='json')
274 self.assertEqual(400, response.status_code)
275 self.assertEqual(expected_data, response.data)
277 @mock.patch("requests.get")
278 def test_nsdm_invalid_filter_subscription(self, mock_requests):
279 dummy_subscription = {
280 "callbackUri": "http://callbackuri.com",
282 "authType": ["BASIC"],
284 "userName": "username",
285 "password": "password"
289 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
290 "nsdInfoId": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
293 mock_requests.return_value.status_code = 204
294 mock_requests.get.return_value.status_code = 204
297 'detail': 'Notification Filter should contain'
298 ' either nsdId or nsdInfoId',
299 'title': 'Creating Subscription Failed!'
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',
328 'title': 'Creating Subscription Failed!'
330 response = self.client.post("/api/nsd/v1/subscriptions",
331 data=dummy_subscription, format='json')
332 self.assertEqual(400, response.status_code)
333 self.assertEqual(expected_data, response.data)
335 @mock.patch.object(NsdmSubscription, 'create')
336 def test_nsdmsubscription_create_when_catch_exception(self, mock_create):
337 mock_create.side_effect = TypeError("Unicode type")
338 response = self.client.post('/api/nsd/v1/subscriptions',
339 data=self.subscription, format='json')
340 self.assertEqual(response.status_code,
341 status.HTTP_500_INTERNAL_SERVER_ERROR)
343 def test_nsdm_get_subscriptions(self):
344 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
345 callback_uri="http://callbackuri.com",
347 notificationTypes=json.dumps(
348 ["NsdOnBoardingNotification"]),
349 nsdId=[], nsdVersion=[],
350 nsdInfoId=[], nsdDesigner=[],
351 nsdName=[], nsdInvariantId=[],
352 vnfPkgIds=[], pnfdInfoIds=[],
353 nestedNsdInfoIds=[], nsdOnboardingState=[],
354 nsdOperationalState=[], nsdUsageState=[],
355 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
356 pnfdName=[], pnfdInvariantId=[],
357 pnfdOnboardingState=[], pnfdUsageState=[],
358 links=json.dumps(self.links)).save()
359 response = self.client.get("/api/nsd/v1/subscriptions",
361 self.assertEqual(status.HTTP_200_OK, response.status_code)
362 self.assertEqual([self.test_subscription], response.data)
364 def test_nsdm_get_subscriptions_filter(self):
365 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
366 callback_uri="http://callbackuri.com",
368 notificationTypes=json.dumps(
369 ["NsdOnBoardingNotification"]),
370 nsdId=[], nsdVersion=[],
371 nsdInfoId=[], nsdDesigner=[],
372 nsdName=[], nsdInvariantId=[],
373 vnfPkgIds=[], pnfdInfoIds=[],
374 nestedNsdInfoIds=[], nsdOnboardingState=[],
375 nsdOperationalState=[], nsdUsageState=[],
376 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
377 pnfdName=[], pnfdInvariantId=[],
378 pnfdOnboardingState=[], pnfdUsageState=[],
379 links=json.dumps(self.links)).save()
380 response = self.client.get("/api/nsd/v1/subscriptions"
382 "=NsdOnBoardingNotification",
384 self.assertEqual(status.HTTP_200_OK, response.status_code)
385 self.assertEqual([self.test_subscription], response.data)
387 def test_nsdm_get_subscriptions_filter_failure(self):
388 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
389 callback_uri="http://callbackuri.com",
391 notificationTypes=json.dumps(
392 ["NsdOnBoardingNotification"]),
393 nsdId=[], nsdVersion=[],
394 nsdInfoId=[], nsdDesigner=[],
395 nsdName=[], nsdInvariantId=[],
396 vnfPkgIds=[], pnfdInfoIds=[],
397 nestedNsdInfoIds=[], nsdOnboardingState=[],
398 nsdOperationalState=[], nsdUsageState=[],
399 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
400 pnfdName=[], pnfdInvariantId=[],
401 pnfdOnboardingState=[], pnfdUsageState=[],
402 links=json.dumps(self.links)).save()
403 response = self.client.get("/api/nsd/v1/subscriptions"
404 "?notificationTypes="
405 "PnfdOnBoardingFailureNotification",
407 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
409 def test_nsdm_get_subscriptions_invalid_filter(self):
410 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
411 callback_uri="http://callbackuri.com",
413 notificationTypes=json.dumps(
414 ["NsdOnBoardingNotification"]),
415 nsdId=[], nsdVersion=[],
416 nsdInfoId=[], nsdDesigner=[],
417 nsdName=[], nsdInvariantId=[],
418 vnfPkgIds=[], pnfdInfoIds=[],
419 nestedNsdInfoIds=[], nsdOnboardingState=[],
420 nsdOperationalState=[], nsdUsageState=[],
421 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
422 pnfdName=[], pnfdInvariantId=[],
423 pnfdOnboardingState=[], pnfdUsageState=[],
424 links=json.dumps(self.links)).save()
425 response = self.client.get("/api/nsd/v1/subscriptions"
426 "?notificationTypes="
427 "PnfdOnBoardingFailureNotificati",
429 self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
431 @mock.patch.object(NsdmSubscription, 'query_multi_subscriptions')
432 def test_nsdmsubscription_get_when_catch_exception(self, mock_create):
433 mock_create.side_effect = TypeError("Unicode type")
434 response = self.client.get('/api/nsd/v1/subscriptions', format='json')
435 self.assertEqual(response.status_code,
436 status.HTTP_500_INTERNAL_SERVER_ERROR)
438 def test_nsdm_get_subscription(self):
439 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
440 callback_uri="http://callbackuri.com",
442 notificationTypes=json.dumps(
443 ["NsdOnBoardingNotification"]),
444 nsdId=[], nsdVersion=[],
445 nsdInfoId=[], nsdDesigner=[],
446 nsdName=[], nsdInvariantId=[],
447 vnfPkgIds=[], pnfdInfoIds=[],
448 nestedNsdInfoIds=[], nsdOnboardingState=[],
449 nsdOperationalState=[], nsdUsageState=[],
450 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
451 pnfdName=[], pnfdInvariantId=[],
452 pnfdOnboardingState=[], pnfdUsageState=[],
453 links=json.dumps(self.links)).save()
454 response = self.client.get('/api/nsd/v1/'
455 'subscriptions/' + self.subscription_id,
457 self.assertEqual(status.HTTP_200_OK, response.status_code)
458 self.assertEqual(self.test_subscription, response.data)
460 def test_nsdm_get_subscription_failure(self):
462 "title": "Query Subscription Failed!",
464 "detail": "Subscription(" + self.subscription_id + ") "
467 response = self.client.get('/api/nsd/v1/'
468 'subscriptions/' + self.subscription_id,
470 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
471 self.assertEqual(expected_data, response.data)
473 def test_nsdm_get_subscription_failure_bad_request(self):
474 response = self.client.get("/api/nsd/v1/subscriptions/123",
476 self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
478 @mock.patch.object(NsdmSubscription, 'query_single_subscription')
479 def test_nsdmsubscription_getsingle_when_catch_exception(
481 mock_create.side_effect = TypeError("Unicode type")
482 response = self.client.get('/api/nsd/v1/'
483 'subscriptions/' + self.subscription_id,
485 self.assertEqual(response.status_code,
486 status.HTTP_500_INTERNAL_SERVER_ERROR)
488 def test_ndsm_delete_subscription(self):
489 NsdmSubscriptionModel(subscriptionid=self.subscription_id,
490 callback_uri="http://callbackuri.com",
492 notificationTypes=json.dumps(
493 ["NsdOnBoardingNotification"]),
494 nsdId=[], nsdVersion=[],
495 nsdInfoId=[], nsdDesigner=[],
496 nsdName=[], nsdInvariantId=[],
497 vnfPkgIds=[], pnfdInfoIds=[],
498 nestedNsdInfoIds=[], nsdOnboardingState=[],
499 nsdOperationalState=[], nsdUsageState=[],
500 pnfdId=[], pnfdVersion=[], pnfdProvider=[],
501 pnfdName=[], pnfdInvariantId=[],
502 pnfdOnboardingState=[], pnfdUsageState=[],
503 links=json.dumps(self.links)).save()
504 response = self.client.delete('/api/nsd/v1/'
505 'subscriptions/' + self.subscription_id,
507 self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code)
509 def test_ndsm_delete_subscription_failure(self):
510 response = self.client.delete('/api/nsd/v1/'
511 'subscriptions/' + self.subscription_id,
513 self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
515 def test_nsdm_delete_subscription_failure_bad_request(self):
516 response = self.client.delete("/api/nsd/v1/subscriptions/123",
518 self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
520 @mock.patch.object(NsdmSubscription, 'delete_single_subscription')
521 def test_nsdmsubscription_delete_when_catch_exception(self, mock_create):
522 mock_create.side_effect = TypeError("Unicode type")
523 response = self.client.delete('/api/nsd/v1/'
524 'subscriptions/' + self.subscription_id,
526 self.assertEqual(response.status_code,
527 status.HTTP_500_INTERNAL_SERVER_ERROR)