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             'title': 'Creating Subscription Failed!'
 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             'title': 'Creating Subscription Failed!'
 153         }
 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)
 158
 159     @mock.patch("requests.get")
 160     def test_nsdm_bad_request(self, mock_requests):
 161         dummy_subscription = {
 162             "callbackUri": "http://callbackuri.com",
 163             "authentication": {
 164                 "authType": ["BASIC"],
 165                 "paramsBasic": {
 166                     "userName": "username",
 167                     "password": "password"
 168                 }
 169             },
 170             "filter": {
 171                 "nsdId": "b632bddc-bccd-4180-bd8d-4e8a9578eff7",
 172             }
 173         }
 174         response = self.client.post("/api/nsd/v1/subscriptions",
 175                                     data=dummy_subscription, format='json')
 176         self.assertEqual(400, response.status_code)
 177
 178     @mock.patch("requests.get")
 179     def test_nsdm_invalid_authtype_subscription(self, mock_requests):
 180         dummy_subscription = {
 181             "callbackUri": "http://callbackuri.com",
 182             "authentication": {
 183                 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
 184                 "paramsBasic": {
 185                     "userName": "username",
 186                     "password": "password"
 187                 }
 188             }
 189         }
 190         mock_requests.return_value.status_code = 204
 191         mock_requests.get.return_value.status_code = 204
 192         expected_data = {
 193             'status': 400,
 194             'detail': 'Auth type should be BASIC',
 195             'title': 'Creating Subscription Failed!'
 196         }
 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)
 201
 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",
 207             "authentication": {
 208                 "authType": ["BASIC"],
 209                 "paramsOauth2ClientCredentials": {
 210                     "clientId": "clientId",
 211                     "clientPassword": "password",
 212                     "tokenEndpoint": "http://tokenEndpoint"
 213                 }
 214             }
 215         }
 216         mock_requests.return_value.status_code = 204
 217         mock_requests.get.return_value.status_code = 204
 218         expected_data = {
 219             'status': 400,
 220             'detail': 'Auth type should be OAUTH2_CLIENT_CREDENTIALS',
 221             'title': 'Creating Subscription Failed!'
 222         }
 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)
 227
 228     @mock.patch("requests.get")
 229     def test_nsdm_invalid_authparams_subscription(self, mock_requests):
 230         dummy_subscription = {
 231             "callbackUri": "http://callbackuri.com",
 232             "authentication": {
 233                 "authType": ["BASIC"],
 234                 "paramsBasic": {
 235                     "userName": "username"
 236                 }
 237             }
 238         }
 239         mock_requests.return_value.status_code = 204
 240         mock_requests.get.return_value.status_code = 204
 241         expected_data = {
 242             'status': 400,
 243             'detail': 'userName and password needed for BASIC',
 244             'title': 'Creating Subscription Failed!'
 245         }
 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)
 250
 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",
 256             "authentication": {
 257                 "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
 258                 "paramsOauth2ClientCredentials": {
 259                     "clientPassword": "password",
 260                     "tokenEndpoint": "http://tokenEndpoint"
 261                 }
 262             }
 263         }
 264         mock_requests.return_value.status_code = 204
 265         mock_requests.get.return_value.status_code = 204
 266         expected_data = {
 267             'status': 400,
 268             'detail': 'clientId, clientPassword and tokenEndpoint'
 269                       ' required for OAUTH2_CLIENT_CREDENTIALS',
 270             'title': 'Creating Subscription Failed!'
 271         }
 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)
 276
 277     @mock.patch("requests.get")
 278     def test_nsdm_invalid_filter_subscription(self, mock_requests):
 279         dummy_subscription = {
 280             "callbackUri": "http://callbackuri.com",
 281             "authentication": {
 282                 "authType": ["BASIC"],
 283                 "paramsBasic": {
 284                     "userName": "username",
 285                     "password": "password"
 286                 }
 287             },
 288             "filter": {
 289                 "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
 290                 "nsdInfoId": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
 291             }
 292         }
 293         mock_requests.return_value.status_code = 204
 294         mock_requests.get.return_value.status_code = 204
 295         expected_data = {
 296             'status': 400,
 297             'detail': 'Notification Filter should contain'
 298                       ' either nsdId or nsdInfoId',
 299             'title': 'Creating Subscription Failed!'
 300         }
 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)
 305
 306     @mock.patch("requests.get")
 307     def test_nsdm_invalid_filter_pnfd_subscription(self, mock_requests):
 308         dummy_subscription = {
 309             "callbackUri": "http://callbackuri.com",
 310             "authentication": {
 311                 "authType": ["BASIC"],
 312                 "paramsBasic": {
 313                     "userName": "username",
 314                     "password": "password"
 315                 }
 316             },
 317             "filter": {
 318                 "pnfdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
 319                 "pnfdInfoIds": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
 320             }
 321         }
 322         mock_requests.return_value.status_code = 204
 323         mock_requests.get.return_value.status_code = 204
 324         expected_data = {
 325             'status': 400,
 326             'detail': 'Notification Filter should contain'
 327                       ' either pnfdId or pnfdInfoIds',
 328             'title': 'Creating Subscription Failed!'
 329         }
 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)
 334
 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)
 342
 343     def test_nsdm_get_subscriptions(self):
 344         NsdmSubscriptionModel(subscriptionid=self.subscription_id,
 345                               callback_uri="http://callbackuri.com",
 346                               auth_info={},
 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",
 360                                    format='json')
 361         self.assertEqual(status.HTTP_200_OK, response.status_code)
 362         self.assertEqual([self.test_subscription], response.data)
 363
 364     def test_nsdm_get_subscriptions_filter(self):
 365         NsdmSubscriptionModel(subscriptionid=self.subscription_id,
 366                               callback_uri="http://callbackuri.com",
 367                               auth_info={},
 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"
 381                                    "?notificationTypes"
 382                                    "=NsdOnBoardingNotification",
 383                                    format='json')
 384         self.assertEqual(status.HTTP_200_OK, response.status_code)
 385         self.assertEqual([self.test_subscription], response.data)
 386
 387     def test_nsdm_get_subscriptions_filter_failure(self):
 388         NsdmSubscriptionModel(subscriptionid=self.subscription_id,
 389                               callback_uri="http://callbackuri.com",
 390                               auth_info={},
 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",
 406                                    format='json')
 407         self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
 408
 409     def test_nsdm_get_subscriptions_invalid_filter(self):
 410         NsdmSubscriptionModel(subscriptionid=self.subscription_id,
 411                               callback_uri="http://callbackuri.com",
 412                               auth_info={},
 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",
 428                                    format='json')
 429         self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
 430
 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)
 437
 438     def test_nsdm_get_subscription(self):
 439         NsdmSubscriptionModel(subscriptionid=self.subscription_id,
 440                               callback_uri="http://callbackuri.com",
 441                               auth_info={},
 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,
 456                                    format='json')
 457         self.assertEqual(status.HTTP_200_OK, response.status_code)
 458         self.assertEqual(self.test_subscription, response.data)
 459
 460     def test_nsdm_get_subscription_failure(self):
 461         expected_data = {
 462             "title": "Query Subscription Failed!",
 463             "status": 404,
 464             "detail": "Subscription(" + self.subscription_id + ") "
 465             "doesn't exists"
 466         }
 467         response = self.client.get('/api/nsd/v1/'
 468                                    'subscriptions/' + self.subscription_id,
 469                                    format='json')
 470         self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
 471         self.assertEqual(expected_data, response.data)
 472
 473     def test_nsdm_get_subscription_failure_bad_request(self):
 474         response = self.client.get("/api/nsd/v1/subscriptions/123",
 475                                    format='json')
 476         self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
 477
 478     @mock.patch.object(NsdmSubscription, 'query_single_subscription')
 479     def test_nsdmsubscription_getsingle_when_catch_exception(
 480             self, mock_create):
 481         mock_create.side_effect = TypeError("Unicode type")
 482         response = self.client.get('/api/nsd/v1/'
 483                                    'subscriptions/' + self.subscription_id,
 484                                    format='json')
 485         self.assertEqual(response.status_code,
 486                          status.HTTP_500_INTERNAL_SERVER_ERROR)
 487
 488     def test_ndsm_delete_subscription(self):
 489         NsdmSubscriptionModel(subscriptionid=self.subscription_id,
 490                               callback_uri="http://callbackuri.com",
 491                               auth_info={},
 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,
 506                                       format='json')
 507         self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code)
 508
 509     def test_ndsm_delete_subscription_failure(self):
 510         response = self.client.delete('/api/nsd/v1/'
 511                                       'subscriptions/' + self.subscription_id,
 512                                       format='json')
 513         self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
 514
 515     def test_nsdm_delete_subscription_failure_bad_request(self):
 516         response = self.client.delete("/api/nsd/v1/subscriptions/123",
 517                                       format='json')
 518         self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
 519
 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,
 525                                       format='json')
 526         self.assertEqual(response.status_code,
 527                          status.HTTP_500_INTERNAL_SERVER_ERROR)