1 # Copyright (c) 2019, CMCC Technologies Co., Ltd.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
17 from django.test import TestCase, Client
18 from rest_framework import status
20 from lcm.pub.database.models import SubscriptionModel
23 class TestQuerySubscriptions(TestCase):
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",
34 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
38 "notificationTypes": ["NsLcmOperationOccurrenceNotification"],
39 "operationTypes": ["INSTANTIATE"],
40 "operationStates": ["STARTING"],
41 # "nsComponentTypes": ["NS"],
42 "nsInstanceSubscriptionFilter": {
44 "nsInstanceIds": [self.ns_instance_id],
53 def test_get_subscriptions(self):
54 ns_instance_filter = {
56 "nsInstanceIds": [self.ns_instance_id],
61 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
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)
75 def test_get_subscriptions_with_ns_instance_id(self):
76 ns_instance_filter = {
78 "nsInstanceIds": [self.ns_instance_id],
83 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
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()
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)
110 def test_get_subscriptions_with_unknown_ns_instance_id(self):
111 ns_instance_filter = {
113 "nsInstanceIds": [self.ns_instance_id],
114 "nsInstanceNames": []
118 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
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)
131 def test_get_subscriptions_with_invalid_filter(self):
132 ns_instance_filter = {
134 "nsInstanceIds": [self.ns_instance_id],
135 "nsInstanceNames": []
139 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
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)
151 def test_get_subscriptions_with_operation_type_filter(self):
152 ns_instance_filter = {
154 "nsInstanceIds": [self.ns_instance_id],
155 "nsInstanceNames": []
159 "href": "/api/v1/subscriptions/99442b18-a5c7-11e8-998c-bf1755941f16"
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()
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)