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