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