X-Git-Url: https://gerrit.onap.org/r/gitweb?a=blobdiff_plain;f=genericparser%2Fpackages%2Ftests%2Ftest_nsdm_subscription.py;fp=genericparser%2Fpackages%2Ftests%2Ftest_nsdm_subscription.py;h=0000000000000000000000000000000000000000;hb=a32c2b20207885d895bd96204cc166fca14db97b;hp=c6c194e521e30a93000f89b17e82bd7fd9c114f7;hpb=431a5a35a8e0a26d21c663167303696db8a7a2a6;p=modeling%2Fetsicatalog.git diff --git a/genericparser/packages/tests/test_nsdm_subscription.py b/genericparser/packages/tests/test_nsdm_subscription.py deleted file mode 100644 index c6c194e..0000000 --- a/genericparser/packages/tests/test_nsdm_subscription.py +++ /dev/null @@ -1,521 +0,0 @@ -# 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)