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(
  71             subscription_id=self.subscription_id,
  72             callback_uri="http://aurl.com",
  73             auth_info="{}",
  74             notification_types="['VnfLcmOperationOccurrenceNotification']",
  75             operation_types="['INSTANTIATE']",
  76             operation_states="['STARTING']",
  77             links=json.dumps(links),
  78             vnf_instance_filter=json.dumps(vnf_instance_filter)
  79         ).save()
  80         response = self.client.get(
  81             "/api/vnflcm/v1/subscriptions",
  82             format='json'
  83         )
  84         self.assertEqual(response.status_code, status.HTTP_200_OK)
  85         self.assertEqual([self.test_single_subscription], response.data)
  86
  87     def test_get_subscriptions_with_vnf_instance_id(self):
  88         vnf_instance_filter = {
  89             "vnfdIds": [],
  90             "vnfInstanceIds": [self.vnf_instance_id],
  91             "vnfInstanceNames": [],
  92             "vnfProductsFromProviders": {
  93                 "vnfProvider": "vendor"
  94             }
  95         }
  96         links = {
  97             "self": {
  98                 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
  99             }
 100         }
 101         SubscriptionModel(
 102             subscription_id=self.subscription_id,
 103             callback_uri="http://aurl.com",
 104             auth_info="{}",
 105             notification_types="['VnfLcmOperationOccurrenceNotification']",
 106             operation_types="['INSTANTIATE']",
 107             operation_states="['STARTING']",
 108             links=json.dumps(links),
 109             vnf_instance_filter=json.dumps(vnf_instance_filter)
 110         ).save()
 111         dummy_vnf_id = "584b35e2-b2a2-11e8-8e11-645106374fd3"
 112         dummy_subscription_id = "947dcd2c-b2a2-11e8-b365-645106374fd4"
 113         vnf_instance_filter["vnfInstanceIds"].append(dummy_vnf_id)
 114         SubscriptionModel(
 115             subscription_id=dummy_subscription_id,
 116             callback_uri="http://aurl.com",
 117             auth_info="{}",
 118             notification_types="['VnfLcmOperationOccurrenceNotification']",
 119             operation_types="['INSTANTIATE']",
 120             operation_states="['STARTING']",
 121             links=json.dumps(links),
 122             vnf_instance_filter=json.dumps(vnf_instance_filter)
 123         ).save()
 124
 125         response = self.client.get(
 126             "/api/vnflcm/v1/subscriptions?vnfInstanceId=" + dummy_vnf_id,
 127             format='json'
 128         )
 129         expected_response = self.test_single_subscription.copy()
 130         expected_response["id"] = dummy_subscription_id
 131         expected_response["filter"]["vnfInstanceSubscriptionFilter"]["vnfInstanceIds"] = \
 132             vnf_instance_filter["vnfInstanceIds"]
 133         self.assertEqual(response.status_code, status.HTTP_200_OK)
 134         self.assertEqual([expected_response], response.data)
 135
 136     def test_get_subscriptions_with_unknown_vnf_instance_id(self):
 137         vnf_instance_filter = {
 138             "vnfdIds": [],
 139             "vnfInstanceIds": [self.vnf_instance_id],
 140             "vnfInstanceNames": [],
 141             "vnfProductsFromProviders": {
 142                 "vnfProvider": "vendor"
 143             }
 144         }
 145         links = {
 146             "self": {
 147                 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
 148             }
 149         }
 150         SubscriptionModel(
 151             subscription_id=self.subscription_id,
 152             callback_uri="http://aurl.com",
 153             auth_info="{}",
 154             notification_types="['VnfLcmOperationOccurrenceNotification']",
 155             operation_types="['INSTANTIATE']",
 156             operation_states="['STARTING']",
 157             links=json.dumps(links),
 158             vnf_instance_filter=json.dumps(vnf_instance_filter)
 159         ).save()
 160         response = self.client.get(
 161             "/api/vnflcm/v1/subscriptions?vnfInstanceId=dummy",
 162             format='json'
 163         )
 164         self.assertEqual(
 165             response.status_code,
 166             status.HTTP_500_INTERNAL_SERVER_ERROR
 167         )
 168
 169     def test_get_subscriptions_with_invalid_filter(self):
 170         vnf_instance_filter = {
 171             "vnfdIds": [],
 172             "vnfInstanceIds": [self.vnf_instance_id],
 173             "vnfInstanceNames": [],
 174             "vnfProductsFromProviders": {
 175                 "vnfProvider": "vendor"
 176             }
 177         }
 178         links = {
 179             "self": {
 180                 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
 181             }
 182         }
 183         SubscriptionModel(
 184             subscription_id=self.subscription_id,
 185             callback_uri="http://aurl.com",
 186             auth_info="{}",
 187             notification_types="['VnfLcmOperationOccurrenceNotification']",
 188             operation_types="['INSTANTIATE']",
 189             operation_states="['STARTING']",
 190             links=json.dumps(links),
 191             vnf_instance_filter=json.dumps(vnf_instance_filter)
 192         ).save()
 193         response = self.client.get(
 194             "/api/vnflcm/v1/subscriptions?dummy=dummy",
 195             format='json'
 196         )
 197         self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
 198
 199     def test_get_subscriptions_with_operation_type_filter(self):
 200         vnf_instance_filter = {
 201             "vnfdIds": [],
 202             "vnfInstanceIds": [self.vnf_instance_id],
 203             "vnfInstanceNames": [],
 204             "vnfProductsFromProviders": {
 205                 "vnfProvider": "vendor"
 206             }
 207         }
 208         links = {
 209             "self": {
 210                 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
 211             }
 212         }
 213         SubscriptionModel(
 214             subscription_id=self.subscription_id,
 215             callback_uri="http://aurl.com",
 216             auth_info="{}",
 217             notification_types="['VnfLcmOperationOccurrenceNotification']",
 218             operation_types="['INSTANTIATE']",
 219             operation_states="['STARTING']",
 220             links=json.dumps(links),
 221             vnf_instance_filter=json.dumps(vnf_instance_filter)
 222         ).save()
 223         dummy_vnf_id = "584b35e2-b2a2-11e8-8e11-645106374fd3"
 224         dummy_subscription_id = "947dcd2c-b2a2-11e8-b365-645106374fd4"
 225         vnf_instance_filter["vnfInstanceIds"].append(dummy_vnf_id)
 226         SubscriptionModel(
 227             subscription_id=dummy_subscription_id,
 228             callback_uri="http://aurl.com",
 229             auth_info="{}",
 230             notification_types="['VnfLcmOperationOccurrenceNotification']",
 231             operation_types="['SCALE']",
 232             operation_states="['STARTING']",
 233             links=json.dumps(links),
 234             vnf_instance_filter=json.dumps(vnf_instance_filter)
 235         ).save()
 236
 237         response = self.client.get(
 238             "/api/vnflcm/v1/subscriptions?operationTypes=SCALE",
 239             format='json'
 240         )
 241         expected_response = self.test_single_subscription.copy()
 242         expected_response["id"] = dummy_subscription_id
 243         expected_response["filter"]["vnfInstanceSubscriptionFilter"]["vnfInstanceIds"] = \
 244             vnf_instance_filter["vnfInstanceIds"]
 245         expected_response["filter"]["operationTypes"] = ["SCALE"]
 246         self.assertEqual(response.status_code, status.HTTP_200_OK)
 247         self.assertEqual([expected_response], response.data)