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             "filter": {},
  34             "callbackUri": "http://callbackuri.com",
  35             "authentication": {
  36                 "authType": ["BASIC"],
  37                 "paramsBasic": {
  38                     "userName": "username",
  39                     "password": "password"
  40                 }
  41             }
  42         }
  43         self.links = {
  44             "self": {
  45                 "href": "/api/v1/subscriptions/" + self.subscription_id
  46             }
  47         }
  48         self.test_subscription = {
  49             "callbackUri": "http://callbackuri.com",
  50             "id": self.subscription_id,
  51             "filter": {
  52                 "notificationTypes": [
  53                     "NsdOnBoardingNotification"
  54                 ],
  55                 "nsdInfoId": [],
  56                 "nsdId": [],
  57                 "nsdName": [],
  58                 "nsdVersion": [],
  59                 "nsdInvariantId": [],
  60                 "vnfPkgIds": [],
  61                 "nestedNsdInfoIds": [],
  62                 "nsdOnboardingState": [],
  63                 "nsdOperationalState": [],
  64                 "nsdUsageState": [],
  65                 "pnfdInfoIds": [],
  66                 "pnfdId": [],
  67                 "pnfdName": [],
  68                 "pnfdVersion": [],
  69                 "pnfdProvider": [],
  70                 "pnfdInvariantId": [],
  71                 "pnfdOnboardingState": [],
  72                 "pnfdUsageState": []
  73             },
  74             "_links": self.links,
  75         }
  76
  77     def tearDown(self):
  78         pass
  79
  80     @mock.patch("requests.get")
  81     @mock.patch.object(uuid, 'uuid4')
  82     def test_nsdm_subscribe_notification(self, mock_uuid4, mock_requests):
  83         temp_uuid = str(uuid.uuid4())
  84         mock_requests.return_value.status_code = 204
  85         mock_requests.get.return_value.status_code = 204
  86         mock_uuid4.return_value = temp_uuid
  87         response = self.client.post("/api/nsd/v1/subscriptions",
  88                                     data=self.subscription, format='json')
  89         self.assertEqual(201, response.status_code)
  90         self.assertEqual(self.subscription["callbackUri"],
  91                          response.data["callbackUri"])
  92         self.assertEqual(temp_uuid, response.data["id"])
  93
  94     @mock.patch("requests.get")
  95     @mock.patch.object(uuid, 'uuid4')
  96     def test_nsdm_subscribe_callbackFailure(self, mock_uuid4, mock_requests):
  97         temp_uuid = str(uuid.uuid4())
  98         mock_requests.return_value.status_code = 500
  99         mock_requests.get.return_value.status_code = 500
 100         mock_uuid4.return_value = temp_uuid
 101         expected_data = {
 102             'status': 500,
 103             'detail': "callbackUri http://callbackuri.com didn't"
 104                       " return 204 statuscode."
 105         }
 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)
 110
 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",
 122             "authentication": {
 123                 "authType": ["BASIC"],
 124                 "paramsBasic": {
 125                     "userName": "username",
 126                     "password": "password"
 127                 }
 128             },
 129             "filter": {
 130                 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
 131             }
 132         }
 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"])
 138
 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"])
 148         expected_data = {
 149             'status': 303,
 150             'detail': 'Already Subscription exists with'
 151                       ' the same callbackUri and filter',
 152         }
 153         response = self.client.post("/api/nsd/v1/subscriptions",
 154                                     data=self.subscription, format='json')
 155         self.assertEqual(303, response.status_code)
 156         self.assertEqual(expected_data, response.data)
 157
 158     @mock.patch("requests.get")
 159     def test_nsdm_bad_request(self, mock_requests):
 160         dummy_subscription = {
 161             "callbackUri": "http://callbackuri.com",
 162             "authentication": {
 163                 "authType": ["BASIC"],
 164                 "paramsBasic": {
 165                     "userName": "username",
 166                     "password": "password"
 167                 }
 168             },
 169             "filter": {
 170                 "nsdId": "b632bddc-bccd-4180-bd8d-4e8a9578eff7",
 171             }
 172         }
 173         response = self.client.post("/api/nsd/v1/subscriptions",
 174                                     data=dummy_subscription, format='json')
 175         self.assertEqual(400, response.status_code)
 176
 177     @mock.patch("requests.get")
 178     def test_nsdm_invalid_authtype_subscription(self, mock_requests):
 179         dummy_subscription = {
 180             "callbackUri": "http://callbackuri.com",
 181             "authentication": {
 182                 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
 183                 "paramsBasic": {
 184                     "userName": "username",
 185                     "password": "password"
 186                 }
 187             }
 188         }
 189         mock_requests.return_value.status_code = 204
 190         mock_requests.get.return_value.status_code = 204
 191         expected_data = {
 192             'status': 400,
 193             'detail': 'Auth type should be BASIC',
 194         }
 195         response = self.client.post("/api/nsd/v1/subscriptions",
 196                                     data=dummy_subscription, format='json')
 197         self.assertEqual(400, response.status_code)
 198         self.assertEqual(expected_data, response.data)
 199
 200     @mock.patch("requests.get")
 201     def test_nsdm_invalid_authtype_oauthclient_subscription(
 202             self, mock_requests):
 203         dummy_subscription = {
 204             "callbackUri": "http://callbackuri.com",
 205             "authentication": {
 206                 "authType": ["BASIC"],
 207                 "paramsOauth2ClientCredentials": {
 208                     "clientId": "clientId",
 209                     "clientPassword": "password",
 210                     "tokenEndpoint": "http://tokenEndpoint"
 211                 }
 212             }
 213         }
 214         mock_requests.return_value.status_code = 204
 215         mock_requests.get.return_value.status_code = 204
 216         expected_data = {
 217             'status': 400,
 218             'detail': 'Auth type should be OAUTH2_CLIENT_CREDENTIALS',
 219         }
 220         response = self.client.post("/api/nsd/v1/subscriptions",
 221                                     data=dummy_subscription, format='json')
 222         self.assertEqual(400, response.status_code)
 223         self.assertEqual(expected_data, response.data)
 224
 225     @mock.patch("requests.get")
 226     def test_nsdm_invalid_authparams_subscription(self, mock_requests):
 227         dummy_subscription = {
 228             "callbackUri": "http://callbackuri.com",
 229             "authentication": {
 230                 "authType": ["BASIC"],
 231                 "paramsBasic": {
 232                     "userName": "username"
 233                 }
 234             }
 235         }
 236         mock_requests.return_value.status_code = 204
 237         mock_requests.get.return_value.status_code = 204
 238         expected_data = {
 239             'status': 400,
 240             'detail': 'userName and password needed for BASIC',
 241         }
 242         response = self.client.post("/api/nsd/v1/subscriptions",
 243                                     data=dummy_subscription, format='json')
 244         self.assertEqual(400, response.status_code)
 245         self.assertEqual(expected_data, response.data)
 246
 247     @mock.patch("requests.get")
 248     def test_nsdm_invalid_authparams_oauthclient_subscription(
 249             self, mock_requests):
 250         dummy_subscription = {
 251             "callbackUri": "http://callbackuri.com",
 252             "authentication": {
 253                 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
 254                 "paramsOauth2ClientCredentials": {
 255                     "clientPassword": "password",
 256                     "tokenEndpoint": "http://tokenEndpoint"
 257                 }
 258             }
 259         }
 260         mock_requests.return_value.status_code = 204
 261         mock_requests.get.return_value.status_code = 204
 262         expected_data = {
 263             'status': 400,
 264             'detail': 'clientId, clientPassword and tokenEndpoint'
 265                       ' required for OAUTH2_CLIENT_CREDENTIALS',
 266         }
 267         response = self.client.post("/api/nsd/v1/subscriptions",
 268                                     data=dummy_subscription, format='json')
 269         self.assertEqual(400, response.status_code)
 270         self.assertEqual(expected_data, response.data)
 271
 272     @mock.patch("requests.get")
 273     def test_nsdm_invalid_filter_subscription(self, mock_requests):
 274         dummy_subscription = {
 275             "callbackUri": "http://callbackuri.com",
 276             "authentication": {
 277                 "authType": ["BASIC"],
 278                 "paramsBasic": {
 279                     "userName": "username",
 280                     "password": "password"
 281                 }
 282             },
 283             "filter": {
 284                 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
 285                 "nsdInfoId": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
 286             }
 287         }
 288         mock_requests.return_value.status_code = 204
 289         mock_requests.get.return_value.status_code = 204
 290         expected_data = {
 291             'status': 400,
 292             'detail': 'Notification Filter should contain'
 293                       ' either nsdId or nsdInfoId',
 294         }
 295         response = self.client.post("/api/nsd/v1/subscriptions",
 296                                     data=dummy_subscription, format='json')
 297         self.assertEqual(400, response.status_code)
 298         self.assertEqual(expected_data, response.data)
 299
 300     @mock.patch("requests.get")
 301     def test_nsdm_invalid_filter_pnfd_subscription(self, mock_requests):
 302         dummy_subscription = {
 303             "callbackUri": "http://callbackuri.com",
 304             "authentication": {
 305                 "authType": ["BASIC"],
 306                 "paramsBasic": {
 307                     "userName": "username",
 308                     "password": "password"
 309                 }
 310             },
 311             "filter": {
 312                 "pnfdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
 313                 "pnfdInfoIds": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
 314             }
 315         }
 316         mock_requests.return_value.status_code = 204
 317         mock_requests.get.return_value.status_code = 204
 318         expected_data = {
 319             'status': 400,
 320             'detail': 'Notification Filter should contain'
 321                       ' either pnfdId or pnfdInfoIds',
 322         }
 323         response = self.client.post("/api/nsd/v1/subscriptions",
 324                                     data=dummy_subscription, format='json')
 325         self.assertEqual(400, response.status_code)
 326         self.assertEqual(expected_data, response.data)
 327
 328     @mock.patch.object(NsdmSubscription, 'create')
 329     def test_nsdmsubscription_create_when_catch_exception(self, mock_create):
 330         mock_create.side_effect = TypeError("Unicode type")
 331         response = self.client.post('/api/nsd/v1/subscriptions',
 332                                     data=self.subscription, format='json')
 333         self.assertEqual(response.status_code,
 334                          status.HTTP_500_INTERNAL_SERVER_ERROR)
 335
 336     def test_nsdm_get_subscriptions(self):
 337         NsdmSubscriptionModel(subscriptionid=self.subscription_id,
 338                               callback_uri="http://callbackuri.com",
 339                               auth_info={},
 340                               notificationTypes=json.dumps(
 341                                   ["NsdOnBoardingNotification"]),
 342                               nsdId=[], nsdVersion=[],
 343                               nsdInfoId=[], nsdDesigner=[],
 344                               nsdName=[], nsdInvariantId=[],
 345                               vnfPkgIds=[], pnfdInfoIds=[],
 346                               nestedNsdInfoIds=[], nsdOnboardingState=[],
 347                               nsdOperationalState=[], nsdUsageState=[],
 348                               pnfdId=[], pnfdVersion=[], pnfdProvider=[],
 349                               pnfdName=[], pnfdInvariantId=[],
 350                               pnfdOnboardingState=[], pnfdUsageState=[],
 351                               links=json.dumps(self.links)).save()
 352         response = self.client.get("/api/nsd/v1/subscriptions",
 353                                    format='json')
 354         self.assertEqual(status.HTTP_200_OK, response.status_code)
 355         self.assertEqual([self.test_subscription], response.data)
 356
 357     def test_nsdm_get_subscriptions_filter(self):
 358         NsdmSubscriptionModel(subscriptionid=self.subscription_id,
 359                               callback_uri="http://callbackuri.com",
 360                               auth_info={},
 361                               notificationTypes=json.dumps(
 362                                   ["NsdOnBoardingNotification"]),
 363                               nsdId=[], nsdVersion=[],
 364                               nsdInfoId=[], nsdDesigner=[],
 365                               nsdName=[], nsdInvariantId=[],
 366                               vnfPkgIds=[], pnfdInfoIds=[],
 367                               nestedNsdInfoIds=[], nsdOnboardingState=[],
 368                               nsdOperationalState=[], nsdUsageState=[],
 369                               pnfdId=[], pnfdVersion=[], pnfdProvider=[],
 370                               pnfdName=[], pnfdInvariantId=[],
 371                               pnfdOnboardingState=[], pnfdUsageState=[],
 372                               links=json.dumps(self.links)).save()
 373         response = self.client.get("/api/nsd/v1/subscriptions"
 374                                    "?notificationTypes"
 375                                    "=NsdOnBoardingNotification",
 376                                    format='json')
 377         self.assertEqual(status.HTTP_200_OK, response.status_code)
 378         self.assertEqual([self.test_subscription], response.data)
 379
 380     def test_nsdm_get_subscriptions_filter_failure(self):
 381         NsdmSubscriptionModel(subscriptionid=self.subscription_id,
 382                               callback_uri="http://callbackuri.com",
 383                               auth_info={},
 384                               notificationTypes=json.dumps(
 385                                   ["NsdOnBoardingNotification"]),
 386                               nsdId=[], nsdVersion=[],
 387                               nsdInfoId=[], nsdDesigner=[],
 388                               nsdName=[], nsdInvariantId=[],
 389                               vnfPkgIds=[], pnfdInfoIds=[],
 390                               nestedNsdInfoIds=[], nsdOnboardingState=[],
 391                               nsdOperationalState=[], nsdUsageState=[],
 392                               pnfdId=[], pnfdVersion=[], pnfdProvider=[],
 393                               pnfdName=[], pnfdInvariantId=[],
 394                               pnfdOnboardingState=[], pnfdUsageState=[],
 395                               links=json.dumps(self.links)).save()
 396         response = self.client.get("/api/nsd/v1/subscriptions"
 397                                    "?notificationTypes="
 398                                    "PnfdOnBoardingFailureNotification",
 399                                    format='json')
 400         self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
 401
 402     def test_nsdm_get_subscriptions_invalid_filter(self):
 403         NsdmSubscriptionModel(subscriptionid=self.subscription_id,
 404                               callback_uri="http://callbackuri.com",
 405                               auth_info={},
 406                               notificationTypes=json.dumps(
 407                                   ["NsdOnBoardingNotification"]),
 408                               nsdId=[], nsdVersion=[],
 409                               nsdInfoId=[], nsdDesigner=[],
 410                               nsdName=[], nsdInvariantId=[],
 411                               vnfPkgIds=[], pnfdInfoIds=[],
 412                               nestedNsdInfoIds=[], nsdOnboardingState=[],
 413                               nsdOperationalState=[], nsdUsageState=[],
 414                               pnfdId=[], pnfdVersion=[], pnfdProvider=[],
 415                               pnfdName=[], pnfdInvariantId=[],
 416                               pnfdOnboardingState=[], pnfdUsageState=[],
 417                               links=json.dumps(self.links)).save()
 418         response = self.client.get("/api/nsd/v1/subscriptions"
 419                                    "?notificationTypes="
 420                                    "PnfdOnBoardingFailureNotificati",
 421                                    format='json')
 422         self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
 423
 424     @mock.patch.object(NsdmSubscription, 'query_multi_subscriptions')
 425     def test_nsdmsubscription_get_when_catch_exception(self, mock_create):
 426         mock_create.side_effect = TypeError("Unicode type")
 427         response = self.client.get('/api/nsd/v1/subscriptions', format='json')
 428         self.assertEqual(response.status_code,
 429                          status.HTTP_500_INTERNAL_SERVER_ERROR)
 430
 431     def test_nsdm_get_subscription(self):
 432         NsdmSubscriptionModel(subscriptionid=self.subscription_id,
 433                               callback_uri="http://callbackuri.com",
 434                               auth_info={},
 435                               notificationTypes=json.dumps(
 436                                   ["NsdOnBoardingNotification"]),
 437                               nsdId=[], nsdVersion=[],
 438                               nsdInfoId=[], nsdDesigner=[],
 439                               nsdName=[], nsdInvariantId=[],
 440                               vnfPkgIds=[], pnfdInfoIds=[],
 441                               nestedNsdInfoIds=[], nsdOnboardingState=[],
 442                               nsdOperationalState=[], nsdUsageState=[],
 443                               pnfdId=[], pnfdVersion=[], pnfdProvider=[],
 444                               pnfdName=[], pnfdInvariantId=[],
 445                               pnfdOnboardingState=[], pnfdUsageState=[],
 446                               links=json.dumps(self.links)).save()
 447         response = self.client.get('/api/nsd/v1/'
 448                                    'subscriptions/' + self.subscription_id,
 449                                    format='json')
 450         self.assertEqual(status.HTTP_200_OK, response.status_code)
 451         self.assertEqual(self.test_subscription, response.data)
 452
 453     def test_nsdm_get_subscription_failure(self):
 454         expected_data = {
 455             "status": 404,
 456             "detail": "Subscription(" + self.subscription_id + ") "
 457                                                                "doesn't exists"
 458         }
 459         response = self.client.get('/api/nsd/v1/'
 460                                    'subscriptions/' + self.subscription_id,
 461                                    format='json')
 462         self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
 463         self.assertEqual(expected_data, response.data)
 464
 465     def test_nsdm_get_subscription_failure_bad_request(self):
 466         response = self.client.get("/api/nsd/v1/subscriptions/123",
 467                                    format='json')
 468         self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
 469
 470     @mock.patch.object(NsdmSubscription, 'query_single_subscription')
 471     def test_nsdmsubscription_getsingle_when_catch_exception(
 472             self, mock_create):
 473         mock_create.side_effect = TypeError("Unicode type")
 474         response = self.client.get('/api/nsd/v1/'
 475                                    'subscriptions/' + self.subscription_id,
 476                                    format='json')
 477         self.assertEqual(response.status_code,
 478                          status.HTTP_500_INTERNAL_SERVER_ERROR)
 479
 480     def test_ndsm_delete_subscription(self):
 481         NsdmSubscriptionModel(subscriptionid=self.subscription_id,
 482                               callback_uri="http://callbackuri.com",
 483                               auth_info={},
 484                               notificationTypes=json.dumps(
 485                                   ["NsdOnBoardingNotification"]),
 486                               nsdId=[], nsdVersion=[],
 487                               nsdInfoId=[], nsdDesigner=[],
 488                               nsdName=[], nsdInvariantId=[],
 489                               vnfPkgIds=[], pnfdInfoIds=[],
 490                               nestedNsdInfoIds=[], nsdOnboardingState=[],
 491                               nsdOperationalState=[], nsdUsageState=[],
 492                               pnfdId=[], pnfdVersion=[], pnfdProvider=[],
 493                               pnfdName=[], pnfdInvariantId=[],
 494                               pnfdOnboardingState=[], pnfdUsageState=[],
 495                               links=json.dumps(self.links)).save()
 496         response = self.client.delete('/api/nsd/v1/'
 497                                       'subscriptions/' + self.subscription_id,
 498                                       format='json')
 499         self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code)
 500
 501     def test_ndsm_delete_subscription_failure(self):
 502         response = self.client.delete('/api/nsd/v1/'
 503                                       'subscriptions/' + self.subscription_id,
 504                                       format='json')
 505         self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
 506
 507     def test_nsdm_delete_subscription_failure_bad_request(self):
 508         response = self.client.delete("/api/nsd/v1/subscriptions/123",
 509                                       format='json')
 510         self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
 511
 512     @mock.patch.object(NsdmSubscription, 'delete_single_subscription')
 513     def test_nsdmsubscription_delete_when_catch_exception(self, mock_create):
 514         mock_create.side_effect = TypeError("Unicode type")
 515         response = self.client.delete('/api/nsd/v1/'
 516                                       'subscriptions/' + self.subscription_id,
 517                                       format='json')
 518         self.assertEqual(response.status_code,
 519                          status.HTTP_500_INTERNAL_SERVER_ERROR)