lcm/lcm/nf/tests/test_query_subscriptions.py

   1 # Copyright (C) 2018 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
  17 from django.test import TestCase, Client
  18 from rest_framework import status
  19
  20 from lcm.pub.database.models import SubscriptionModel
  21
  22
  23 class TestQuerySubscriptions(TestCase):
  24     def setUp(self):
  25         self.client = Client()
  26         self.subscription_id = "99442b18-a5c7-11e8-998c-bf1755941f16"
  27         self.vnf_instance_id = "cd552c9c-ab6f-11e8-b354-236c32aa91a1"
  28         SubscriptionModel.objects.all().delete()
  29         self.test_single_subscription = {
  30             "id": self.subscription_id,
  31             "callbackUri": "http://aurl.com",
  32             "_links": {
  33                 "self": {
  34                     "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
  35                 }
  36             },
  37             "filter": {
  38                 "notificationTypes": ["VnfLcmOperationOccurrenceNotification"],
  39                 "operationTypes": ["INSTANTIATE"],
  40                 "operationStates": ["STARTING"],
  41                 "vnfInstanceSubscriptionFilter": {
  42                     "vnfdIds": [],
  43                     "vnfInstanceIds": [self.vnf_instance_id],
  44                     "vnfInstanceNames": [],
  45                     "vnfProductsFromProviders": {
  46                         "vnfProvider": "vendor"
  47                     }
  48                 }
  49
  50             }
  51         }
  52
  53     def tearDown(self):
  54         pass
  55
  56     # def test_get_subscriptions(self):
  57     #     vnf_instance_filter = {
  58     #         "vnfdIds": [],
  59     #         "vnfInstanceIds": [self.vnf_instance_id],
  60     #         "vnfInstanceNames": [],
  61     #         "vnfProductsFromProviders": {
  62     #             "vnfProvider": "vendor"
  63     #         }
  64     #     }
  65     #     links = {
  66     #         "self": {
  67     #             "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
  68     #         }
  69     #     }
  70     #     SubscriptionModel(subscription_id=self.subscription_id, callback_uri="http://aurl.com",
  71     #                       auth_info="{}", notification_types="['VnfLcmOperationOccurrenceNotification']",
  72     #                       operation_types="['INSTANTIATE']",
  73     #                       operation_states="['STARTING']",
  74     #                       links=json.dumps(links),
  75     #                       vnf_instance_filter=json.dumps(vnf_instance_filter)).save()
  76     #     response = self.client.get("/api/vnflcm/v1/subscriptions", format='json')
  77     #     self.assertEqual(response.status_code, status.HTTP_200_OK)
  78     #     self.assertEqual([self.test_single_subscription], response.data)
  79
  80     # def test_get_subscriptions_with_vnf_instance_id(self):
  81     #     vnf_instance_filter = {
  82     #         "vnfdIds": [],
  83     #         "vnfInstanceIds": [self.vnf_instance_id],
  84     #         "vnfInstanceNames": [],
  85     #         "vnfProductsFromProviders": {
  86     #             "vnfProvider": "vendor"
  87     #         }
  88     #     }
  89     #     links = {
  90     #         "self": {
  91     #             "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
  92     #         }
  93     #     }
  94     #     SubscriptionModel(subscription_id=self.subscription_id, callback_uri="http://aurl.com",
  95     #                       auth_info="{}", notification_types="['VnfLcmOperationOccurrenceNotification']",
  96     #                       operation_types="['INSTANTIATE']",
  97     #                       operation_states="['STARTING']",
  98     #                       links=json.dumps(links),
  99     #                       vnf_instance_filter=json.dumps(vnf_instance_filter)).save()
 100     #     dummy_vnf_id = "584b35e2-b2a2-11e8-8e11-645106374fd3"
 101     #     dummy_subscription_id = "947dcd2c-b2a2-11e8-b365-645106374fd4"
 102     #     vnf_instance_filter["vnfInstanceIds"].append(dummy_vnf_id)
 103     #     SubscriptionModel(subscription_id=dummy_subscription_id, callback_uri="http://aurl.com",
 104     #                       auth_info="{}", notification_types="['VnfLcmOperationOccurrenceNotification']",
 105     #                       operation_types="['INSTANTIATE']",
 106     #                       operation_states="['STARTING']",
 107     #                       links=json.dumps(links),
 108     #                       vnf_instance_filter=json.dumps(vnf_instance_filter)).save()
 109     #
 110     #     response = self.client.get("/api/vnflcm/v1/subscriptions?vnfInstanceId=" + dummy_vnf_id, format='json')
 111     #     expected_response = self.test_single_subscription.copy()
 112     #     expected_response["id"] = dummy_subscription_id
 113     #     expected_response["filter"]["vnfInstanceSubscriptionFilter"]["vnfInstanceIds"] = \
 114     #         vnf_instance_filter["vnfInstanceIds"]
 115     #     self.assertEqual(response.status_code, status.HTTP_200_OK)
 116     #     self.assertEqual([expected_response], response.data)
 117
 118     def test_get_subscriptions_with_unknown_vnf_instance_id(self):
 119         vnf_instance_filter = {
 120             "vnfdIds": [],
 121             "vnfInstanceIds": [self.vnf_instance_id],
 122             "vnfInstanceNames": [],
 123             "vnfProductsFromProviders": {
 124                 "vnfProvider": "vendor"
 125             }
 126         }
 127         links = {
 128             "self": {
 129                 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
 130             }
 131         }
 132         SubscriptionModel(subscription_id=self.subscription_id, callback_uri="http://aurl.com",
 133                           auth_info="{}", notification_types="['VnfLcmOperationOccurrenceNotification']",
 134                           operation_types="['INSTANTIATE']",
 135                           operation_states="['STARTING']",
 136                           links=json.dumps(links),
 137                           vnf_instance_filter=json.dumps(vnf_instance_filter)).save()
 138         response = self.client.get("/api/vnflcm/v1/subscriptions?vnfInstanceId=dummy", format='json')
 139         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
 140
 141     # def test_get_subscriptions_with_invalid_filter(self):
 142     #     vnf_instance_filter = {
 143     #         "vnfdIds": [],
 144     #         "vnfInstanceIds": [self.vnf_instance_id],
 145     #         "vnfInstanceNames": [],
 146     #         "vnfProductsFromProviders": {
 147     #             "vnfProvider": "vendor"
 148     #         }
 149     #     }
 150     #     links = {
 151     #         "self": {
 152     #             "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
 153     #         }
 154     #     }
 155     #     SubscriptionModel(subscription_id=self.subscription_id, callback_uri="http://aurl.com",
 156     #                       auth_info="{}", notification_types="['VnfLcmOperationOccurrenceNotification']",
 157     #                       operation_types="['INSTANTIATE']",
 158     #                       operation_states="['STARTING']",
 159     #                       links=json.dumps(links),
 160     #                       vnf_instance_filter=json.dumps(vnf_instance_filter)).save()
 161     #     response = self.client.get("/api/vnflcm/v1/subscriptions?dummy=dummy", format='json')
 162     #     self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
 163
 164     # def test_get_subscriptions_with_operation_type_filter(self):
 165     #     vnf_instance_filter = {
 166     #         "vnfdIds": [],
 167     #         "vnfInstanceIds": [self.vnf_instance_id],
 168     #         "vnfInstanceNames": [],
 169     #         "vnfProductsFromProviders": {
 170     #             "vnfProvider": "vendor"
 171     #         }
 172     #     }
 173     #     links = {
 174     #         "self": {
 175     #             "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
 176     #         }
 177     #     }
 178     #     SubscriptionModel(subscription_id=self.subscription_id, callback_uri="http://aurl.com",
 179     #                       auth_info="{}", notification_types="['VnfLcmOperationOccurrenceNotification']",
 180     #                       operation_types="['INSTANTIATE']",
 181     #                       operation_states="['STARTING']",
 182     #                       links=json.dumps(links),
 183     #                       vnf_instance_filter=json.dumps(vnf_instance_filter)).save()
 184     #     dummy_vnf_id = "584b35e2-b2a2-11e8-8e11-645106374fd3"
 185     #     dummy_subscription_id = "947dcd2c-b2a2-11e8-b365-645106374fd4"
 186     #     vnf_instance_filter["vnfInstanceIds"].append(dummy_vnf_id)
 187     #     SubscriptionModel(subscription_id=dummy_subscription_id, callback_uri="http://aurl.com",
 188     #                       auth_info="{}", notification_types="['VnfLcmOperationOccurrenceNotification']",
 189     #                       operation_types="['SCALE']",
 190     #                       operation_states="['STARTING']",
 191     #                       links=json.dumps(links),
 192     #                       vnf_instance_filter=json.dumps(vnf_instance_filter)).save()
 193     #
 194     #     response = self.client.get("/api/vnflcm/v1/subscriptions?operationTypes=SCALE", format='json')
 195     #     expected_response = self.test_single_subscription.copy()
 196     #     expected_response["id"] = dummy_subscription_id
 197     #     expected_response["filter"]["vnfInstanceSubscriptionFilter"]["vnfInstanceIds"] = \
 198     #         vnf_instance_filter["vnfInstanceIds"]
 199     #     expected_response["filter"]["operationTypes"] = ["SCALE"]
 200     #     self.assertEqual(response.status_code, status.HTTP_200_OK)
 201     #     self.assertEqual([expected_response], response.data)