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