genericparser/packages/tests/test_nsdm_subscription.py

   1 # Copyright (C) 2019 Verizon. All Rights Reserved
   2 #
   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
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   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.
  14
  15 import json
  16 import mock
  17 import uuid
  18 from django.test import TestCase
  19 from rest_framework.test import APIClient
  20 from rest_framework import status
  21
  22 from genericparser.packages.biz.nsdm_subscription import NsdmSubscription
  23 from genericparser.pub.database.models import NsdmSubscriptionModel
  24
  25
  26 class TestNsdmSubscription(TestCase):
  27
  28     def setUp(self):
  29         self.client = APIClient()
  30         NsdmSubscriptionModel.objects.all().delete()
  31         self.subscription_id = str(uuid.uuid4())
  32         self.subscription = {
  33             "callbackUri": "http://callbackuri.com",
  34             "authentication": {
  35                 "authType": ["BASIC"],
  36                 "paramsBasic": {
  37                     "userName": "username",
  38                     "password": "password"
  39                 }
  40             }
  41         }
  42         self.links = {
  43             "self": {
  44                 "href": "/api/v1/subscriptions/" + self.subscription_id
  45             }
  46         }
  47         self.test_subscription = {
  48             "callbackUri": "http://callbackuri.com",
  49             "id": self.subscription_id,
  50             "filter": {
  51                 "notificationTypes": [
  52                     "NsdOnBoardingNotification"
  53                 ],
  54                 "nsdInfoId": [],
  55                 "nsdId": [],
  56                 "nsdName": [],
  57                 "nsdVersion": [],
  58                 "nsdInvariantId": [],
  59                 "vnfPkgIds": [],
  60                 "nestedNsdInfoIds": [],
  61                 "nsdOnboardingState": [],
  62                 "nsdOperationalState": [],
  63                 "nsdUsageState": [],
  64                 "pnfdInfoIds": [],
  65                 "pnfdId": [],
  66                 "pnfdName": [],
  67                 "pnfdVersion": [],
  68                 "pnfdProvider": [],
  69                 "pnfdInvariantId": [],
  70                 "pnfdOnboardingState": [],
  71                 "pnfdUsageState": []
  72             },
  73             "_links": self.links,
  74         }
  75
  76     def tearDown(self):
  77         pass
  78
  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"])
  92
  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
 100         expected_data = {
 101             'status': 500,
 102             'detail': "callbackUri http://callbackuri.com didn't"
 103                       " return 204 statuscode."
 104         }
 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)
 109
 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",
 121             "authentication": {
 122                 "authType": ["BASIC"],
 123                 "paramsBasic": {
 124                     "userName": "username",
 125                     "password": "password"
 126                 }
 127             },
 128             "filter": {
 129                 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
 130             }
 131         }
 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"])
 137
 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"])
 147         expected_data = {
 148             'status': 303,
 149             'detail': 'Already Subscription exists with'
 150                       ' the same callbackUri and filter',
 151         }
 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)
 156
 157     @mock.patch("requests.get")
 158     def test_nsdm_bad_request(self, mock_requests):
 159         dummy_subscription = {
 160             "callbackUri": "http://callbackuri.com",
 161             "authentication": {
 162                 "authType": ["BASIC"],
 163                 "paramsBasic": {
 164                     "userName": "username",
 165                     "password": "password"
 166                 }
 167             },
 168             "filter": {
 169                 "nsdId": "b632bddc-bccd-4180-bd8d-4e8a9578eff7",
 170             }
 171         }
 172         response = self.client.post("/api/nsd/v1/subscriptions",
 173                                     data=dummy_subscription, format='json')
 174         self.assertEqual(400, response.status_code)
 175
 176     @mock.patch("requests.get")
 177     def test_nsdm_invalid_authtype_subscription(self, mock_requests):
 178         dummy_subscription = {
 179             "callbackUri": "http://callbackuri.com",
 180             "authentication": {
 181                 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
 182                 "paramsBasic": {
 183                     "userName": "username",
 184                     "password": "password"
 185                 }
 186             }
 187         }
 188         mock_requests.return_value.status_code = 204
 189         mock_requests.get.return_value.status_code = 204
 190         expected_data = {
 191             'status': 400,
 192             'detail': 'Auth type should be BASIC',
 193         }
 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)
 198
 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",
 204             "authentication": {
 205                 "authType": ["BASIC"],
 206                 "paramsOauth2ClientCredentials": {
 207                     "clientId": "clientId",
 208                     "clientPassword": "password",
 209                     "tokenEndpoint": "http://tokenEndpoint"
 210                 }
 211             }
 212         }
 213         mock_requests.return_value.status_code = 204
 214         mock_requests.get.return_value.status_code = 204
 215         expected_data = {
 216             'status': 400,
 217             'detail': 'Auth type should be OAUTH2_CLIENT_CREDENTIALS',
 218         }
 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)
 223
 224     @mock.patch("requests.get")
 225     def test_nsdm_invalid_authparams_subscription(self, mock_requests):
 226         dummy_subscription = {
 227             "callbackUri": "http://callbackuri.com",
 228             "authentication": {
 229                 "authType": ["BASIC"],
 230                 "paramsBasic": {
 231                     "userName": "username"
 232                 }
 233             }
 234         }
 235         mock_requests.return_value.status_code = 204
 236         mock_requests.get.return_value.status_code = 204
 237         expected_data = {
 238             'status': 400,
 239             'detail': 'userName and password needed for BASIC',
 240         }
 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)
 245
 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",
 251             "authentication": {
 252                 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
 253                 "paramsOauth2ClientCredentials": {
 254                     "clientPassword": "password",
 255                     "tokenEndpoint": "http://tokenEndpoint"
 256                 }
 257             }
 258         }
 259         mock_requests.return_value.status_code = 204
 260         mock_requests.get.return_value.status_code = 204
 261         expected_data = {
 262             'status': 400,
 263             'detail': 'clientId, clientPassword and tokenEndpoint'
 264                       ' required for OAUTH2_CLIENT_CREDENTIALS',
 265         }
 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)
 270
 271     @mock.patch("requests.get")
 272     def test_nsdm_invalid_filter_subscription(self, mock_requests):
 273         dummy_subscription = {
 274             "callbackUri": "http://callbackuri.com",
 275             "authentication": {
 276                 "authType": ["BASIC"],
 277                 "paramsBasic": {
 278                     "userName": "username",
 279                     "password": "password"
 280                 }
 281             },
 282             "filter": {
 283                 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
 284                 "nsdInfoId": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
 285             }
 286         }
 287         mock_requests.return_value.status_code = 204
 288         mock_requests.get.return_value.status_code = 204
 289         expected_data = {
 290             'status': 400,
 291             'detail': 'Notification Filter should contain'
 292                       ' either nsdId or nsdInfoId',
 293         }
 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)
 298
 299     @mock.patch("requests.get")
 300     def test_nsdm_invalid_filter_pnfd_subscription(self, mock_requests):
 301         dummy_subscription = {
 302             "callbackUri": "http://callbackuri.com",
 303             "authentication": {
 304                 "authType": ["BASIC"],
 305                 "paramsBasic": {
 306                     "userName": "username",
 307                     "password": "password"
 308                 }
 309             },
 310             "filter": {
 311                 "pnfdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
 312                 "pnfdInfoIds": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
 313             }
 314         }
 315         mock_requests.return_value.status_code = 204
 316         mock_requests.get.return_value.status_code = 204
 317         expected_data = {
 318             'status': 400,
 319             'detail': 'Notification Filter should contain'
 320                       ' either pnfdId or pnfdInfoIds',
 321         }
 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)
 326
 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)
 334
 335     def test_nsdm_get_subscriptions(self):
 336         NsdmSubscriptionModel(subscriptionid=self.subscription_id,
 337                               callback_uri="http://callbackuri.com",
 338                               auth_info={},
 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",
 352                                    format='json')
 353         self.assertEqual(status.HTTP_200_OK, response.status_code)
 354         self.assertEqual([self.test_subscription], response.data)
 355
 356     def test_nsdm_get_subscriptions_filter(self):
 357         NsdmSubscriptionModel(subscriptionid=self.subscription_id,
 358                               callback_uri="http://callbackuri.com",
 359                               auth_info={},
 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"
 373                                    "?notificationTypes"
 374                                    "=NsdOnBoardingNotification",
 375                                    format='json')
 376         self.assertEqual(status.HTTP_200_OK, response.status_code)
 377         self.assertEqual([self.test_subscription], response.data)
 378
 379     def test_nsdm_get_subscriptions_filter_failure(self):
 380         NsdmSubscriptionModel(subscriptionid=self.subscription_id,
 381                               callback_uri="http://callbackuri.com",
 382                               auth_info={},
 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",
 398                                    format='json')
 399         self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
 400
 401     def test_nsdm_get_subscriptions_invalid_filter(self):
 402         NsdmSubscriptionModel(subscriptionid=self.subscription_id,
 403                               callback_uri="http://callbackuri.com",
 404                               auth_info={},
 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",
 420                                    format='json')
 421         self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
 422
 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)
 429
 430     def test_nsdm_get_subscription(self):
 431         NsdmSubscriptionModel(subscriptionid=self.subscription_id,
 432                               callback_uri="http://callbackuri.com",
 433                               auth_info={},
 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,
 448                                    format='json')
 449         self.assertEqual(status.HTTP_200_OK, response.status_code)
 450         self.assertEqual(self.test_subscription, response.data)
 451
 452     def test_nsdm_get_subscription_failure(self):
 453         expected_data = {
 454             "status": 404,
 455             "detail": "Subscription(" + self.subscription_id + ") "
 456             "doesn't exists"
 457         }
 458         response = self.client.get('/api/nsd/v1/'
 459                                    'subscriptions/' + self.subscription_id,
 460                                    format='json')
 461         self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
 462         self.assertEqual(expected_data, response.data)
 463
 464     def test_nsdm_get_subscription_failure_bad_request(self):
 465         response = self.client.get("/api/nsd/v1/subscriptions/123",
 466                                    format='json')
 467         self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
 468
 469     @mock.patch.object(NsdmSubscription, 'query_single_subscription')
 470     def test_nsdmsubscription_getsingle_when_catch_exception(
 471             self, mock_create):
 472         mock_create.side_effect = TypeError("Unicode type")
 473         response = self.client.get('/api/nsd/v1/'
 474                                    'subscriptions/' + self.subscription_id,
 475                                    format='json')
 476         self.assertEqual(response.status_code,
 477                          status.HTTP_500_INTERNAL_SERVER_ERROR)
 478
 479     def test_ndsm_delete_subscription(self):
 480         NsdmSubscriptionModel(subscriptionid=self.subscription_id,
 481                               callback_uri="http://callbackuri.com",
 482                               auth_info={},
 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,
 497                                       format='json')
 498         self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code)
 499
 500     def test_ndsm_delete_subscription_failure(self):
 501         response = self.client.delete('/api/nsd/v1/'
 502                                       'subscriptions/' + self.subscription_id,
 503                                       format='json')
 504         self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
 505
 506     def test_nsdm_delete_subscription_failure_bad_request(self):
 507         response = self.client.delete("/api/nsd/v1/subscriptions/123",
 508                                       format='json')
 509         self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
 510
 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,
 516                                       format='json')
 517         self.assertEqual(response.status_code,
 518                          status.HTTP_500_INTERNAL_SERVER_ERROR)