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)