lcm/ns/tests/test_query_subscriptions.py

   1 # Copyright (c) 2019, CMCC Technologies Co., Ltd.
   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.ns_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": ["NsLcmOperationOccurrenceNotification"],
  39                 "operationTypes": ["INSTANTIATE"],
  40                 "operationStates": ["STARTING"],
  41                 # "nsComponentTypes": ["NS"],
  42                 "nsInstanceSubscriptionFilter": {
  43                     "nsdIds": [],
  44                     "nsInstanceIds": [self.ns_instance_id],
  45                     "nsInstanceNames": []
  46                 }
  47             }
  48         }
  49
  50     def tearDown(self):
  51         pass
  52
  53     def test_get_subscriptions(self):
  54         ns_instance_filter = {
  55             "nsdIds": [],
  56             "nsInstanceIds": [self.ns_instance_id],
  57             "nsInstanceNames": []
  58         }
  59         links = {
  60             "self": {
  61                 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
  62             }
  63         }
  64         SubscriptionModel(subscription_id=self.subscription_id, callback_uri="http://aurl.com",
  65                           auth_info="{}", notification_types="['NsLcmOperationOccurrenceNotification']",
  66                           operation_types="['INSTANTIATE']",
  67                           operation_states="['STARTING']",
  68                           # ns_component_types="['NS']",
  69                           links=json.dumps(links),
  70                           ns_instance_filter=json.dumps(ns_instance_filter)).save()
  71         response = self.client.get("/api/nslcm/v1/subscriptions", format='json')
  72         self.assertEqual(response.status_code, status.HTTP_200_OK)
  73         self.assertEqual([self.test_single_subscription], response.data)
  74
  75     def test_get_subscriptions_with_ns_instance_id(self):
  76         ns_instance_filter = {
  77             "nsdIds": [],
  78             "nsInstanceIds": [self.ns_instance_id],
  79             "nsInstanceNames": []
  80         }
  81         links = {
  82             "self": {
  83                 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
  84             }
  85         }
  86         SubscriptionModel(subscription_id=self.subscription_id, callback_uri="http://aurl.com",
  87                           auth_info="{}", notification_types="['NsLcmOperationOccurrenceNotification']",
  88                           operation_types="['INSTANTIATE']",
  89                           operation_states="['STARTING']",
  90                           links=json.dumps(links),
  91                           ns_instance_filter=json.dumps(ns_instance_filter)).save()
  92         dummy_ns_id = "584b35e2-b2a2-11e8-8e11-645106374fd3"
  93         dummy_subscription_id = "947dcd2c-b2a2-11e8-b365-645106374fd4"
  94         ns_instance_filter["nsInstanceIds"].append(dummy_ns_id)
  95         SubscriptionModel(subscription_id=dummy_subscription_id, callback_uri="http://aurl.com",
  96                           auth_info="{}", notification_types="['NsLcmOperationOccurrenceNotification']",
  97                           operation_types="['INSTANTIATE']",
  98                           operation_states="['STARTING']",
  99                           links=json.dumps(links),
 100                           ns_instance_filter=json.dumps(ns_instance_filter)).save()
 101
 102         response = self.client.get("/api/nslcm/v1/subscriptions?nsInstanceId=" + dummy_ns_id, format='json')
 103         expected_response = self.test_single_subscription.copy()
 104         expected_response["id"] = dummy_subscription_id
 105         expected_response["filter"]["nsInstanceSubscriptionFilter"]["nsInstanceIds"] = \
 106             ns_instance_filter["nsInstanceIds"]
 107         self.assertEqual(response.status_code, status.HTTP_200_OK)
 108         self.assertEqual([expected_response], response.data)
 109
 110     def test_get_subscriptions_with_unknown_ns_instance_id(self):
 111         ns_instance_filter = {
 112             "nsdIds": [],
 113             "nsInstanceIds": [self.ns_instance_id],
 114             "nsInstanceNames": []
 115         }
 116         links = {
 117             "self": {
 118                 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
 119             }
 120         }
 121         SubscriptionModel(subscription_id=self.subscription_id, callback_uri="http://aurl.com",
 122                           auth_info="{}", notification_types="['NsLcmOperationOccurrenceNotification']",
 123                           operation_types="['INSTANTIATE']",
 124                           operation_states="['STARTING']",
 125                           links=json.dumps(links),
 126                           ns_instance_filter=json.dumps(ns_instance_filter)).save()
 127         response = self.client.get("/api/nslcm/v1/subscriptions?nsInstanceId=dummy", format='json')
 128         self.assertEqual(response.status_code, status.HTTP_200_OK)
 129         self.assertEqual([], response.data)
 130
 131     def test_get_subscriptions_with_invalid_filter(self):
 132         ns_instance_filter = {
 133             "nsdIds": [],
 134             "nsInstanceIds": [self.ns_instance_id],
 135             "nsInstanceNames": []
 136         }
 137         links = {
 138             "self": {
 139                 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
 140             }
 141         }
 142         SubscriptionModel(subscription_id=self.subscription_id, callback_uri="http://aurl.com",
 143                           auth_info="{}", notification_types="['NsLcmOperationOccurrenceNotification']",
 144                           operation_types="['INSTANTIATE']",
 145                           operation_states="['STARTING']",
 146                           links=json.dumps(links),
 147                           ns_instance_filter=json.dumps(ns_instance_filter)).save()
 148         response = self.client.get("/api/nslcm/v1/subscriptions?dummy=dummy", format='json')
 149         self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
 150
 151     def test_get_subscriptions_with_operation_type_filter(self):
 152         ns_instance_filter = {
 153             "nsdIds": [],
 154             "nsInstanceIds": [self.ns_instance_id],
 155             "nsInstanceNames": []
 156         }
 157         links = {
 158             "self": {
 159                 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
 160             }
 161         }
 162         SubscriptionModel(subscription_id=self.subscription_id, callback_uri="http://aurl.com",
 163                           auth_info="{}", notification_types="['NsLcmOperationOccurrenceNotification']",
 164                           operation_types="['INSTANTIATE']",
 165                           operation_states="['STARTING']",
 166                           links=json.dumps(links),
 167                           ns_instance_filter=json.dumps(ns_instance_filter)).save()
 168         dummy_ns_id = "584b35e2-b2a2-11e8-8e11-645106374fd3"
 169         dummy_subscription_id = "947dcd2c-b2a2-11e8-b365-645106374fd4"
 170         ns_instance_filter["nsInstanceIds"].append(dummy_ns_id)
 171         SubscriptionModel(subscription_id=dummy_subscription_id, callback_uri="http://aurl.com",
 172                           auth_info="{}", notification_types="['NsLcmOperationOccurrenceNotification']",
 173                           operation_types="['SCALE']",
 174                           operation_states="['STARTING']",
 175                           links=json.dumps(links),
 176                           ns_instance_filter=json.dumps(ns_instance_filter)).save()
 177
 178         response = self.client.get("/api/nslcm/v1/subscriptions?operationTypes=SCALE", format='json')
 179         expected_response = self.test_single_subscription.copy()
 180         expected_response["id"] = dummy_subscription_id
 181         expected_response["filter"]["nsInstanceSubscriptionFilter"]["nsInstanceIds"] = \
 182             ns_instance_filter["nsInstanceIds"]
 183         expected_response["filter"]["operationTypes"] = ["SCALE"]
 184         self.assertEqual(response.status_code, status.HTTP_200_OK)
 185         self.assertEqual([expected_response], response.data)