+++ /dev/null
-# Copyright (C) 2019 Verizon. All Rights Reserved
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import json
-import mock
-import uuid
-from django.test import TestCase
-from rest_framework.test import APIClient
-from rest_framework import status
-
-from genericparser.packages.biz.nsdm_subscription import NsdmSubscription
-from genericparser.pub.database.models import NsdmSubscriptionModel
-
-
-class TestNsdmSubscription(TestCase):
-
- def setUp(self):
- self.client = APIClient()
- NsdmSubscriptionModel.objects.all().delete()
- self.subscription_id = str(uuid.uuid4())
- self.subscription = {
- "callbackUri": "http://callbackuri.com",
- "authentication": {
- "authType": ["BASIC"],
- "paramsBasic": {
- "userName": "username",
- "password": "password"
- }
- },
- "filter": {
- "nsdId": ["b632bddc-abcd-4180-bd8d-4e8a9578eff7"],
- }
- }
- self.links = {
- "self": {
- "href": "/api/v1/subscriptions/" + self.subscription_id
- }
- }
- self.test_subscription = {
- "callbackUri": "http://callbackuri.com",
- "id": self.subscription_id,
- "filter": {
- "notificationTypes": [
- "NsdOnBoardingNotification"
- ],
- "nsdInfoId": [],
- "nsdId": [],
- "nsdName": [],
- "nsdVersion": [],
- "nsdInvariantId": [],
- "vnfPkgIds": [],
- "nestedNsdInfoIds": [],
- "nsdOnboardingState": [],
- "nsdOperationalState": [],
- "nsdUsageState": [],
- "pnfdInfoIds": [],
- "pnfdId": [],
- "pnfdName": [],
- "pnfdVersion": [],
- "pnfdProvider": [],
- "pnfdInvariantId": [],
- "pnfdOnboardingState": [],
- "pnfdUsageState": []
- },
- "_links": self.links,
- }
-
- def tearDown(self):
- pass
-
- @mock.patch("requests.get")
- @mock.patch.object(uuid, 'uuid4')
- def test_nsdm_subscribe_notification(self, mock_uuid4, mock_requests):
- temp_uuid = str(uuid.uuid4())
- mock_requests.return_value.status_code = 204
- mock_requests.get.return_value.status_code = 204
- mock_uuid4.return_value = temp_uuid
- response = self.client.post("/api/nsd/v1/subscriptions",
- data=self.subscription, format='json')
- self.assertEqual(201, response.status_code)
- self.assertEqual(self.subscription["callbackUri"],
- response.data["callbackUri"])
- self.assertEqual(temp_uuid, response.data["id"])
-
- @mock.patch("requests.get")
- @mock.patch.object(uuid, 'uuid4')
- def test_nsdm_subscribe_callbackFailure(self, mock_uuid4, mock_requests):
- temp_uuid = str(uuid.uuid4())
- mock_requests.return_value.status_code = 500
- mock_requests.get.return_value.status_code = 500
- mock_uuid4.return_value = temp_uuid
- expected_data = {
- 'status': 500,
- 'detail': "callbackUri http://callbackuri.com didn't"
- " return 204 statuscode."
- }
- response = self.client.post("/api/nsd/v1/subscriptions",
- data=self.subscription, format='json')
- self.assertEqual(500, response.status_code)
- self.assertEqual(expected_data, response.data)
-
- @mock.patch("requests.get")
- def test_nsdm_second_subscription(self, mock_requests):
- mock_requests.return_value.status_code = 204
- mock_requests.get.return_value.status_code = 204
- response = self.client.post("/api/nsd/v1/subscriptions",
- data=self.subscription, format='json')
- self.assertEqual(201, response.status_code)
- self.assertEqual(self.subscription["callbackUri"],
- response.data["callbackUri"])
- dummy_subscription = {
- "callbackUri": "http://callbackuri.com",
- "authentication": {
- "authType": ["BASIC"],
- "paramsBasic": {
- "userName": "username",
- "password": "password"
- }
- },
- "filter": {
- "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
- }
- }
- response = self.client.post("/api/nsd/v1/subscriptions",
- data=dummy_subscription, format='json')
- self.assertEqual(201, response.status_code)
- self.assertEqual(dummy_subscription["callbackUri"],
- response.data["callbackUri"])
-
- @mock.patch("requests.get")
- def test_nsdm_duplicate_subscription(self, mock_requests):
- mock_requests.return_value.status_code = 204
- mock_requests.get.return_value.status_code = 204
- response = self.client.post("/api/nsd/v1/subscriptions",
- data=self.subscription, format='json')
- self.assertEqual(201, response.status_code)
- self.assertEqual(self.subscription["callbackUri"],
- response.data["callbackUri"])
- expected_data = {
- 'status': 303,
- 'detail': 'Already Subscription exists with'
- ' the same callbackUri and filter',
- }
- response = self.client.post("/api/nsd/v1/subscriptions",
- data=self.subscription, format='json')
- self.assertEqual(303, response.status_code)
- self.assertEqual(expected_data, response.data)
-
- @mock.patch("requests.get")
- def test_nsdm_bad_request(self, mock_requests):
- dummy_subscription = {
- "callbackUri": "http://callbackuri.com",
- "authentication": {
- "authType": ["BASIC"],
- "paramsBasic": {
- "userName": "username",
- "password": "password"
- }
- },
- "filter": {
- "nsdId": "b632bddc-bccd-4180-bd8d-4e8a9578eff7",
- }
- }
- response = self.client.post("/api/nsd/v1/subscriptions",
- data=dummy_subscription, format='json')
- self.assertEqual(400, response.status_code)
-
- @mock.patch("requests.get")
- def test_nsdm_invalid_authtype_subscription(self, mock_requests):
- dummy_subscription = {
- "callbackUri": "http://callbackuri.com",
- "authentication": {
- "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
- "paramsBasic": {
- "userName": "username",
- "password": "password"
- }
- }
- }
- mock_requests.return_value.status_code = 204
- mock_requests.get.return_value.status_code = 204
- expected_data = {
- 'status': 400,
- 'detail': 'Auth type should be BASIC',
- }
- response = self.client.post("/api/nsd/v1/subscriptions",
- data=dummy_subscription, format='json')
- self.assertEqual(400, response.status_code)
- self.assertEqual(expected_data, response.data)
-
- @mock.patch("requests.get")
- def test_nsdm_invalid_authtype_oauthclient_subscription(
- self, mock_requests):
- dummy_subscription = {
- "callbackUri": "http://callbackuri.com",
- "authentication": {
- "authType": ["BASIC"],
- "paramsOauth2ClientCredentials": {
- "clientId": "clientId",
- "clientPassword": "password",
- "tokenEndpoint": "http://tokenEndpoint"
- }
- }
- }
- mock_requests.return_value.status_code = 204
- mock_requests.get.return_value.status_code = 204
- expected_data = {
- 'status': 400,
- 'detail': 'Auth type should be OAUTH2_CLIENT_CREDENTIALS',
- }
- response = self.client.post("/api/nsd/v1/subscriptions",
- data=dummy_subscription, format='json')
- self.assertEqual(400, response.status_code)
- self.assertEqual(expected_data, response.data)
-
- @mock.patch("requests.get")
- def test_nsdm_invalid_authparams_subscription(self, mock_requests):
- dummy_subscription = {
- "callbackUri": "http://callbackuri.com",
- "authentication": {
- "authType": ["BASIC"],
- "paramsBasic": {
- "userName": "username"
- }
- }
- }
- mock_requests.return_value.status_code = 204
- mock_requests.get.return_value.status_code = 204
- expected_data = {
- 'status': 400,
- 'detail': 'userName and password needed for BASIC',
- }
- response = self.client.post("/api/nsd/v1/subscriptions",
- data=dummy_subscription, format='json')
- self.assertEqual(400, response.status_code)
- self.assertEqual(expected_data, response.data)
-
- @mock.patch("requests.get")
- def test_nsdm_invalid_authparams_oauthclient_subscription(
- self, mock_requests):
- dummy_subscription = {
- "callbackUri": "http://callbackuri.com",
- "authentication": {
- "authType": ["OAUTH2_CLIENT_CREDENTIALS"],
- "paramsOauth2ClientCredentials": {
- "clientPassword": "password",
- "tokenEndpoint": "http://tokenEndpoint"
- }
- }
- }
- mock_requests.return_value.status_code = 204
- mock_requests.get.return_value.status_code = 204
- expected_data = {
- 'status': 400,
- 'detail': 'clientId, clientPassword and tokenEndpoint'
- ' required for OAUTH2_CLIENT_CREDENTIALS',
- }
- response = self.client.post("/api/nsd/v1/subscriptions",
- data=dummy_subscription, format='json')
- self.assertEqual(400, response.status_code)
- self.assertEqual(expected_data, response.data)
-
- @mock.patch("requests.get")
- def test_nsdm_invalid_filter_subscription(self, mock_requests):
- dummy_subscription = {
- "callbackUri": "http://callbackuri.com",
- "authentication": {
- "authType": ["BASIC"],
- "paramsBasic": {
- "userName": "username",
- "password": "password"
- }
- },
- "filter": {
- "nsdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
- "nsdInfoId": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
- }
- }
- mock_requests.return_value.status_code = 204
- mock_requests.get.return_value.status_code = 204
- expected_data = {
- 'status': 400,
- 'detail': 'Notification Filter should contain'
- ' either nsdId or nsdInfoId',
- }
- response = self.client.post("/api/nsd/v1/subscriptions",
- data=dummy_subscription, format='json')
- self.assertEqual(400, response.status_code)
- self.assertEqual(expected_data, response.data)
-
- @mock.patch("requests.get")
- def test_nsdm_invalid_filter_pnfd_subscription(self, mock_requests):
- dummy_subscription = {
- "callbackUri": "http://callbackuri.com",
- "authentication": {
- "authType": ["BASIC"],
- "paramsBasic": {
- "userName": "username",
- "password": "password"
- }
- },
- "filter": {
- "pnfdId": ["b632bddc-bccd-4180-bd8d-4e8a9578eff7"],
- "pnfdInfoIds": ["d0ea5ec3-0b98-438a-9bea-488230cff174"]
- }
- }
- mock_requests.return_value.status_code = 204
- mock_requests.get.return_value.status_code = 204
- expected_data = {
- 'status': 400,
- 'detail': 'Notification Filter should contain'
- ' either pnfdId or pnfdInfoIds',
- }
- response = self.client.post("/api/nsd/v1/subscriptions",
- data=dummy_subscription, format='json')
- self.assertEqual(400, response.status_code)
- self.assertEqual(expected_data, response.data)
-
- @mock.patch.object(NsdmSubscription, 'create')
- def test_nsdmsubscription_create_when_catch_exception(self, mock_create):
- mock_create.side_effect = TypeError("Unicode type")
- response = self.client.post('/api/nsd/v1/subscriptions',
- data=self.subscription, format='json')
- self.assertEqual(response.status_code,
- status.HTTP_500_INTERNAL_SERVER_ERROR)
-
- def test_nsdm_get_subscriptions(self):
- NsdmSubscriptionModel(subscriptionid=self.subscription_id,
- callback_uri="http://callbackuri.com",
- auth_info={},
- notificationTypes=json.dumps(
- ["NsdOnBoardingNotification"]),
- nsdId=[], nsdVersion=[],
- nsdInfoId=[], nsdDesigner=[],
- nsdName=[], nsdInvariantId=[],
- vnfPkgIds=[], pnfdInfoIds=[],
- nestedNsdInfoIds=[], nsdOnboardingState=[],
- nsdOperationalState=[], nsdUsageState=[],
- pnfdId=[], pnfdVersion=[], pnfdProvider=[],
- pnfdName=[], pnfdInvariantId=[],
- pnfdOnboardingState=[], pnfdUsageState=[],
- links=json.dumps(self.links)).save()
- response = self.client.get("/api/nsd/v1/subscriptions",
- format='json')
- self.assertEqual(status.HTTP_200_OK, response.status_code)
- self.assertEqual([self.test_subscription], response.data)
-
- def test_nsdm_get_subscriptions_filter(self):
- NsdmSubscriptionModel(subscriptionid=self.subscription_id,
- callback_uri="http://callbackuri.com",
- auth_info={},
- notificationTypes=json.dumps(
- ["NsdOnBoardingNotification"]),
- nsdId=[], nsdVersion=[],
- nsdInfoId=[], nsdDesigner=[],
- nsdName=[], nsdInvariantId=[],
- vnfPkgIds=[], pnfdInfoIds=[],
- nestedNsdInfoIds=[], nsdOnboardingState=[],
- nsdOperationalState=[], nsdUsageState=[],
- pnfdId=[], pnfdVersion=[], pnfdProvider=[],
- pnfdName=[], pnfdInvariantId=[],
- pnfdOnboardingState=[], pnfdUsageState=[],
- links=json.dumps(self.links)).save()
- response = self.client.get("/api/nsd/v1/subscriptions"
- "?notificationTypes"
- "=NsdOnBoardingNotification",
- format='json')
- self.assertEqual(status.HTTP_200_OK, response.status_code)
- self.assertEqual([self.test_subscription], response.data)
-
- def test_nsdm_get_subscriptions_filter_failure(self):
- NsdmSubscriptionModel(subscriptionid=self.subscription_id,
- callback_uri="http://callbackuri.com",
- auth_info={},
- notificationTypes=json.dumps(
- ["NsdOnBoardingNotification"]),
- nsdId=[], nsdVersion=[],
- nsdInfoId=[], nsdDesigner=[],
- nsdName=[], nsdInvariantId=[],
- vnfPkgIds=[], pnfdInfoIds=[],
- nestedNsdInfoIds=[], nsdOnboardingState=[],
- nsdOperationalState=[], nsdUsageState=[],
- pnfdId=[], pnfdVersion=[], pnfdProvider=[],
- pnfdName=[], pnfdInvariantId=[],
- pnfdOnboardingState=[], pnfdUsageState=[],
- links=json.dumps(self.links)).save()
- response = self.client.get("/api/nsd/v1/subscriptions"
- "?notificationTypes="
- "PnfdOnBoardingFailureNotification",
- format='json')
- self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
-
- def test_nsdm_get_subscriptions_invalid_filter(self):
- NsdmSubscriptionModel(subscriptionid=self.subscription_id,
- callback_uri="http://callbackuri.com",
- auth_info={},
- notificationTypes=json.dumps(
- ["NsdOnBoardingNotification"]),
- nsdId=[], nsdVersion=[],
- nsdInfoId=[], nsdDesigner=[],
- nsdName=[], nsdInvariantId=[],
- vnfPkgIds=[], pnfdInfoIds=[],
- nestedNsdInfoIds=[], nsdOnboardingState=[],
- nsdOperationalState=[], nsdUsageState=[],
- pnfdId=[], pnfdVersion=[], pnfdProvider=[],
- pnfdName=[], pnfdInvariantId=[],
- pnfdOnboardingState=[], pnfdUsageState=[],
- links=json.dumps(self.links)).save()
- response = self.client.get("/api/nsd/v1/subscriptions"
- "?notificationTypes="
- "PnfdOnBoardingFailureNotificati",
- format='json')
- self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
-
- @mock.patch.object(NsdmSubscription, 'query_multi_subscriptions')
- def test_nsdmsubscription_get_when_catch_exception(self, mock_create):
- mock_create.side_effect = TypeError("Unicode type")
- response = self.client.get('/api/nsd/v1/subscriptions', format='json')
- self.assertEqual(response.status_code,
- status.HTTP_500_INTERNAL_SERVER_ERROR)
-
- def test_nsdm_get_subscription(self):
- NsdmSubscriptionModel(subscriptionid=self.subscription_id,
- callback_uri="http://callbackuri.com",
- auth_info={},
- notificationTypes=json.dumps(
- ["NsdOnBoardingNotification"]),
- nsdId=[], nsdVersion=[],
- nsdInfoId=[], nsdDesigner=[],
- nsdName=[], nsdInvariantId=[],
- vnfPkgIds=[], pnfdInfoIds=[],
- nestedNsdInfoIds=[], nsdOnboardingState=[],
- nsdOperationalState=[], nsdUsageState=[],
- pnfdId=[], pnfdVersion=[], pnfdProvider=[],
- pnfdName=[], pnfdInvariantId=[],
- pnfdOnboardingState=[], pnfdUsageState=[],
- links=json.dumps(self.links)).save()
- response = self.client.get('/api/nsd/v1/'
- 'subscriptions/' + self.subscription_id,
- format='json')
- self.assertEqual(status.HTTP_200_OK, response.status_code)
- self.assertEqual(self.test_subscription, response.data)
-
- def test_nsdm_get_subscription_failure(self):
- expected_data = {
- "status": 404,
- "detail": "Subscription(" + self.subscription_id + ") "
- "doesn't exists"
- }
- response = self.client.get('/api/nsd/v1/'
- 'subscriptions/' + self.subscription_id,
- format='json')
- self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
- self.assertEqual(expected_data, response.data)
-
- def test_nsdm_get_subscription_failure_bad_request(self):
- response = self.client.get("/api/nsd/v1/subscriptions/123",
- format='json')
- self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
-
- @mock.patch.object(NsdmSubscription, 'query_single_subscription')
- def test_nsdmsubscription_getsingle_when_catch_exception(
- self, mock_create):
- mock_create.side_effect = TypeError("Unicode type")
- response = self.client.get('/api/nsd/v1/'
- 'subscriptions/' + self.subscription_id,
- format='json')
- self.assertEqual(response.status_code,
- status.HTTP_500_INTERNAL_SERVER_ERROR)
-
- def test_ndsm_delete_subscription(self):
- NsdmSubscriptionModel(subscriptionid=self.subscription_id,
- callback_uri="http://callbackuri.com",
- auth_info={},
- notificationTypes=json.dumps(
- ["NsdOnBoardingNotification"]),
- nsdId=[], nsdVersion=[],
- nsdInfoId=[], nsdDesigner=[],
- nsdName=[], nsdInvariantId=[],
- vnfPkgIds=[], pnfdInfoIds=[],
- nestedNsdInfoIds=[], nsdOnboardingState=[],
- nsdOperationalState=[], nsdUsageState=[],
- pnfdId=[], pnfdVersion=[], pnfdProvider=[],
- pnfdName=[], pnfdInvariantId=[],
- pnfdOnboardingState=[], pnfdUsageState=[],
- links=json.dumps(self.links)).save()
- response = self.client.delete('/api/nsd/v1/'
- 'subscriptions/' + self.subscription_id,
- format='json')
- self.assertEqual(status.HTTP_204_NO_CONTENT, response.status_code)
-
- def test_ndsm_delete_subscription_failure(self):
- response = self.client.delete('/api/nsd/v1/'
- 'subscriptions/' + self.subscription_id,
- format='json')
- self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
-
- def test_nsdm_delete_subscription_failure_bad_request(self):
- response = self.client.delete("/api/nsd/v1/subscriptions/123",
- format='json')
- self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
-
- @mock.patch.object(NsdmSubscription, 'delete_single_subscription')
- def test_nsdmsubscription_delete_when_catch_exception(self, mock_create):
- mock_create.side_effect = TypeError("Unicode type")
- response = self.client.delete('/api/nsd/v1/'
- 'subscriptions/' + self.subscription_id,
- format='json')
- self.assertEqual(response.status_code,
- status.HTTP_500_INTERNAL_SERVER_ERROR)
Code Review / modeling / etsicatalog.git / blobdiff